4 changed files with 197 additions and 6 deletions
@@ -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
-			// its anchor protects nothing and its transaction can only retry
-			// a refusal forever. Blocked is different, because a reopen still
-			// produces a successor.
+			// Failed and completed are terminal: no successor will ever lease
+			// the task, so the anchor protects nothing and the transaction can
+			// only retry a refusal forever. Blocked is different, because a
+			// 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)
 				}
@@ -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)
+	}
+}
@@ -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 {
@@ -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)
+	}
+}