Stop publishing captures for a task the worker no longer holds

publishCaptures iterated every session the worker had, with no reference to
whether it still owned the lease. Run 10's task was blocked and unleased for
twenty-six minutes while this called the coordinator every five seconds and
logged "409 Conflict: lease not owned" each time.

The cost was not only the traffic. The single last_error slot stayed pinned to
that dead task, so run 11's own expiry reason was never visible, which is F18
turning a small leak into a blind spot.

A capture is lease-scoped. holdsLease names that invariant so the guard is
testable, rather than inline where a stub backend makes the test vacuous.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVbaKucEYBjMqVeUgJUsc1

cmd/orchestra-worker/main.go

@@ -1156,10 +1156,26 @@ func (w *worker) renewLeases(ctx context.Context) {
 	}
 }
 
+// holdsLease reports whether this worker still owns an unexpired lease on a
+// task. Anything it sends the coordinator about a task it no longer holds is
+// refused, so this is the guard that keeps a refusal from becoming a loop.
+func (w *worker) holdsLease(taskID string) bool {
+	l, ok := w.leases[taskID]
+	return ok && time.Now().Before(l.Until)
+}
+
 // publishCaptures makes remote panes observable without allowing the
 // coordinator to touch their unix herdr socket.
 func (w *worker) publishCaptures(ctx context.Context) {
 	for taskID, session := range w.sessions {
+		// A capture is lease-scoped. Without this the loop published every
+		// session it had ever held: run 10's task was blocked and unleased for
+		// twenty-six minutes while this called the coordinator every five
+		// seconds and logged "409 Conflict: lease not owned" each time, which
+		// also kept the single last_error slot pinned to a dead task.
+		if !w.holdsLease(taskID) {
+			continue
+		}
 		text, err := (herdr.CLIAdapter{Backend: w.executionBackend(), Harness: w.harness}).PaneCapture(ctx, session, "recent")
 		if err != nil {
 			continue

cmd/orchestra-worker/main_test.go

@@ -1102,3 +1102,24 @@ func TestStageExcludeSurvivesTheIgnoredMarker(t *testing.T) {
 		t.Fatalf("staged %q, want only a.txt", strings.TrimSpace(staged))
 	}
 }
+
+// A capture is lease-scoped. Without that guard, publishCaptures called the
+// coordinator for every session the worker had ever held: run 10's task was
+// blocked and unleased for twenty-six minutes while this logged "409 Conflict:
+// lease not owned" every five seconds, pinning the single last_error slot to a
+// dead task.
+func TestHoldsLeaseGatesWorkOnATaskTheWorkerLost(t *testing.T) {
+	w := &worker{leases: map[string]lease{
+		"held":    {Until: time.Now().Add(time.Minute)},
+		"expired": {Until: time.Now().Add(-time.Minute)},
+	}}
+	if !w.holdsLease("held") {
+		t.Error("an unexpired lease is not held")
+	}
+	if w.holdsLease("expired") {
+		t.Error("an expired lease is still held")
+	}
+	if w.holdsLease("never-leased") {
+		t.Error("a task this worker never leased is held")
+	}
+}