From 8174400b1ab11fea1fa018c2f29e87ff56c088bf Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 30 Jul 2026 14:37:34 +0400 Subject: [PATCH] Preserve leases needing recovery --- AUDIT.md | 11 +++++--- cmd/orchestra-worker/main.go | 11 +++++++- cmd/orchestra/main.go | 10 ++++---- internal/domain/domain.go | 11 +++++--- internal/domain/fuzz_test.go | 2 +- internal/integration/end_to_end_test.go | 9 ++++--- internal/operations/operations.go | 4 ++- internal/orchestrator/orchestrator.go | 10 ++++---- internal/orchestrator/rotation_test.go | 14 +++++----- internal/router/router.go | 2 +- internal/store/store.go | 27 ++++++++++++++------ internal/store/store_test.go | 34 +++++++++++++++++++++++++ 12 files changed, 106 insertions(+), 39 deletions(-) diff --git a/AUDIT.md b/AUDIT.md index 804b69a..c5a4b8a 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -29,10 +29,13 @@ released agent, or reject a valid completion. ## P1 — autonomy and recovery -- **Recovery:** `TaskBlocked` destroys the worker session needed for late - completion; aggregate-version changes also stale the lease. Use a separate - lease epoch and a recoverable `needs_attention` state that retains ownership - until explicit release, expiry, or reconciled completion. +- **Recovery:** **Closed 2026-07-30.** Launch/recovery faults now emit + `TaskNeedsAttention`, retaining the durable harness owner and lease epoch. + Renew, release, expiry, and a late reconciled completion accept that same + fenced lease; worker state advances its expected aggregate version without + 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. diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index ff78b74..3908b4c 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -814,6 +814,15 @@ func (w *worker) once(ctx context.Context) error { 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 + // sync so a late, otherwise valid completion is not self-staled. + if l, ok := w.leases[e.TaskID]; ok { + l.Version = e.Version + w.leases[e.TaskID] = l + } + } if e.Type == "TaskCompleted" { delete(w.leases, e.TaskID) if session, active := w.sessions[e.TaskID]; active { @@ -932,7 +941,7 @@ func (w *worker) reconcileLeases(ctx context.Context) error { active := make(map[string]lease) for _, task := range tasks { w.tasks[task.ID] = task - if task.State == domain.StateLeased && task.Lease != nil && task.Lease.HarnessID == w.harnessID { + if (task.State == domain.StateLeased || task.State == domain.StateNeedsAttention) && task.Lease != nil && task.Lease.HarnessID == w.harnessID { active[task.ID] = lease{Epoch: task.Lease.Epoch, HandoffRef: task.HandoffRef, TransactionID: task.ReleaseTransaction, AnchorSHA: task.ReleaseAnchor, Version: task.Version, Until: task.Lease.Until} } } diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 04fb534..ebda641 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -799,7 +799,7 @@ func main() { } var e domain.Event var err error - actionTypes := map[string]string{"lease": "TaskLeased", "release": "TaskReleased", "complete": "TaskCompleted", "block": "TaskBlocked"} + actionTypes := map[string]string{"lease": "TaskLeased", "release": "TaskReleased", "complete": "TaskCompleted", "block": "TaskBlocked", "attention": "TaskNeedsAttention"} if typ, known := actionTypes[action]; known { if err := authz.AuthorizeEvent(surface(r), typ); err != nil { http.Error(w, err.Error(), http.StatusForbidden) @@ -820,13 +820,13 @@ func main() { p.TTLSeconds = 1800 } e, err = s.Lease(taskID, p.HarnessID, time.Duration(p.TTLSeconds)*time.Second) - case "release", "complete", "block": + case "release", "complete", "block", "attention": t, ok := s.Task(taskID) if !ok { http.Error(w, "task not found", 404) return } - types := map[string]string{"release": "TaskReleased", "complete": "TaskCompleted", "block": "TaskBlocked"} + types := map[string]string{"release": "TaskReleased", "complete": "TaskCompleted", "block": "TaskBlocked", "attention": "TaskNeedsAttention"} var p map[string]any if r.Body == nil || json.NewDecoder(r.Body).Decode(&p) != nil || p == nil { http.Error(w, "invalid lifecycle payload", http.StatusBadRequest) @@ -836,7 +836,7 @@ func main() { http.Error(w, "reason or handoff_ref required", http.StatusBadRequest) return } - if action == "block" && p["blocker"] == nil { + if (action == "block" || action == "attention") && p["blocker"] == nil { http.Error(w, "blocker required", http.StatusBadRequest) return } @@ -1133,7 +1133,7 @@ func main() { http.Error(w, "task not found", 404) return } - ownedLease := t.State == domain.StateLeased && t.Lease != nil && t.Lease.HarnessID == parts[3] && t.Lease.Epoch == b.LeaseEpoch + ownedLease := (t.State == domain.StateLeased || t.State == domain.StateNeedsAttention) && t.Lease != nil && t.Lease.HarnessID == parts[3] && t.Lease.Epoch == b.LeaseEpoch // A response can be lost after the append/fsync. Retrying the exact // release transaction is therefore a successful no-op, never a second // TaskReleased event and never a reason to discard the predecessor. diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 2aff8a4..6805a3f 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -36,6 +36,11 @@ const ( StateCompleted TaskState = "completed" StateFailed TaskState = "failed" StateBlocked TaskState = "blocked" + // StateNeedsAttention records a recoverable fault without abandoning the + // current fenced lease. The owning worker may still reconcile a late + // completion, explicitly release it, or renew it while an operator + // investigates; expiry remains the only automatic reclaim. + StateNeedsAttention TaskState = "needs_attention" ) // BlockReason is the machine-readable diagnosis for a TaskBlocked event. @@ -185,7 +190,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, "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, "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) } @@ -308,7 +313,7 @@ func ValidatePayload(typ string, p map[string]any) error { if err := requiredString("reason"); err != nil { return err } - case "TaskBlocked": + case "TaskBlocked", "TaskNeedsAttention": if err := requiredString("blocker"); err != nil { return err } @@ -345,7 +350,7 @@ func ValidatePayload(typ string, p map[string]any) error { return fmt.Errorf("%w: state must be a string", ErrInvalid) } switch TaskState(s) { - case StateQueued, StateLeased, StateCompleted, StateFailed, StateBlocked: + case StateQueued, StateLeased, StateCompleted, StateFailed, StateBlocked, StateNeedsAttention: default: return fmt.Errorf("%w: state invalid", ErrInvalid) } diff --git a/internal/domain/fuzz_test.go b/internal/domain/fuzz_test.go index 1a1b1ca..7829935 100644 --- a/internal/domain/fuzz_test.go +++ b/internal/domain/fuzz_test.go @@ -11,7 +11,7 @@ import ( // producers happen to send well-formed payloads. var eventTypesUnderTest = []string{ "TaskCreated", "TaskLeased", "TaskReleased", "TaskCompleted", "TaskFailed", - "TaskBlocked", "ApprovalRequested", "ApprovalGranted", "ApprovalDenied", + "TaskBlocked", "TaskNeedsAttention", "ApprovalRequested", "ApprovalGranted", "ApprovalDenied", "TaskAmended", "QuotaReported", "StandupAdvisory", } diff --git a/internal/integration/end_to_end_test.go b/internal/integration/end_to_end_test.go index 9b1479c..22c320e 100644 --- a/internal/integration/end_to_end_test.go +++ b/internal/integration/end_to_end_test.go @@ -154,7 +154,7 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) { // handoff. Once TaskLeased carries a real handoff_ref, pickup correctly // refuses it instead of silently continuing (the valid pickup contract is // covered by the orchestrator continuity tests). - if got.State != domain.StateBlocked || got.Version != 5 { + if got.State != domain.StateNeedsAttention || got.Lease == nil || got.Version != 5 { t.Fatalf("invalid pickup state=%s version=%d", got.State, got.Version) } ref, err = s.PutArtifact([]byte("report")) @@ -162,8 +162,11 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) { t.Fatal(err) } if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: got.Version + 1, Surface: string(authz.System), Payload: mustJSON(map[string]any{ - "report_ref": ref, - "receipt": map[string]any{"harness_id": "h1", "consumed": 1}, + "report_ref": ref, + "receipt": map[string]any{"harness_id": "h1", "consumed": 1}, + "harness_id": got.Lease.HarnessID, + "lease_epoch": got.Lease.Epoch, + "expected_version": got.Version, })}); err != nil { t.Fatal(err) } diff --git a/internal/operations/operations.go b/internal/operations/operations.go index a74e985..fa4fa58 100644 --- a/internal/operations/operations.go +++ b/internal/operations/operations.go @@ -80,7 +80,7 @@ func AggregateQuota(events []domain.Event, from, to time.Time) map[string]float6 func StandupItems(tasks []domain.Task) []StandupItem { out := make([]StandupItem, 0) for _, t := range tasks { - if t.State == domain.StateQueued || t.State == domain.StateLeased || t.State == domain.StateBlocked { + if t.State == domain.StateQueued || t.State == domain.StateLeased || t.State == domain.StateNeedsAttention || t.State == domain.StateBlocked { out = append(out, StandupItem{t.ID, t.State, t.Title}) } } @@ -108,6 +108,8 @@ func BuildBrief(events []domain.Event, from, to time.Time, git map[string]GitSyn case "TaskBlocked": b.Blocked++ b.NeedsAttention = append(b.NeedsAttention, e) + case "TaskNeedsAttention": + b.NeedsAttention = append(b.NeedsAttention, e) case "ApprovalRequested": b.NeedsAttention = append(b.NeedsAttention, e) } diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 0ad3221..52d8ff5 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -445,7 +445,7 @@ func (c *Coordinator) Reconcile(ctx context.Context) error { c.mu.Lock() for taskID, session := range c.sessions { t, ok := c.Store.Task(taskID) - if ok && (t.State == domain.StateLeased || t.State == domain.StateBlocked) { + if ok && (t.State == domain.StateLeased || t.State == domain.StateNeedsAttention) { continue } if a, err := c.adapterFor(taskID, session); err == nil { @@ -589,7 +589,7 @@ func (c *Coordinator) expire(ctx context.Context) ([]domain.Event, error) { c.loadSessions() c.mu.Lock() for taskID, s := range c.sessions { - if t, ok := c.Store.Task(taskID); ok && t.State == domain.StateLeased { + if t, ok := c.Store.Task(taskID); ok && (t.State == domain.StateLeased || t.State == domain.StateNeedsAttention) { if a, ae := c.adapterFor(taskID, s); ae == nil { if p, ok := a.(herdr.PaneExit); ok { if exited, ee := p.PaneExited(ctx, s); ee == nil && exited { @@ -604,7 +604,7 @@ func (c *Coordinator) expire(ctx context.Context) ([]domain.Event, error) { var events []domain.Event var firstErr error for _, task := range c.Store.Tasks() { - if task.State != domain.StateLeased || task.Lease == nil || task.Lease.Until.After(time.Now()) { + if (task.State != domain.StateLeased && task.State != domain.StateNeedsAttention) || task.Lease == nil || task.Lease.Until.After(time.Now()) { continue } // Stop a local predecessor before making its lease eligible for a @@ -947,7 +947,7 @@ func (c *Coordinator) Start(ctx context.Context, e domain.Event) error { } if err != nil { // A UI-changing prompt can time out after herdr accepted it. Keep the - // live pane mapped before recording TaskBlocked so a later completion + // live pane mapped before recording TaskNeedsAttention so a later completion // can reconcile the lifecycle instead of becoming an orphan (B15). if s.PaneID != "" { s.HerdrID = p.HarnessID @@ -1034,7 +1034,7 @@ func (c *Coordinator) block(t domain.Task, reason string) error { p["expected_version"] = t.Version b, _ = json.Marshal(p) } - return c.Store.Append(domain.Event{ID: domain.NewID(), Type: "TaskBlocked", TaskID: t.ID, Version: t.Version + 1, Payload: b, Surface: string(authz.System)}) + return c.Store.Append(domain.Event{ID: domain.NewID(), Type: "TaskNeedsAttention", TaskID: t.ID, Version: t.Version + 1, Payload: b, Surface: string(authz.System)}) } func (c *Coordinator) Session(taskID string) (herdr.Session, bool) { diff --git a/internal/orchestrator/rotation_test.go b/internal/orchestrator/rotation_test.go index ad6ef8e..7483768 100644 --- a/internal/orchestrator/rotation_test.go +++ b/internal/orchestrator/rotation_test.go @@ -98,8 +98,8 @@ func TestPromptFailureRetainsLivePaneForBlockedTaskAcrossRestart(t *testing.T) { if err := c.Start(context.Background(), lease); err != nil { t.Fatal(err) } - if got, ok := s.Task(task.ID); !ok || got.State != domain.StateBlocked { - t.Fatalf("task state = %+v, want blocked", got) + if got, ok := s.Task(task.ID); !ok || got.State != domain.StateNeedsAttention || got.Lease == nil { + t.Fatalf("task state = %+v, want needs_attention with retained lease", got) } if session, ok := c.Session(task.ID); !ok || session.PaneID != "pane-created-before-timeout" || session.HerdrID != "h1" { t.Fatalf("retained session = %+v, present=%v", session, ok) @@ -168,8 +168,8 @@ func TestCoordinatorRefusesRemoteHerdrOperations(t *testing.T) { if a.leases != 0 { t.Fatal("remote adapter was started by coordinator") } - if task, ok := s.Task("remote"); !ok || task.State != domain.StateBlocked { - t.Fatalf("remote task state = %#v, present=%v; want blocked", task, ok) + if task, ok := s.Task("remote"); !ok || task.State != domain.StateNeedsAttention || task.Lease == nil { + t.Fatalf("remote task state = %#v, present=%v; want needs_attention with retained lease", task, ok) } } @@ -431,7 +431,7 @@ func TestTurnDecision(t *testing.T) { // TestStartBlocksOnInvalidPickup guards AUDIT.md's B6/Phase 4 item 4: // Coordinator.Start must run §6.2 pickup validation against the real // worktree before bootstrapping a successor onto a handoff_ref, and refuse -// (TaskBlocked) rather than bootstrap on a mismatched anchor. +// (TaskNeedsAttention) rather than bootstrap on a mismatched anchor. func TestStartBlocksOnInvalidPickup(t *testing.T) { repo := t.TempDir() run(t, repo, "init") @@ -476,8 +476,8 @@ func TestStartBlocksOnInvalidPickup(t *testing.T) { } got, ok := s.Task(task.ID) - if !ok || got.State != domain.StateBlocked { - t.Fatalf("expected TaskBlocked on invalid pickup, got state=%v ok=%v", got.State, ok) + if !ok || got.State != domain.StateNeedsAttention || got.Lease == nil { + t.Fatalf("expected recoverable needs_attention on invalid pickup, got state=%v lease=%v ok=%v", got.State, got.Lease, ok) } } diff --git a/internal/router/router.go b/internal/router/router.go index 7b8e840..8054fc5 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -221,7 +221,7 @@ func occupied(s *store.Store, id string, limit int) bool { } n := 0 for _, t := range s.Tasks() { - if t.State == domain.StateLeased && t.Lease != nil && t.Lease.HarnessID == id { + if (t.State == domain.StateLeased || t.State == domain.StateNeedsAttention) && t.Lease != nil && t.Lease.HarnessID == id { n++ } } diff --git a/internal/store/store.go b/internal/store/store.go index 69e9d43..56080f9 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -163,9 +163,15 @@ func (s *Store) apply(e domain.Event) error { case "TaskFailed": t.State = domain.StateFailed t.Lease = nil - case "TaskBlocked": - t.State = domain.StateBlocked - t.Lease = nil + case "TaskBlocked", "TaskNeedsAttention": + if e.Type == "TaskBlocked" { + t.State = domain.StateBlocked + t.Lease = nil + } else { + // Recovery diagnostics must not revoke the fenced owner. A late + // completion is still valid only from this exact lease epoch. + t.State = domain.StateNeedsAttention + } t.Blocker, _ = p["blocker"].(string) t.BlockReason = domain.InferBlockReason(t.Blocker) if v, ok := p["block_reason"].(string); ok && domain.BlockReason(v).Valid() { @@ -316,7 +322,7 @@ func (s *Store) Append(e domain.Event) error { if !taskExists && e.Type != "TaskCreated" && !global { return domain.ErrNotFound } - if e.Type != "TaskCreated" && (e.Type == "TaskCompleted" || e.Type == "TaskBlocked" || e.Type == "TaskReleased") { + if e.Type != "TaskCreated" && (e.Type == "TaskCompleted" || e.Type == "TaskBlocked" || e.Type == "TaskNeedsAttention" || e.Type == "TaskReleased") { var p map[string]any _ = json.Unmarshal(e.Payload, &p) for _, k := range []string{"handoff_ref", "report_ref"} { @@ -366,11 +372,16 @@ func (s *Store) validateTransition(e domain.Event, t domain.Task, exists bool, p if e.Type == "TaskLeased" && t.State != domain.StateQueued { return domain.ErrConflict } - if t.State != domain.StateLeased || t.Lease == nil { + // Needs-attention is specifically a recoverable leased state, never a + // second spelling of a terminal operator block on an unowned task. + if e.Type == "TaskNeedsAttention" && ((t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil) { + return domain.ErrConflict + } + if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil { return nil } switch e.Type { - case "TaskLeaseRenewed", "TaskReleased", "TaskPickupValidated", "TaskCompleted", "TaskBlocked", "TaskFailed": + case "TaskLeaseRenewed", "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 @@ -561,7 +572,7 @@ func (s *Store) RenewLease(id, harness, epoch string, expectedVersion int, ttl t if !ok { return domain.Event{}, domain.ErrNotFound } - if t.State != domain.StateLeased || t.Lease == nil || t.Lease.HarnessID != harness || t.Lease.Epoch != epoch || t.Version != expectedVersion { + if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil || t.Lease.HarnessID != harness || t.Lease.Epoch != epoch || t.Version != expectedVersion { return domain.Event{}, domain.ErrConflict } p, _ := json.Marshal(map[string]any{"harness_id": harness, "lease_epoch": epoch, "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": expectedVersion}) @@ -592,7 +603,7 @@ func (s *Store) ExpireLease(id string, now time.Time) (domain.Event, error) { if !ok { return domain.Event{}, domain.ErrNotFound } - if t.State != domain.StateLeased || t.Lease == nil || t.Lease.Until.After(now) { + if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil || t.Lease.Until.After(now) { return domain.Event{}, domain.ErrConflict } p, _ := json.Marshal(map[string]any{"reason": "lease_expired", "harness_id": t.Lease.HarnessID, "lease_epoch": t.Lease.Epoch, "expected_version": t.Version}) diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 0f27be2..d8aca19 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -125,6 +125,40 @@ func TestLeaseEpochFencesStaleOwnerLifecycleWrites(t *testing.T) { } } +func TestNeedsAttentionRetainsFencedLeaseForLateCompletion(t *testing.T) { + s, err := Open(t.TempDir()) + 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":"attention","project":"p"}`), Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + if _, err := s.Lease("t", "worker", time.Hour); err != nil { + t.Fatal(err) + } + leased, _ := s.Task("t") + attention, _ := json.Marshal(map[string]any{"blocker": "prompt response uncertain", "block_reason": "lease_failure", "harness_id": leased.Lease.HarnessID, "lease_epoch": leased.Lease.Epoch, "expected_version": leased.Version}) + if err := s.Append(domain.Event{ID: "attention", Type: "TaskNeedsAttention", TaskID: "t", Version: leased.Version + 1, Payload: attention, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + current, _ := s.Task("t") + if current.State != domain.StateNeedsAttention || current.Lease == nil || current.Lease.Epoch != leased.Lease.Epoch { + t.Fatalf("attention revoked or replaced lease: %+v", current) + } + report, err := s.PutArtifact([]byte("late completion")) + if err != nil { + t.Fatal(err) + } + completion, _ := json.Marshal(map[string]any{"report_ref": report, "receipt": map[string]any{"consumed": 1}, "harness_id": current.Lease.HarnessID, "lease_epoch": current.Lease.Epoch, "expected_version": current.Version}) + if err := s.Append(domain.Event{ID: "late-complete", Type: "TaskCompleted", TaskID: "t", Version: current.Version + 1, Payload: completion, Surface: string(authz.System)}); err != nil { + t.Fatalf("late completion from retained owner: %v", err) + } + completed, _ := s.Task("t") + if completed.State != domain.StateCompleted || completed.Lease != nil { + t.Fatalf("late completion did not settle task: %+v", completed) + } +} + func TestOpenRebuildsOnlyFromLogAndIgnoresCorruptSnapshot(t *testing.T) { dir := t.TempDir() s, err := Open(dir)