diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index 3a7ef25..d781649 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -1214,7 +1214,17 @@ func (w *worker) once(ctx context.Context) error { // A releasing predecessor remains intentionally recoverable until // TaskPickupValidated for its transaction. Do not erase its pane // mapping merely because our own release event was replayed. - if _, releasing := w.releases[e.TaskID]; !releasing { + // + // Recoverable means an anchor was actually pushed. Before that + // tx.Ref is empty and no successor can pick anything up, so the + // mapping protects nothing. F30: a transaction stuck at "prepared" + // held the session forever once its pane was gone, health() kept + // reporting ActiveTask, and the harness never leased again. + tx, releasing := w.releases[e.TaskID] + if !releasing || tx.Ref == "" { + if releasing { + delete(w.releases, e.TaskID) + } if session, active := w.sessions[e.TaskID]; active { w.quarantine(ctx, e.TaskID, session) } diff --git a/cmd/orchestra-worker/main_test.go b/cmd/orchestra-worker/main_test.go index e12f065..a4967c8 100644 --- a/cmd/orchestra-worker/main_test.go +++ b/cmd/orchestra-worker/main_test.go @@ -955,3 +955,93 @@ func TestRenewLeasesIgnoresPaneInput(t *testing.T) { t.Fatalf("renewals=%d, want 1 after real progress", renewals) } } + +// TestBlockedTaskWithUnpushedReleaseFreesTheSessionSlot guards F30. A release +// transaction that never reached anchor_pushed has no artifact for a successor +// to pick up, so keeping its session mapping protects nothing. Live, one stuck +// at "prepared" pinned workpc-claude's only capacity slot: health() kept +// reporting ActiveTask and the harness never leased again, with no log line. +func TestBlockedTaskWithUnpushedReleaseFreesTheSessionSlot(t *testing.T) { + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/federation/events": + _, _ = w.Write([]byte(`{"cursor":0,"events":[{"seq":1,"id":"b","type":"TaskBlocked","task_id":"t","version":2,"payload":{"blocker":"parked"},"surface":"web"}]}`)) + case "/v1/federation/events/ack": + w.WriteHeader(http.StatusNoContent) + default: + // A worker with a live backend also polls captures and controls. + // Neither is what this test asserts on. + w.WriteHeader(http.StatusNoContent) + } + })) + defer s.Close() + w := &worker{ + api: federation.Client{BaseURL: s.URL, WorkerID: "h", Token: "t"}, + harnessID: "h", + backend: deadTmuxBackend(t), + tasks: map[string]domain.Task{}, + sessions: map[string]herdr.Session{"t": {PaneID: "pane"}}, + leases: map[string]lease{"t": {Epoch: "e", Version: 1}}, + releases: map[string]releaseTransaction{"t": {ID: "tx", Phase: "prepared"}}, + quarantined: map[string]bool{}, + statePath: t.TempDir() + "/state.json", + hard: .75, + } + if err := w.once(context.Background()); err != nil { + t.Fatal(err) + } + if len(w.sessions) != 0 { + t.Fatalf("session still pinned: %v", w.sessions) + } + if len(w.releases) != 0 { + t.Fatalf("unrecoverable transaction retained: %v", w.releases) + } + if got := w.health(context.Background()).ActiveTask; got != "" { + t.Fatalf("ActiveTask=%q, worker still advertises itself as busy", got) + } +} + +// The opposite branch must not regress: once an anchor exists, a successor can +// still pick it up, so the mapping stays until TaskPickupValidated. +func TestBlockedTaskWithPushedAnchorKeepsItsSession(t *testing.T) { + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/federation/events": + _, _ = w.Write([]byte(`{"cursor":0,"events":[{"seq":1,"id":"b","type":"TaskBlocked","task_id":"t","version":2,"payload":{"blocker":"parked"},"surface":"web"}]}`)) + case "/v1/federation/events/ack": + w.WriteHeader(http.StatusNoContent) + default: + t.Fatalf("unexpected %s", r.URL.Path) + } + })) + defer s.Close() + w := &worker{ + api: federation.Client{BaseURL: s.URL, WorkerID: "h", Token: "t"}, + harnessID: "h", + tasks: map[string]domain.Task{}, + sessions: map[string]herdr.Session{"t": {PaneID: "pane"}}, + leases: map[string]lease{"t": {Epoch: "e", Version: 1}}, + releases: map[string]releaseTransaction{"t": {ID: "tx", Phase: "anchor_pushed", Ref: "sha256:abc"}}, + quarantined: map[string]bool{}, + statePath: t.TempDir() + "/state.json", + hard: .75, + } + if err := w.once(context.Background()); err != nil { + t.Fatal(err) + } + if len(w.sessions) != 1 || len(w.releases) != 1 { + t.Fatalf("recoverable handoff dropped: sessions=%v releases=%v", w.sessions, w.releases) + } +} + +// deadTmuxBackend stands in for a runtime whose server is gone. tmux answers +// "no server running", which TmuxBackend.Kill reports as an already-dead +// session rather than an error. +func deadTmuxBackend(t *testing.T) *herdr.TmuxBackend { + t.Helper() + stub := filepath.Join(t.TempDir(), "tmux") + if err := os.WriteFile(stub, []byte("#!/bin/sh\necho 'no server running' >&2\nexit 1\n"), 0o755); err != nil { + t.Fatal(err) + } + return &herdr.TmuxBackend{Socket: "gone", Binary: stub} +}