Settle a release transaction deterministically in every case

F60, and the general rule F58 and F59 were reaching for one case at a
time: a transaction must settle or be abandoned deterministically, and
must never spin on an answer that cannot change.

Terminal now means failed or completed. Both drop the transaction and
free the session; nothing will ever lease either task again.

Blocked keeps the transaction, because a reopen returns the task to the
queue and that exact owner can still commit. TaskBlocked therefore
retains the ending epoch the way TaskReleased already did, or the
late-handoff path would have nothing to fence against after the reopen.

A refusal parks the commit instead of retrying every five seconds. It
is the coordinator's answer about who owns the task, so it stays true
until an event about that task arrives, and any such event un-parks it.
A reopen arrives as TaskCorrected, so the rule cannot be a list of
event types. Backoff runs 30s to a 5 minute cap.

A transport failure is not an answer and keeps retrying at once. That
distinction is the whole reason the park keys on a 4xx StatusError
rather than on any error at all.

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

cmd/orchestra-worker/main.go

@@ -157,6 +157,22 @@ type releaseTransaction struct {
 	AgentReleased bool      `json:"agent_released,omitempty"`
 	LastError     string    `json:"last_error,omitempty"`
 	UpdatedAt     time.Time `json:"updated_at"`
+	// NextAttemptAt parks a commit the coordinator has refused. A refusal is
+	// an answer about the task, not a transport failure, so it stays true
+	// until something about the task changes.
+	NextAttemptAt time.Time `json:"next_attempt_at,omitempty"`
+	Attempts      int       `json:"attempts,omitempty"`
+}
+
+// releaseBackoff spaces out refused commits. The first wait is long enough
+// that a parked transaction stops filling the observation ring, and the cap
+// keeps a reopen from waiting more than five minutes to be noticed.
+func releaseBackoff(attempts int) time.Duration {
+	d := 30 * time.Second << (attempts - 1)
+	if attempts < 1 || d > 5*time.Minute {
+		return 5 * time.Minute
+	}
+	return d
 }
 type projectConfig struct {
 	Repo           string   `json:"repo"`
@@ -894,18 +910,31 @@ func (w *worker) advanceRelease(ctx context.Context, id string, s herdr.Session)
 		w.releases[id] = tx
 		_ = w.save()
 	}
+	if tx.Phase == "anchor_pushed" && time.Now().Before(tx.NextAttemptAt) {
+		return
+	}
 	if tx.Phase == "anchor_pushed" {
 		// The epoch comes from the transaction, not from w.leases: an expiry
 		// replay deletes the lease, and the coordinator needs the epoch of the
 		// lease this anchor was pushed under to accept the late commit.
 		if err := w.api.Release(ctx, id, tx.Ref, tx.AnchorSHA, tx.ID, tx.LeaseEpoch, tx.LeaseVersion, w.sessionEvidence(ctx, id, s)); err != nil {
 			tx.LastError, tx.UpdatedAt = err.Error(), time.Now().UTC()
+			// A refusal is the coordinator's answer about who owns the task.
+			// It cannot change until an event about that task does, so asking
+			// again every five seconds only burns the observation ring. A
+			// transport failure is the opposite and must retry at once.
+			var refused *federation.StatusError
+			if errors.As(err, &refused) && refused.Code >= 400 && refused.Code < 500 {
+				tx.Attempts++
+				tx.NextAttemptAt = time.Now().UTC().Add(releaseBackoff(tx.Attempts))
+			}
 			w.releases[id] = tx
 			_ = w.save()
 			w.recordError(fmt.Errorf("release %s commit: %w", id, err))
 			return
 		}
 		tx.Phase, tx.LastError, tx.UpdatedAt = "event_committed", "", time.Now().UTC()
+		tx.NextAttemptAt, tx.Attempts = time.Time{}, 0
 		w.releases[id] = tx
 		_ = w.save()
 	}
@@ -1313,6 +1342,13 @@ func (w *worker) once(ctx context.Context) error {
 		if t, ok := created(e); ok {
 			w.tasks[t.ID] = t
 		}
+		// Any event about this task is the change a parked commit was waiting
+		// for. A reopen arrives as TaskCorrected, so this cannot be a list of
+		// specific types without going stale.
+		if tx, parked := w.releases[e.TaskID]; parked && !tx.NextAttemptAt.IsZero() {
+			tx.NextAttemptAt, tx.Attempts = time.Time{}, 0
+			w.releases[e.TaskID] = tx
+		}
 		if e.Type == "TaskLeased" {
 			var p struct {
 				HarnessID     string `json:"harness_id"`
@@ -1403,7 +1439,7 @@ func (w *worker) once(ctx context.Context) error {
 				w.leases[e.TaskID] = l
 			}
 		}
-		if e.Type == "TaskReleased" || e.Type == "TaskFailed" || e.Type == "TaskBlocked" {
+		if e.Type == "TaskReleased" || e.Type == "TaskFailed" || e.Type == "TaskBlocked" || e.Type == "TaskCompleted" {
 			if e.Type == "TaskReleased" {
 				var p struct {
 					TransactionID string `json:"transaction_id"`
@@ -1425,12 +1461,13 @@ func (w *worker) once(ctx context.Context) error {
 			// 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.
-			// A failed task is terminal: no successor will ever lease it, so
+			// Failed and completed are terminal: no successor will ever lease
-			// its anchor protects nothing and its transaction can only retry
+			// the task, so the anchor protects nothing and the transaction can
-			// a refusal forever. Blocked is different, because a reopen still
+			// only retry a refusal forever. Blocked is different, because a
-			// produces a successor.
+			// reopen returns the task to the queue and the epoch that ended is
+			// still on record, so that exact commit can still be accepted.
 			tx, releasing := w.releases[e.TaskID]
-			if !releasing || tx.Ref == "" || e.Type == "TaskFailed" {
+			if !releasing || tx.Ref == "" || e.Type == "TaskFailed" || e.Type == "TaskCompleted" {
 				if releasing {
 					delete(w.releases, e.TaskID)
 				}

cmd/orchestra-worker/main_test.go

@@ -1327,3 +1327,120 @@ func TestObservationRingEvictsLeastRecentlySeen(t *testing.T) {
 		t.Fatal("least recently seen entry survived")
 	}
 }
+
+// F60. A refusal is an answer about the task, not a transport failure, and it
+// stays true until something about that task changes. Run 10's blocked task
+// asked 5,000 times over seven hours and got the same 409 every time.
+func TestRefusedCommitParksUntilSomethingChanges(t *testing.T) {
+	var commits int
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case strings.HasSuffix(r.URL.Path, "/handoff"):
+			commits++
+			http.Error(w, "lease not owned", http.StatusConflict)
+		case strings.HasSuffix(r.URL.Path, "/events"):
+			_, _ = w.Write([]byte(`{"cursor":0,"events":[{"seq":1,"id":"c","type":"TaskCorrected","task_id":"t","version":9,"payload":{"state":"queued"},"surface":"web"}]}`))
+		default:
+			w.WriteHeader(http.StatusNoContent)
+		}
+	}))
+	defer s.Close()
+	tx := releaseTransaction{ID: "tx", Phase: "anchor_pushed", Ref: "sha256:abc", AnchorSHA: "abc", LeaseEpoch: "e1"}
+	w := &worker{
+		api:         federation.Client{BaseURL: s.URL, WorkerID: "h", Token: "t"},
+		harnessID:   "h",
+		tasks:       map[string]domain.Task{"t": {ID: "t"}},
+		sessions:    map[string]herdr.Session{"t": {PaneID: "pane"}},
+		leases:      map[string]lease{},
+		releases:    map[string]releaseTransaction{"t": tx},
+		quarantined: map[string]bool{},
+		statePath:   t.TempDir() + "/state.json",
+		hard:        .75,
+	}
+	w.advanceRelease(context.Background(), "t", w.sessions["t"])
+	w.advanceRelease(context.Background(), "t", w.sessions["t"])
+	if commits != 1 {
+		t.Fatalf("refused commit retried %d times without waiting", commits)
+	}
+	if w.releases["t"].NextAttemptAt.IsZero() {
+		t.Fatal("refused commit was not parked")
+	}
+	// A reopen arrives as TaskCorrected. Any event about the task is the
+	// change the parked commit was waiting for, so the same tick retries it
+	// and, still refused, parks it again.
+	if err := w.once(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if commits != 2 {
+		t.Fatalf("an event about the task did not un-park its commit, commits=%d", commits)
+	}
+	if w.releases["t"].NextAttemptAt.IsZero() {
+		t.Fatal("the second refusal did not park it again")
+	}
+}
+
+// The opposite case, and the one a backoff must not break: the coordinator is
+// unreachable or broken rather than answering. That says nothing about who
+// owns the task, so it has to retry at once.
+func TestTransientCommitFailureKeepsRetryingAtOnce(t *testing.T) {
+	var commits int
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/handoff") {
+			commits++
+			http.Error(w, "upstream unavailable", http.StatusBadGateway)
+			return
+		}
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer s.Close()
+	w := &worker{
+		api:         federation.Client{BaseURL: s.URL, WorkerID: "h", Token: "t"},
+		harnessID:   "h",
+		tasks:       map[string]domain.Task{"t": {ID: "t"}},
+		sessions:    map[string]herdr.Session{"t": {PaneID: "pane"}},
+		leases:      map[string]lease{},
+		releases:    map[string]releaseTransaction{"t": {ID: "tx", Phase: "anchor_pushed", Ref: "sha256:abc", AnchorSHA: "abc", LeaseEpoch: "e1"}},
+		quarantined: map[string]bool{},
+		statePath:   t.TempDir() + "/state.json",
+		hard:        .75,
+	}
+	w.advanceRelease(context.Background(), "t", w.sessions["t"])
+	w.advanceRelease(context.Background(), "t", w.sessions["t"])
+	if commits != 2 {
+		t.Fatalf("transport failure was parked like a refusal, commits=%d", commits)
+	}
+	if !w.releases["t"].NextAttemptAt.IsZero() {
+		t.Fatal("transport failure must not park the transaction")
+	}
+}
+
+// Completion is terminal for a release transaction just as failure is. The
+// task is done; nothing will ever lease it again to pick the anchor up.
+func TestCompletedTaskDropsItsReleaseTransaction(t *testing.T) {
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/events") {
+			_, _ = w.Write([]byte(`{"cursor":0,"events":[{"seq":1,"id":"c","type":"TaskCompleted","task_id":"t","version":9,"payload":{"report_ref":"sha256:r"},"surface":"system"}]}`))
+			return
+		}
+		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{"t": {ID: "t"}},
+		sessions:    map[string]herdr.Session{"t": {PaneID: "pane"}},
+		leases:      map[string]lease{},
+		releases:    map[string]releaseTransaction{"t": {ID: "tx", Phase: "anchor_pushed", Ref: "sha256:abc", AnchorSHA: "abc", LeaseEpoch: "e1"}},
+		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.releases) != 0 || len(w.sessions) != 0 {
+		t.Fatalf("completed task kept its release: releases=%v sessions=%v", w.releases, w.sessions)
+	}
+}

internal/store/store.go

@@ -424,6 +424,13 @@ func (s *Store) apply(e domain.Event) error {
 		t.Lease = nil
 	case "TaskBlocked", "TaskNeedsAttention":
 		if e.Type == "TaskBlocked" {
+			if t.Lease != nil {
+				// Same reason as TaskReleased: a worker may hold a pushed
+				// anchor whose commit was refused. A reopen returns the task
+				// to the queue, and the late-handoff path can only accept it
+				// if the epoch that ended is still on record.
+				t.LastLeaseEpoch = t.Lease.Epoch
+			}
 			t.State = domain.StateBlocked
 			t.Lease = nil
 		} else {

internal/store/store_test.go

@@ -693,3 +693,33 @@ func TestExpiryRetainsLeaseEpoch(t *testing.T) {
 		t.Fatalf("last lease epoch %q, want %q", after.LastLeaseEpoch, epoch)
 	}
 }
+
+// A worker can hold a pushed anchor whose commit was refused when an operator
+// blocks the task. A reopen returns it to the queue, and the late-handoff path
+// can only accept that exact owner if the epoch that ended is still recorded.
+func TestBlockRetainsLeaseEpochForALaterReopen(t *testing.T) {
+	s, err := Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Append(created("e1")); err != nil {
+		t.Fatal(err)
+	}
+	id := s.Tasks()[0].ID
+	if _, err := s.Lease(id, "h1", time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	leased, _ := s.Task(id)
+	epoch := leased.Lease.Epoch
+	p, _ := json.Marshal(map[string]any{"blocker": "parked by the operator", "harness_id": "h1", "lease_epoch": epoch})
+	if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskBlocked", TaskID: id, Version: leased.Version + 1, Payload: p, Surface: string(authz.System)}); err != nil {
+		t.Fatal(err)
+	}
+	after, _ := s.Task(id)
+	if after.State != domain.StateBlocked || after.Lease != nil {
+		t.Fatalf("expected a blocked unleased task, got %s lease=%v", after.State, after.Lease)
+	}
+	if after.LastLeaseEpoch != epoch || epoch == "" {
+		t.Fatalf("last lease epoch %q, want %q", after.LastLeaseEpoch, epoch)
+	}
+}