kami/orchestra
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Preserve leases needing recovery

Browse Source
This commit is contained in:
kami
2026-07-30 14:37:34 +04:00
parent f6ee0e3060
commit 8174400b1a
12 changed files with 106 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AUDIT.md
+7 -4
View File
@@ -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.
cmd/orchestra-worker/main.go
+10 -1
View File
@@ -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}
}
}
cmd/orchestra/main.go
+5 -5
View File
@@ -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.
internal/domain/domain.go
+8 -3
View File
@@ -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)
}
internal/domain/fuzz_test.go
+1 -1
View File
@@ -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",
}
internal/integration/end_to_end_test.go
+6 -3
View File
@@ -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)
}
internal/operations/operations.go
+3 -1
View File
@@ -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)
}
internal/orchestrator/orchestrator.go
+5 -5
View File
@@ -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) {
internal/orchestrator/rotation_test.go
+7 -7
View File
@@ -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)
}
}
internal/router/router.go
+1 -1
View File
@@ -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++
}
}
internal/store/store.go
+19 -8
View File
@@ -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})
internal/store/store_test.go
+34
View File
@@ -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)