From e8fadfc9983a6dafb2fb2b60a7a11e16b8ea7d35 Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 30 Jul 2026 14:57:25 +0400 Subject: [PATCH] Complete autonomous recovery controls --- AUDIT.md | 42 ++++++---- cmd/orchestra-worker/main.go | 80 ++++++++++++++++--- cmd/orchestra-worker/main_test.go | 12 +++ cmd/orchestra/main.go | 53 +++++++++++- deploy/worker-projects.example.json | 6 +- internal/domain/domain.go | 16 +++- internal/domain/fuzz_test.go | 2 +- internal/federation/client.go | 14 ++++ internal/federation/client_test.go | 25 ++++++ internal/integration/federation_lease_test.go | 4 + internal/operations/operations.go | 11 +++ internal/orchestrator/orchestrator.go | 14 ++-- internal/registry/registry.go | 12 +++ internal/registry/registry_test.go | 10 +++ internal/router/router.go | 59 +++++--------- internal/router/router_test.go | 8 +- internal/store/store.go | 41 +++++++++- internal/store/store_test.go | 32 ++++++++ 18 files changed, 364 insertions(+), 77 deletions(-) diff --git a/AUDIT.md b/AUDIT.md index c5a4b8a..25323b8 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -36,23 +36,31 @@ released agent, or reject a valid completion. dropping the live session. `TaskBlocked` remains terminal for an explicit operator block. `TestNeedsAttentionRetainsFencedLeaseForLateCompletion` covers the durable recovery path. -- **Retries:** expiry bypasses `Router.HandleEvent`; attempts/backoff are - in-memory and unsynchronised. Project durable `attempt`, `next_retry_at`, - and failure class; route every reclaim through one transition. -- **Launch:** repeated start failures hold a lease for up to 30 minutes. - Workers must ACK start or NACK with typed evidence; retry transient failures, - block invalid handoffs, and immediately free unusable capacity. -- **Completion:** `.orchestra/done` is the only worker completion signal. - Combine an explicit completion intent with native idle/exit identity, the - worker-owned quality gate, verified commit, and verified push. -- **Quota:** worker sessions do not retain a usage source, so live receipts - are zero and quota routing is ineffective. Record per-lease deltas and - publish both 5-hour and weekly projections; unknown quota fails closed. -- **Approvals:** the continuity probe required six manual grants. Add audited - per-project policy for safe worktree-local reads, edits, tests, and Git; - keep destructive, secret, network, and out-of-worktree actions gated. -- **Observability:** replace release/rotation `continue` paths with durable - phase, last error, retry time, lease epoch, pane state, and anchor fields. +- **Retries:** **Closed 2026-07-30.** Hand-off-less `TaskReleased` is the + single durable reclaim transition. It projects exponential `attempt`, + `next_retry_at`, and `failure_class`; router assignment reads those fields, + so coordinator restarts cannot reset a backoff or retry limit. + `TestReclaimPersistsAttemptAndBackoffAcrossReopen` covers replay. +- **Launch:** **Closed 2026-07-30.** Workers emit a fenced + `TaskLaunchAcknowledged` only after a local start/prompt is persisted. + Typed NACKs immediately reclaim transient unusable capacity, terminally + block invalid handoffs, and retain uncertain live panes for reconciliation. +- **Completion:** **Closed 2026-07-30.** `.orchestra/done` is explicit + intent only; the worker also requires native non-busy identity, runs its + quality gate, verifies immutable `TASK.md`, commits, pushes, and checks + the remote SHA before it emits completion. +- **Quota:** **Closed 2026-07-30.** Completion receipts contain native + per-lease deltas plus a known/unknown marker. Five-hour and weekly + projections are published from the same receipts; any bounded harness + without fresh known usage fails routing closed. +- **Approvals:** **Closed 2026-07-30.** Projects have a validated audited + `safe_operations` policy limited to worktree-local read/edit/test/Git. + Workers inject it into the task prompt; network, secrets, destructive + actions, and paths outside the worktree remain operator-gated. +- **Observability:** **Closed 2026-07-30.** Task projections now retain + lifecycle phase, last error, retry time/failure class, lease epoch, pane + state, and anchor. Release/anchor certification faults enter durable + `needs_attention` instead of disappearing through retry `continue` paths. ## P2 — performance diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index 3908b4c..60f44ad 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -87,6 +87,7 @@ type lease struct { PickupAcknowledged bool `json:"pickup_acknowledged,omitempty"` Version int `json:"version"` Until time.Time `json:"until"` + UsageBaseline float64 `json:"usage_baseline,omitempty"` } type releaseTransaction struct { ID string `json:"id"` @@ -99,10 +100,11 @@ type releaseTransaction struct { UpdatedAt time.Time `json:"updated_at"` } type projectConfig struct { - Repo string `json:"repo"` - Root string `json:"worktree_root"` - Remote string `json:"remote"` - QualityGate string `json:"quality_gate,omitempty"` + Repo string `json:"repo"` + Root string `json:"worktree_root"` + Remote string `json:"remote"` + QualityGate string `json:"quality_gate,omitempty"` + SafeOperations []string `json:"safe_operations,omitempty"` } type completionEvidence struct { TaskID string `json:"task_id"` @@ -323,6 +325,9 @@ func (w *worker) start(ctx context.Context, t domain.Task, ref string) error { } s.TaskFileSHA = taskHash(t) prompt := "Read TASK.md at the worktree root and execute it." + if len(p.SafeOperations) > 0 { + prompt += " This project's audited no-grant policy permits only worktree-local " + strings.Join(p.SafeOperations, ", ") + ". Network, secrets, destructive actions, and paths outside this worktree still require an explicit operator approval." + } if ref != "" { prompt += " A validated handoff exists; inspect local Git history and the recorded checkpoint before continuing." } @@ -335,12 +340,35 @@ func (w *worker) start(ctx context.Context, t domain.Task, ref string) error { if err := w.herdr.Prompt(ctx, s.PaneID, prompt, 0); err != nil { return err } + if l, ok := w.leases[t.ID]; ok { + if err := w.api.Start(ctx, t.ID, l.Epoch, l.Version, w.sessionEvidence(ctx, t.ID, s)); err != nil { + return fmt.Errorf("ack start: %w", err) + } + l.Version++ + w.leases[t.ID] = l + if err := w.save(); err != nil { + return err + } + } if ref != "" { return w.ackPickup(ctx, t.ID, s) } return nil } +func classifyLaunchError(err error, sessionStarted bool) string { + if sessionStarted { + // A prompt response can be lost after herdr accepted it. Never reclaim + // that pane just because its acknowledgement was uncertain. + return "launch_uncertain" + } + text := strings.ToLower(err.Error()) + if strings.Contains(text, "handoff") || strings.Contains(text, "pickup") || strings.Contains(text, "task.md") { + return "invalid_handoff" + } + return "launch_transient" +} + func taskHash(t domain.Task) string { b := continuity.RenderTaskFile(t); return domain.Hash(b) } func (w *worker) releaseReady(ctx context.Context) { @@ -355,6 +383,17 @@ func (w *worker) releaseReady(ctx context.Context) { } } if _, err := os.Stat(filepath.Join(s.Worktree, ".orchestra", "done")); err == nil { + // A done marker is an intent, not enough on its own: do not race a + // still-running native harness into committing half-written work. + status, statusErr := (herdr.CLIAdapter{Client: w.herdr, Harness: w.harness}).AgentStatus(ctx, s) + if statusErr != nil { + w.recordError(fmt.Errorf("completion identity %s: %w", id, statusErr)) + continue + } + if herdr.IsBusy(status) { + w.recordError(fmt.Errorf("completion %s deferred: agent status %s", id, status)) + continue + } evidence, err := w.finalize(ctx, id, s) if err != nil { w.recordError(fmt.Errorf("complete %s: %w", id, err)) @@ -368,7 +407,7 @@ func (w *worker) releaseReady(ctx context.Context) { log.Printf("upload completion %s: %v", id, err) continue } - if err = w.api.Complete(ctx, id, ref, evidence.ResultSHA, evidence.Branch, evidence.Remote, w.leases[id].Epoch, w.leases[id].Version, w.usageReceipt(s), w.sessionEvidence(ctx, id, s)); err != nil { + if err = w.api.Complete(ctx, id, ref, evidence.ResultSHA, evidence.Branch, evidence.Remote, w.leases[id].Epoch, w.leases[id].Version, w.usageReceipt(s, w.leases[id]), w.sessionEvidence(ctx, id, s)); err != nil { w.recordError(fmt.Errorf("complete %s: %w", id, err)) log.Printf("complete %s: %v", id, err) continue @@ -576,9 +615,9 @@ func (w *worker) sessionEvidence(ctx context.Context, taskID string, s herdr.Ses return e } -func (w *worker) usageReceipt(s herdr.Session) map[string]any { +func (w *worker) usageReceipt(s herdr.Session, l lease) map[string]any { if s.SessionFile == "" && !(w.harness == "opencode" && s.SessionID != "") { - return map[string]any{"harness_id": w.harnessID, "consumed": 0} + return map[string]any{"harness_id": w.harnessID, "consumed": 0, "known": false, "error": "native usage identity missing"} } var usage herdr.Usage var err error @@ -591,9 +630,13 @@ func (w *worker) usageReceipt(s herdr.Session) map[string]any { usage, err = herdr.OpenCodeSessionUsage(s.SessionID) } if err != nil { - return map[string]any{"harness_id": w.harnessID, "consumed": 0, "error": err.Error()} + return map[string]any{"harness_id": w.harnessID, "consumed": 0, "known": false, "error": err.Error()} } - return map[string]any{"harness_id": w.harnessID, "input_tokens": usage.Input, "cache_read_tokens": usage.CacheRead, "cache_write_tokens": usage.CacheWrite, "output_tokens": usage.Output, "consumed": usage.Numerator()} + delta := float64(usage.Numerator()) - l.UsageBaseline + if delta < 0 { + delta = 0 + } + return map[string]any{"harness_id": w.harnessID, "input_tokens": usage.Input, "cache_read_tokens": usage.CacheRead, "cache_write_tokens": usage.CacheWrite, "consumed": delta, "lease_usage_delta": delta, "known": true} } func git(ctx context.Context, dir string, args ...string) ([]byte, error) { @@ -814,6 +857,12 @@ func (w *worker) once(ctx context.Context) error { w.leases[e.TaskID] = l } } + if e.Type == "TaskLaunchAcknowledged" { + if l, ok := w.leases[e.TaskID]; ok { + l.Version = e.Version + w.leases[e.TaskID] = l + } + } if e.Type == "TaskNeedsAttention" { // The diagnostic event increments the aggregate version but leaves // ownership intact. Keep our locally persisted expected version in @@ -914,6 +963,19 @@ func (w *worker) once(ctx context.Context) error { if t, ok := w.tasks[taskID]; ok { if err := w.start(ctx, t, l.HandoffRef); err != nil { log.Printf("lease %s: %v", t.ID, err) + _, started := w.sessions[taskID] + class := classifyLaunchError(err, started) + var evidence domain.SessionEvidence + if session, ok := w.sessions[taskID]; ok { + evidence = w.sessionEvidence(ctx, taskID, session) + } + if nackErr := w.api.NackStart(ctx, taskID, l.Epoch, l.Version, class, err.Error(), evidence); nackErr != nil { + w.recordError(fmt.Errorf("nack launch %s: %w", taskID, nackErr)) + continue + } + if class != "launch_uncertain" { + delete(w.leases, taskID) + } } } } diff --git a/cmd/orchestra-worker/main_test.go b/cmd/orchestra-worker/main_test.go index 144b20a..323d198 100644 --- a/cmd/orchestra-worker/main_test.go +++ b/cmd/orchestra-worker/main_test.go @@ -51,6 +51,18 @@ func TestWorkerRefusesCorruptDurableState(t *testing.T) { } } +func TestClassifyLaunchErrorPreservesUncertainLivePane(t *testing.T) { + if got := classifyLaunchError(errors.New("prompt response lost"), true); got != "launch_uncertain" { + t.Fatalf("live pane class=%q", got) + } + if got := classifyLaunchError(errors.New("pickup anchor mismatch"), false); got != "invalid_handoff" { + t.Fatalf("bad handoff class=%q", got) + } + if got := classifyLaunchError(errors.New("temporary herdr outage"), false); got != "launch_transient" { + t.Fatalf("transient class=%q", got) + } +} + func TestInitialReplayDoesNotResurrectReleasedLease(t *testing.T) { s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index ebda641..843bc81 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -1060,7 +1060,7 @@ func main() { w.WriteHeader(http.StatusNoContent) }) mux.HandleFunc("/v1/federation/workers/", func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost || (!strings.HasSuffix(r.URL.Path, "/heartbeat") && !strings.HasSuffix(r.URL.Path, "/renew") && !strings.HasSuffix(r.URL.Path, "/handoff") && !strings.HasSuffix(r.URL.Path, "/pickup") && !strings.HasSuffix(r.URL.Path, "/complete") && !strings.HasSuffix(r.URL.Path, "/captures")) { + if r.Method != http.MethodPost || (!strings.HasSuffix(r.URL.Path, "/heartbeat") && !strings.HasSuffix(r.URL.Path, "/renew") && !strings.HasSuffix(r.URL.Path, "/start") && !strings.HasSuffix(r.URL.Path, "/nack") && !strings.HasSuffix(r.URL.Path, "/handoff") && !strings.HasSuffix(r.URL.Path, "/pickup") && !strings.HasSuffix(r.URL.Path, "/complete") && !strings.HasSuffix(r.URL.Path, "/captures")) { http.Error(w, "not found", 404) return } @@ -1122,6 +1122,8 @@ func main() { Branch string `json:"branch"` Remote string `json:"remote"` Receipt map[string]any `json:"receipt"` + FailureClass string `json:"failure_class"` + LastError string `json:"last_error"` SessionEvidence domain.SessionEvidence `json:"session_evidence"` } if json.NewDecoder(r.Body).Decode(&b) != nil || b.TaskID == "" { @@ -1149,6 +1151,55 @@ func main() { http.Error(w, "lease version conflict", http.StatusConflict) return } + if strings.HasSuffix(r.URL.Path, "/start") { + // A lost response after append is an idempotent start ACK, not a + // reason to strand the running pane behind a stale version. + if t.LifecyclePhase == "started" && t.Lease != nil && t.Lease.HarnessID == parts[3] && t.Lease.Epoch == b.LeaseEpoch { + w.WriteHeader(http.StatusNoContent) + return + } + if b.ExpectedVersion != t.Version { + http.Error(w, "lease version conflict", http.StatusConflict) + return + } + p, _ := json.Marshal(map[string]any{"harness_id": parts[3], "lease_epoch": b.LeaseEpoch, "expected_version": t.Version, "lifecycle_phase": "started", "session_evidence": b.SessionEvidence}) + e := domain.Event{ID: id(), Type: "TaskLaunchAcknowledged", TaskID: b.TaskID, Version: t.Version + 1, Payload: p, Surface: string(authz.System)} + if err := s.Append(e); err != nil { + http.Error(w, err.Error(), http.StatusConflict) + return + } + json.NewEncoder(w).Encode(e) + return + } + if strings.HasSuffix(r.URL.Path, "/nack") { + if b.ExpectedVersion != t.Version || b.FailureClass == "" || b.LastError == "" { + http.Error(w, "current lease version, failure_class, and last_error required", http.StatusConflict) + return + } + var typ string + var p []byte + switch b.FailureClass { + case "invalid_handoff": + typ = "TaskBlocked" + p, _ = json.Marshal(map[string]any{"blocker": b.LastError, "block_reason": string(domain.BlockReasonHandoffValidation), "harness_id": parts[3], "lease_epoch": b.LeaseEpoch, "expected_version": t.Version, "lifecycle_phase": "launch_nacked", "last_error": b.LastError, "session_evidence": b.SessionEvidence}) + case "launch_uncertain": + typ = "TaskNeedsAttention" + p, _ = json.Marshal(map[string]any{"blocker": b.LastError, "block_reason": string(domain.BlockReasonLeaseFailure), "harness_id": parts[3], "lease_epoch": b.LeaseEpoch, "expected_version": t.Version, "lifecycle_phase": "launch_uncertain", "last_error": b.LastError, "session_evidence": b.SessionEvidence}) + default: + typ = "TaskReleased" + p, _ = json.Marshal(map[string]any{"reason": "launch_failed", "failure_class": b.FailureClass, "harness_id": parts[3], "lease_epoch": b.LeaseEpoch, "expected_version": t.Version, "lifecycle_phase": "launch_nacked", "last_error": b.LastError, "session_evidence": b.SessionEvidence}) + } + e := domain.Event{ID: id(), Type: typ, TaskID: b.TaskID, Version: t.Version + 1, Payload: p, Surface: string(authz.System)} + if err := s.Append(e); err != nil { + http.Error(w, err.Error(), http.StatusConflict) + return + } + if typ == "TaskReleased" && rt != nil { + _, _ = rt.HandleEvent(e) + } + json.NewEncoder(w).Encode(e) + return + } if strings.HasSuffix(r.URL.Path, "/renew") { ttl := b.TTLSeconds if ttl == 0 { diff --git a/deploy/worker-projects.example.json b/deploy/worker-projects.example.json index 41094ab..1de53f1 100644 --- a/deploy/worker-projects.example.json +++ b/deploy/worker-projects.example.json @@ -3,12 +3,14 @@ "repo": "/srv/orchestra/repos/test-e2e", "worktree_root": "/srv/orchestra/worktrees/test-e2e", "remote": "origin", - "quality_gate": "go test ./..." + "quality_gate": "go test ./...", + "safe_operations": ["read", "edit", "test", "git"] }, "correx": { "repo": "/srv/orchestra/repos/correx", "worktree_root": "/srv/orchestra/worktrees/correx", "remote": "origin", - "quality_gate": "go test ./... && go vet ./..." + "quality_gate": "go test ./... && go vet ./...", + "safe_operations": ["read", "edit", "test", "git"] } } diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 6805a3f..aa2861d 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -151,6 +151,14 @@ type Task struct { LastHarness string `json:"last_harness_id,omitempty"` PaneState string `json:"pane_state,omitempty"` // open, closed, unreachable, unknown LastSession SessionEvidence `json:"last_session,omitempty"` + // Recovery state is part of the durable projection, never process-local + // router memory. This makes retry and operator diagnostics survive a + // coordinator restart. + Attempt int `json:"attempt,omitempty"` + NextRetryAt time.Time `json:"next_retry_at,omitempty"` + FailureClass string `json:"failure_class,omitempty"` + LifecyclePhase string `json:"lifecycle_phase,omitempty"` + LastError string `json:"last_error,omitempty"` } type Event struct { @@ -190,7 +198,7 @@ func ValidateEvent(e Event) error { if e.SchemaVersion >= 2 && strings.TrimSpace(e.Surface) == "" { return fmt.Errorf("%w: surface required", ErrInvalid) } - allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskLeaseRenewed": true, "TaskReleased": true, "TaskPickupValidated": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "TaskNeedsAttention": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true, "TaskCorrected": true, "QuotaReported": true, "StandupAdvisory": true} + allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskLeaseRenewed": true, "TaskReleased": true, "TaskLaunchAcknowledged": true, "TaskPickupValidated": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "TaskNeedsAttention": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true, "TaskCorrected": true, "QuotaReported": true, "StandupAdvisory": true} if !allowed[e.Type] { return fmt.Errorf("%w: unknown type %q", ErrInvalid, e.Type) } @@ -206,7 +214,7 @@ func ValidateEvent(e Event) error { } if e.SchemaVersion >= 3 { switch e.Type { - case "TaskLeased", "TaskLeaseRenewed", "TaskPickupValidated": + case "TaskLeased", "TaskLeaseRenewed", "TaskLaunchAcknowledged", "TaskPickupValidated": if v, ok := p["lease_epoch"].(string); !ok || strings.TrimSpace(v) == "" { return fmt.Errorf("%w: lease_epoch required", ErrInvalid) } @@ -313,6 +321,10 @@ func ValidatePayload(typ string, p map[string]any) error { if err := requiredString("reason"); err != nil { return err } + case "TaskLaunchAcknowledged": + if err := requiredString("harness_id"); err != nil { + return err + } case "TaskBlocked", "TaskNeedsAttention": if err := requiredString("blocker"); err != nil { return err diff --git a/internal/domain/fuzz_test.go b/internal/domain/fuzz_test.go index 7829935..9a5c5e9 100644 --- a/internal/domain/fuzz_test.go +++ b/internal/domain/fuzz_test.go @@ -10,7 +10,7 @@ import ( // of these on adversarial input is a defect regardless of whether real // producers happen to send well-formed payloads. var eventTypesUnderTest = []string{ - "TaskCreated", "TaskLeased", "TaskReleased", "TaskCompleted", "TaskFailed", + "TaskCreated", "TaskLeased", "TaskReleased", "TaskLaunchAcknowledged", "TaskCompleted", "TaskFailed", "TaskBlocked", "TaskNeedsAttention", "ApprovalRequested", "ApprovalGranted", "ApprovalDenied", "TaskAmended", "QuotaReported", "StandupAdvisory", } diff --git a/internal/federation/client.go b/internal/federation/client.go index e0e9fe6..90c491c 100644 --- a/internal/federation/client.go +++ b/internal/federation/client.go @@ -138,6 +138,20 @@ func (c Client) Renew(ctx context.Context, taskID, epoch string, expectedVersion } return err } +func (c Client) Start(ctx context.Context, taskID, epoch string, expectedVersion int, evidence domain.SessionEvidence) error { + resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/start", map[string]any{"task_id": taskID, "lease_epoch": epoch, "expected_version": expectedVersion, "session_evidence": evidence}) + if resp != nil { + resp.Body.Close() + } + return err +} +func (c Client) NackStart(ctx context.Context, taskID, epoch string, expectedVersion int, failureClass, detail string, evidence domain.SessionEvidence) error { + resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/nack", map[string]any{"task_id": taskID, "lease_epoch": epoch, "expected_version": expectedVersion, "failure_class": failureClass, "last_error": detail, "session_evidence": evidence}) + if resp != nil { + resp.Body.Close() + } + return err +} func (c Client) Artifact(ctx context.Context, ref string) ([]byte, error) { resp, err := c.request(ctx, http.MethodGet, "/v1/artifacts/"+url.PathEscape(ref), nil) if err != nil { diff --git a/internal/federation/client_test.go b/internal/federation/client_test.go index 5d92970..5b85a0d 100644 --- a/internal/federation/client_test.go +++ b/internal/federation/client_test.go @@ -2,8 +2,10 @@ package federation import ( "context" + "encoding/json" "net/http" "net/http/httptest" + "orchestra/internal/domain" "testing" ) @@ -49,3 +51,26 @@ func TestClientRegistersPollsAndReadsArtifactAsWorker(t *testing.T) { } } } + +func TestClientReportsTypedLaunchAckAndNack(t *testing.T) { + seen := map[string]map[string]any{} + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatal(err) + } + seen[r.URL.Path] = body + w.WriteHeader(http.StatusNoContent) + })) + defer s.Close() + c := Client{BaseURL: s.URL, WorkerID: "h1", Token: "worker"} + if err := c.Start(context.Background(), "task", "epoch", 7, domain.SessionEvidence{PaneID: "p"}); err != nil { + t.Fatal(err) + } + if err := c.NackStart(context.Background(), "task", "epoch", 8, "invalid_handoff", "anchor mismatch", domain.SessionEvidence{PaneID: "p"}); err != nil { + t.Fatal(err) + } + if seen["/v1/federation/workers/h1/start"]["expected_version"] != float64(7) || seen["/v1/federation/workers/h1/nack"]["failure_class"] != "invalid_handoff" { + t.Fatalf("launch reports=%#v", seen) + } +} diff --git a/internal/integration/federation_lease_test.go b/internal/integration/federation_lease_test.go index 7b5a184..abcea7b 100644 --- a/internal/integration/federation_lease_test.go +++ b/internal/integration/federation_lease_test.go @@ -180,6 +180,10 @@ func TestCrossMachineLeaseAnchorAndQuotaArePerHost(t *testing.T) { } } quotaReport("homesrv-h1", 95) + // A zero native delta is still a receipt: it proves workpc's independent + // usage source is available, rather than treating missing quota data as + // harmless headroom. + quotaReport("workpc-h1", 0) limits := map[string]router.QuotaWindowLimits{ "homesrv-h1": {Weekly: 100}, "workpc-h1": {Weekly: 100}, diff --git a/internal/operations/operations.go b/internal/operations/operations.go index fa4fa58..544f9eb 100644 --- a/internal/operations/operations.go +++ b/internal/operations/operations.go @@ -61,6 +61,17 @@ type QuotaReceipt struct { At time.Time `json:"at"` } +// QuotaWindows publishes the two routing windows from the same additive +// per-lease receipts used by the scheduler. +type QuotaWindows struct { + FiveHour map[string]float64 `json:"five_hour"` + Weekly map[string]float64 `json:"weekly"` +} + +func ProjectQuotaWindows(events []domain.Event, now time.Time) QuotaWindows { + return QuotaWindows{FiveHour: AggregateQuota(events, now.Add(-5*time.Hour), now), Weekly: AggregateQuota(events, now.Add(-7*24*time.Hour), now)} +} + // AggregateQuota sums native receipts in the requested window. It does not // de-duplicate rotations or use a cumulative session total. func AggregateQuota(events []domain.Event, from, to time.Time) map[string]float64 { diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 52d8ff5..615b7a8 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -752,17 +752,21 @@ func (c *Coordinator) rotate(ctx context.Context, hard float64) { } ref, err := a.Release(ctx, session) if err != nil { + // A release failure is operational state, not a silent retry. Keep + // the fenced lease and pane for recovery while durably exposing the + // failed phase to the worker and operator. + _ = c.block(task, "rotation release: "+err.Error()) continue } if ref == "" { + _ = c.block(task, "rotation release: empty handoff reference") continue } anchorSHA, err := herdr.HeadSHA(session.Worktree) if err != nil { - // Cannot certify the anchor: do not release with an invalid - // TaskReleased payload (it would fail validation and strand - // the lease/session). Leave the lease intact for the next - // tick or TTL expiry to reclaim. + // Cannot certify the anchor. Record the fault while retaining the + // owner; an unseen bare continue used to leave this state opaque. + _ = c.block(task, "rotation anchor: "+err.Error()) continue } b, _ := json.Marshal(map[string]any{"handoff_ref": ref, "reason": reason, "anchor_sha": anchorSHA, "harness_id": task.Lease.HarnessID, "lease_epoch": task.Lease.Epoch, "expected_version": task.Version}) @@ -1020,7 +1024,7 @@ func (c *Coordinator) rememberSession(taskID string, s herdr.Session) error { } func (c *Coordinator) block(t domain.Task, reason string) error { - p := map[string]any{"blocker": reason, "block_reason": string(domain.InferBlockReason(reason)), "pane_state": "unknown", "session_evidence": domain.SessionEvidence{PaneState: "unknown", Source: "coordinator", CheckedAt: time.Now().UTC()}} + p := map[string]any{"blocker": reason, "block_reason": string(domain.InferBlockReason(reason)), "lifecycle_phase": "needs_attention", "last_error": reason, "pane_state": "unknown", "session_evidence": domain.SessionEvidence{PaneState: "unknown", Source: "coordinator", CheckedAt: time.Now().UTC()}} if s, ok := c.Session(t.ID); ok { p["pane_id"] = s.PaneID p["harness_id"] = s.HerdrID diff --git a/internal/registry/registry.go b/internal/registry/registry.go index 5349c09..8c36134 100644 --- a/internal/registry/registry.go +++ b/internal/registry/registry.go @@ -29,6 +29,11 @@ type Project struct { Repo string `json:"repo,omitempty"` WorktreeRoot string `json:"worktree_root,omitempty"` QualityGate string `json:"quality_gate,omitempty"` + // SafeOperations is an audited, deliberately small allow-list for work + // inside this project's task worktree. It documents what workers may + // perform without an operator grant; network, secrets, destructive Git, + // and paths outside the worktree are never represented here. + SafeOperations []string `json:"safe_operations,omitempty"` } type Machine struct { ID string `json:"id"` @@ -131,6 +136,13 @@ func New(c Config) (Registry, error) { if len(p.MachineAffinity) == 0 { return Registry{}, fmt.Errorf("project %q: %w", p.ID, ErrNoAffinity) } + for _, op := range p.SafeOperations { + switch op { + case "read", "edit", "test", "git": + default: + return Registry{}, fmt.Errorf("project %q: unsafe operation %q is not policy-configurable", p.ID, op) + } + } r.projects[p.ID] = p } for _, m := range c.Machines { diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go index f5611b9..62be0b6 100644 --- a/internal/registry/registry_test.go +++ b/internal/registry/registry_test.go @@ -34,3 +34,13 @@ func TestNewRejectsBrokenReferences(t *testing.T) { t.Fatalf("err=%v", err) } } + +func TestProjectSafeOperationsAreNarrowAndAudited(t *testing.T) { + _, err := New(Config{Projects: []Project{{ID: "p", MachineAffinity: []string{"m"}, SafeOperations: []string{"read", "network"}}}, Machines: []Machine{{ID: "m", Address: "m:1"}}}) + if err == nil { + t.Fatal("network operation was accepted into no-grant policy") + } + if _, err := New(Config{Projects: []Project{{ID: "p", MachineAffinity: []string{"m"}, SafeOperations: []string{"read", "edit", "test", "git"}}}, Machines: []Machine{{ID: "m", Address: "m:1"}}}); err != nil { + t.Fatalf("safe policy rejected: %v", err) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 8054fc5..9c95f89 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -53,8 +53,9 @@ type QuotaAvailability struct { Now func() time.Time } -func (q QuotaAvailability) sumSince(harnessID string, since time.Time) float64 { +func (q QuotaAvailability) sumSince(harnessID string, since time.Time) (float64, bool) { var consumed float64 + known := false for _, e := range q.Store.Events(0) { if e.Type != "QuotaReported" || e.At.Before(since) { continue @@ -62,12 +63,17 @@ func (q QuotaAvailability) sumSince(harnessID string, since time.Time) float64 { var p struct { HarnessID string `json:"harness_id"` Consumed float64 `json:"consumed"` + Known *bool `json:"known"` } if json.Unmarshal(e.Payload, &p) == nil && p.HarnessID == harnessID && p.Consumed >= 0 { + if p.Known != nil && !*p.Known { + return 0, false + } + known = true consumed += p.Consumed } } - return consumed + return consumed, known } func (q QuotaAvailability) Available(h registry.Herdr) bool { @@ -82,11 +88,17 @@ func (q QuotaAvailability) Available(h registry.Herdr) bool { if q.Now != nil { now = q.Now() } - if limits.FiveHour > 0 && q.sumSince(h.ID, now.Add(-fiveHourWindow)) >= limits.FiveHour*quotaConservativeFraction { - return false + if limits.FiveHour > 0 { + used, known := q.sumSince(h.ID, now.Add(-fiveHourWindow)) + if !known || used >= limits.FiveHour*quotaConservativeFraction { + return false + } } - if limits.Weekly > 0 && q.sumSince(h.ID, now.Add(-weeklyWindow)) >= limits.Weekly*quotaConservativeFraction { - return false + if limits.Weekly > 0 { + used, known := q.sumSince(h.ID, now.Add(-weeklyWindow)) + if !known || used >= limits.Weekly*quotaConservativeFraction { + return false + } } return true } @@ -103,8 +115,6 @@ type Router struct { Timeout time.Duration Retry RetryPolicy Now func() time.Time - backoff map[string]time.Time - attempts map[string]int OnLease func(domain.Event) error } @@ -115,12 +125,6 @@ func (r *Router) init() { if r.Now == nil { r.Now = time.Now } - if r.backoff == nil { - r.backoff = map[string]time.Time{} - } - if r.attempts == nil { - r.attempts = map[string]int{} - } } // HandleEvent evaluates the sink after creation and after a lease is freed. @@ -129,23 +133,6 @@ func (r *Router) HandleEvent(e domain.Event) ([]domain.Event, error) { if e.Type != "TaskCreated" && e.Type != "TaskReleased" { return nil, nil } - if e.Type == "TaskReleased" { - // Rotation *is* TaskReleased (spec §5.3: "rotation = intra-task - // lease transfer") — a task healthy enough to rotate repeatedly - // must not be killed by the retry limit meant for genuine failures - // (expiry, crash). Only a release without a valid handoff_ref - // (expiry/crash) counts against MaxAttempts. - var p struct { - HandoffRef string `json:"handoff_ref"` - } - isRotation := json.Unmarshal(e.Payload, &p) == nil && p.HandoffRef != "" - if !isRotation { - r.attempts[e.TaskID]++ - if r.Retry.Backoff > 0 { - r.backoff[e.TaskID] = r.Now().Add(r.Retry.Backoff) - } - } - } return r.AssignPending() } @@ -156,14 +143,14 @@ func (r *Router) AssignPending() ([]domain.Event, error) { } var queued []domain.Task for _, t := range r.Store.Tasks() { - if t.State == domain.StateQueued && !r.Now().Before(r.backoff[t.ID]) { + if t.State == domain.StateQueued && (t.NextRetryAt.IsZero() || !r.Now().Before(t.NextRetryAt)) { queued = append(queued, t) } } sort.SliceStable(queued, func(i, j int) bool { return importance(queued[i], r.Now()).Before(importance(queued[j], r.Now())) }) var out []domain.Event for _, t := range queued { - if r.Retry.MaxAttempts > 0 && r.attempts[t.ID] >= r.Retry.MaxAttempts { + if r.Retry.MaxAttempts > 0 && t.Attempt >= r.Retry.MaxAttempts { e, err := r.fail(t) if err != nil { return out, err @@ -187,10 +174,6 @@ func (r *Router) AssignPending() ([]domain.Event, error) { if err != nil { continue } - // attempts is the failure counter checked against MaxAttempts - // above; it advances only on a non-rotation TaskReleased (see - // HandleEvent), not here, so a task that leases and rotates - // repeatedly is not double-counted toward the retry limit. out = append(out, e) if r.OnLease != nil { if err := r.OnLease(e); err != nil { @@ -234,7 +217,7 @@ func importance(t domain.Task, now time.Time) time.Time { return now.Add(-time.Duration(t.InherentPriority) * time.Hour) } func (r *Router) fail(t domain.Task) (domain.Event, error) { - b, _ := json.Marshal(map[string]any{"reason": "retry_limit", "attempts": r.attempts[t.ID]}) + b, _ := json.Marshal(map[string]any{"reason": "retry_limit", "attempts": t.Attempt, "failure_class": t.FailureClass}) e := domain.Event{ID: domain.NewID(), Type: "TaskFailed", TaskID: t.ID, Version: t.Version + 1, Payload: b, Surface: string(authz.System)} return e, r.Store.Append(e) } diff --git a/internal/router/router_test.go b/internal/router/router_test.go index f382ab9..f01a6fe 100644 --- a/internal/router/router_test.go +++ b/internal/router/router_test.go @@ -184,7 +184,9 @@ func TestQuotaWindowsAreIndependent(t *testing.T) { } // Case 2: only a 5h limit configured. The same 6h-old receipt is outside - // the 5h window and must not count. + // the 5h window and must not count; a fresh zero receipt proves the native + // usage source is known for this window. + report(now, 0) fiveHourOnly := QuotaAvailability{Store: s, Limits: map[string]QuotaWindowLimits{"h1": {FiveHour: 100}}, Now: func() time.Time { return now }} if !fiveHourOnly.Available(registry.Herdr{ID: "h1"}) { t.Fatal("receipt outside the 5h window incorrectly counted against it") @@ -198,4 +200,8 @@ func TestQuotaWindowsAreIndependent(t *testing.T) { if both.Available(registry.Herdr{ID: "h1"}) { t.Fatal("5h window should be exhausted at 90/100 (>=80%) regardless of weekly headroom") } + unknown := QuotaAvailability{Store: s, Limits: map[string]QuotaWindowLimits{"unknown": {FiveHour: 100}}, Now: func() time.Time { return now }} + if unknown.Available(registry.Herdr{ID: "unknown"}) { + t.Fatal("bounded harness without a native usage receipt must fail closed") + } } diff --git a/internal/store/store.go b/internal/store/store.go index 56080f9..41fcdad 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -131,6 +131,8 @@ func (s *Store) apply(e domain.Event) error { s.external[t.Source+"\x00"+t.ExternalID] = t.ID case "TaskLeased": t.State = domain.StateLeased + t.LifecyclePhase = "lease_issued" + t.LastError = "" epoch, _ := p["lease_epoch"].(string) if epoch == "" { // A pre-fencing event cannot safely be renewed by an old worker. @@ -145,13 +147,30 @@ func (s *Store) apply(e domain.Event) error { epoch = t.Lease.Epoch } t.Lease = &domain.Lease{HarnessID: p["harness_id"].(string), Epoch: epoch, Until: time.Unix(0, int64(p["until_ns"].(float64)))} + case "TaskLaunchAcknowledged": + t.LifecyclePhase = "started" case "TaskReleased": t.State = domain.StateQueued + t.LifecyclePhase = "reclaimed" t.Lease = nil t.HandoffRef, _ = p["handoff_ref"].(string) t.ReleaseTransaction, _ = p["transaction_id"].(string) t.ReleaseAnchor, _ = p["anchor_sha"].(string) t.PickupTransaction, t.PickupLeaseVersion = "", 0 + if t.HandoffRef == "" { + // A handoff-less release is a reclaim. Persist the retry decision + // here so expiry, pane exit, and a worker NACK all use the same + // crash-safe transition instead of router-local counters. + t.Attempt++ + t.FailureClass, _ = p["failure_class"].(string) + if t.FailureClass == "" { + t.FailureClass, _ = p["reason"].(string) + } + t.NextRetryAt = e.At.Add(retryBackoff(t.Attempt)) + } else { + t.NextRetryAt = time.Time{} + t.FailureClass = "" + } case "TaskPickupValidated": t.PickupTransaction, _ = p["transaction_id"].(string) if v, ok := p["lease_version"].(float64); ok { @@ -221,6 +240,12 @@ func (s *Store) apply(e domain.Event) error { } } } + if phase, ok := p["lifecycle_phase"].(string); ok && phase != "" { + t.LifecyclePhase = phase + } + if last, ok := p["last_error"].(string); ok { + t.LastError = last + } // Terminal and release events may carry an owner-produced snapshot from // immediately before a worker/coordinator drops its live session mapping. // Preserve it independently of the current task state so historical task @@ -244,6 +269,20 @@ func (s *Store) apply(e domain.Event) error { s.tasks[e.TaskID] = t return nil } + +func retryBackoff(attempt int) time.Duration { + if attempt < 1 { + attempt = 1 + } + backoff := time.Minute + for i := 1; i < attempt && backoff < 30*time.Minute; i++ { + backoff *= 2 + } + if backoff > 30*time.Minute { + return 30 * time.Minute + } + return backoff +} func (s *Store) Append(e domain.Event) error { s.mu.Lock() defer s.mu.Unlock() @@ -381,7 +420,7 @@ func (s *Store) validateTransition(e domain.Event, t domain.Task, exists bool, p return nil } switch e.Type { - case "TaskLeaseRenewed", "TaskReleased", "TaskPickupValidated", "TaskCompleted", "TaskBlocked", "TaskNeedsAttention", "TaskFailed": + case "TaskLeaseRenewed", "TaskLaunchAcknowledged", "TaskReleased", "TaskPickupValidated", "TaskCompleted", "TaskBlocked", "TaskNeedsAttention", "TaskFailed": owner, _ := p["harness_id"].(string) epoch, _ := p["lease_epoch"].(string) // Expiry is the one coordinator-owned relinquish path. It still binds diff --git a/internal/store/store_test.go b/internal/store/store_test.go index d8aca19..fe159fa 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -159,6 +159,38 @@ func TestNeedsAttentionRetainsFencedLeaseForLateCompletion(t *testing.T) { } } +func TestReclaimPersistsAttemptAndBackoffAcrossReopen(t *testing.T) { + dir := t.TempDir() + s, err := Open(dir) + if err != nil { + t.Fatal(err) + } + if err := s.Append(domain.Event{ID: "create", Type: "TaskCreated", TaskID: "t", Version: 1, Payload: []byte(`{"source":"s","external_id":"retry","project":"p"}`), Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + if _, err := s.Lease("t", "worker", time.Hour); err != nil { + t.Fatal(err) + } + task, _ := s.Task("t") + at := time.Date(2026, 7, 30, 12, 0, 0, 0, time.UTC) + p, _ := json.Marshal(map[string]any{"reason": "lease_expired", "failure_class": "worker_lost", "harness_id": task.Lease.HarnessID, "lease_epoch": task.Lease.Epoch, "expected_version": task.Version}) + if err := s.Append(domain.Event{ID: "reclaim", Type: "TaskReleased", TaskID: "t", Version: task.Version + 1, At: at, Payload: p, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + got, _ := s.Task("t") + if got.Attempt != 1 || got.FailureClass != "worker_lost" || !got.NextRetryAt.Equal(at.Add(time.Minute)) { + t.Fatalf("reclaim projection=%+v", got) + } + reopened, err := Open(dir) + if err != nil { + t.Fatal(err) + } + got, _ = reopened.Task("t") + if got.Attempt != 1 || !got.NextRetryAt.Equal(at.Add(time.Minute)) { + t.Fatalf("reopen lost retry state: %+v", got) + } +} + func TestOpenRebuildsOnlyFromLogAndIgnoresCorruptSnapshot(t *testing.T) { dir := t.TempDir() s, err := Open(dir)