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)
 				}

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}
+}