Give an operator one way to retry a terminal task

A task that reached the router's MaxAttempts was permanently terminal.
TaskReleased only ever increments Attempt, TaskCorrected could not touch it,
and no HTTP route emitted a correction at all. The only way to work an
exhausted issue again was to invent a second task for it, which defeats
(source, external_id) dedupe and abandons the task's own history.

POST /v1/tasks/{id}/retry, full-control surfaces only. It requires the task
to be failed, unleased, and failed with reason retry_limit: restoring a retry
budget is not an answer to a failure that was not the budget running out. The
effect is one TaskCorrected naming that failure, setting state queued and
attempt 0 and clearing next_retry_at, failure_class and last_error. Task id,
source pair, goal, acceptance, decisions, work phase and artifact refs all
stay, and the original failure events stay in the log.

operation_id is required and makes the call idempotent, so a repeated request
cannot reset an attempt that has since started running.

This is RetryTask, not a generic correction endpoint: arbitrary task mutation
over HTTP is a different and much larger authority. It also does not address
F9, which is an operator releasing a lease someone else owns.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

cmd/orchestra/main.go

@@ -954,7 +954,7 @@ func main() {
 		}
 		var e domain.Event
 		var err error
-		actionTypes := map[string]string{"lease": "TaskLeased", "release": "TaskReleased", "complete": "TaskCompleted", "block": "TaskBlocked", "attention": "TaskNeedsAttention"}
+		actionTypes := map[string]string{"lease": "TaskLeased", "release": "TaskReleased", "complete": "TaskCompleted", "block": "TaskBlocked", "attention": "TaskNeedsAttention", "retry": "TaskCorrected"}
 		if typ, known := actionTypes[action]; known {
 			if err := authz.AuthorizeEvent(surface(r), typ); err != nil {
 				http.Error(w, err.Error(), http.StatusForbidden)
@@ -962,6 +962,26 @@ func main() {
 			}
 		}
 		switch action {
+		case "retry":
+			// Narrow on purpose: RetryTask restores a retry budget on a
+			// terminal task. There is no generic correction endpoint, because
+			// one would be arbitrary task mutation over HTTP.
+			var p struct {
+				OperationID string `json:"operation_id"`
+			}
+			if r.Body == nil || json.NewDecoder(r.Body).Decode(&p) != nil || p.OperationID == "" {
+				http.Error(w, "operation_id required", http.StatusBadRequest)
+				return
+			}
+			e, err = operations.RetryTask(s, surface(r), taskID, p.OperationID)
+			switch {
+			case errors.Is(err, domain.ErrNotFound):
+				http.Error(w, err.Error(), http.StatusNotFound)
+				return
+			case errors.Is(err, operations.ErrNotRetryable), errors.Is(err, domain.ErrInvalid):
+				http.Error(w, err.Error(), http.StatusConflict)
+				return
+			}
 		case "lease":
 			var p struct {
 				HarnessID  string `json:"harness_id"`

internal/domain/domain.go

@@ -437,6 +437,23 @@ func ValidatePayload(typ string, p map[string]any) error {
 				return fmt.Errorf("%w: state invalid", ErrInvalid)
 			}
 		}
+		if v, ok := p["attempt"]; ok {
+			n, ok := v.(float64)
+			if !ok || n < 0 || n != float64(int(n)) {
+				return fmt.Errorf("%w: attempt must be a non-negative whole number", ErrInvalid)
+			}
+		}
+		if v, ok := p["next_retry_at"]; ok {
+			s, ok := v.(string)
+			if !ok {
+				return fmt.Errorf("%w: next_retry_at must be a string", ErrInvalid)
+			}
+			if s != "" {
+				if _, err := time.Parse(time.RFC3339, s); err != nil {
+					return fmt.Errorf("%w: next_retry_at must be RFC3339 or empty", ErrInvalid)
+				}
+			}
+		}
 		if len(p) < 2 {
 			return fmt.Errorf("%w: correction must change at least one field", ErrInvalid)
 		}

internal/operations/retry.go

@@ -0,0 +1,114 @@
+package operations
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"orchestra/internal/authz"
+	"orchestra/internal/domain"
+	"orchestra/internal/store"
+)
+
+// ErrNotRetryable reports that a task cannot be given another retry budget.
+// Retry is deliberately narrow: it recovers a task the router gave up on, and
+// nothing else.
+var ErrNotRetryable = errors.New("task is not retryable")
+
+// RetryTask gives a retry-exhausted task a fresh attempt budget.
+//
+// A task that reaches the router's MaxAttempts is terminal, and nothing in the
+// log resets Attempt, so before this the only way to work an exhausted issue
+// again was to invent a second task for it. That defeats (source, external_id)
+// dedupe and loses the task's own history. This keeps the task: its id, its
+// source pair, its goal, acceptance, decisions, work phase and artifact refs
+// all stay as they are, and the original failure events stay in the log.
+//
+// This is not an escape hatch for a live lease. An owned task still has to be
+// released by its owner (F9); retrying a terminal task is a different
+// operation and stays separate on purpose.
+//
+// operationID makes the call idempotent. A retry that already ran under the
+// same id returns its event instead of resetting an attempt that has since
+// started running.
+func RetryTask(s *store.Store, surface authz.Surface, taskID, operationID string) (domain.Event, error) {
+	if err := authz.AuthorizeEvent(surface, "TaskCorrected"); err != nil {
+		return domain.Event{}, err
+	}
+	if operationID == "" {
+		return domain.Event{}, fmt.Errorf("%w: operation_id required", domain.ErrInvalid)
+	}
+	if e, ok := retryOperation(s, taskID, operationID); ok {
+		return e, nil
+	}
+	t, ok := s.Task(taskID)
+	if !ok {
+		return domain.Event{}, domain.ErrNotFound
+	}
+	if t.State != domain.StateFailed {
+		return domain.Event{}, fmt.Errorf("%w: task %s is %s, not failed", ErrNotRetryable, taskID, t.State)
+	}
+	if t.Lease != nil {
+		return domain.Event{}, fmt.Errorf("%w: task %s still holds a lease owned by %s", ErrNotRetryable, taskID, t.Lease.HarnessID)
+	}
+	failure, reason := lastFailure(s, taskID)
+	if failure.ID == "" {
+		return domain.Event{}, fmt.Errorf("%w: task %s has no failure to correct", ErrNotRetryable, taskID)
+	}
+	if reason != "retry_limit" {
+		// Retry restores a retry budget. A failure that was not the budget
+		// running out needs a decision about the failure itself, not another
+		// identical attempt.
+		return domain.Event{}, fmt.Errorf("%w: task %s failed with %q, not retry exhaustion", ErrNotRetryable, taskID, reason)
+	}
+	b, err := json.Marshal(map[string]any{
+		"corrects":      failure.ID,
+		"operation_id":  operationID,
+		"state":         string(domain.StateQueued),
+		"attempt":       0,
+		"next_retry_at": "",
+		"failure_class": "",
+		"last_error":    "",
+	})
+	if err != nil {
+		return domain.Event{}, err
+	}
+	e := domain.Event{ID: domain.NewID(), Type: "TaskCorrected", TaskID: taskID, Version: t.Version + 1, Payload: b, Surface: string(surface)}
+	if err := s.Append(e); err != nil {
+		return domain.Event{}, err
+	}
+	return e, nil
+}
+
+// retryOperation finds a retry already recorded under this operation id.
+func retryOperation(s *store.Store, taskID, operationID string) (domain.Event, bool) {
+	for _, e := range s.Events(0) {
+		if e.TaskID != taskID || e.Type != "TaskCorrected" {
+			continue
+		}
+		var p struct {
+			OperationID string `json:"operation_id"`
+		}
+		if json.Unmarshal(e.Payload, &p) == nil && p.OperationID != "" && p.OperationID == operationID {
+			return e, true
+		}
+	}
+	return domain.Event{}, false
+}
+
+// lastFailure returns the most recent TaskFailed event and its reason.
+func lastFailure(s *store.Store, taskID string) (domain.Event, string) {
+	var found domain.Event
+	var reason string
+	for _, e := range s.Events(0) {
+		if e.TaskID != taskID || e.Type != "TaskFailed" {
+			continue
+		}
+		var p struct {
+			Reason string `json:"reason"`
+		}
+		_ = json.Unmarshal(e.Payload, &p)
+		found, reason = e, p.Reason
+	}
+	return found, reason
+}

internal/operations/retry_test.go

@@ -0,0 +1,211 @@
+package operations
+
+import (
+	"encoding/json"
+	"errors"
+	"testing"
+
+	"orchestra/internal/authz"
+	"orchestra/internal/domain"
+	"orchestra/internal/store"
+)
+
+// exhausted builds a task that the router gave up on: three reclaims and a
+// retry-limit failure, with a work phase and a research ref to prove retry
+// preserves them.
+func exhausted(t *testing.T) *store.Store {
+	t.Helper()
+	s, err := store.Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	created := mustJSONBytes(t, map[string]any{
+		"source": "gitea:test-e2e", "external_id": "3", "project": "p",
+		"title": "extend the healthcheck", "description": "do the thing",
+		"acceptance": []string{"--help keeps working", "no arguments exits 0"},
+	})
+	if err := s.Append(domain.Event{ID: "task", TaskID: "task", Type: "TaskCreated", Version: 1, Payload: created, Surface: string(authz.System)}); err != nil {
+		t.Fatal(err)
+	}
+	phase := mustJSONBytes(t, map[string]any{"phase": string(domain.WorkPhaseResearch), "research_ref": domain.Hash([]byte("research"))})
+	task, _ := s.Task("task")
+	if err := s.Append(domain.Event{ID: "phase", TaskID: "task", Type: domain.EventWorkPhaseChanged, Version: task.Version + 1, Payload: phase, Surface: string(authz.System)}); err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 3; i++ {
+		lease(t, s, "task")
+		release(t, s)
+	}
+	task, _ = s.Task("task")
+	failed := mustJSONBytes(t, map[string]any{"reason": "retry_limit", "attempts": task.Attempt, "failure_class": "prompt_not_submitted"})
+	if err := s.Append(domain.Event{ID: "failed", TaskID: "task", Type: "TaskFailed", Version: task.Version + 1, Payload: failed, Surface: string(authz.System)}); err != nil {
+		t.Fatal(err)
+	}
+	return s
+}
+
+func release(t *testing.T, s *store.Store) {
+	t.Helper()
+	task, _ := s.Task("task")
+	p := mustJSONBytes(t, map[string]any{
+		"reason": "lease_expired", "failure_class": "prompt_not_submitted",
+		"harness_id": task.Lease.HarnessID, "lease_epoch": task.Lease.Epoch,
+	})
+	if err := s.Append(domain.Event{ID: domain.NewID(), TaskID: "task", Type: "TaskReleased", Version: task.Version + 1, Payload: p, Surface: string(authz.System)}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestRetryTaskRestoresTheAttemptBudget(t *testing.T) {
+	s := exhausted(t)
+	before, _ := s.Task("task")
+	if before.State != domain.StateFailed || before.Attempt != 3 {
+		t.Fatalf("setup: state=%s attempt=%d, want failed at 3", before.State, before.Attempt)
+	}
+	if _, err := RetryTask(s, authz.TUI, "task", "op-1"); err != nil {
+		t.Fatal(err)
+	}
+	got, _ := s.Task("task")
+	if got.State != domain.StateQueued || got.Attempt != 0 {
+		t.Fatalf("state=%s attempt=%d, want queued at 0", got.State, got.Attempt)
+	}
+	if !got.NextRetryAt.IsZero() || got.FailureClass != "" || got.LastError != "" {
+		t.Fatalf("retry left failure bookkeeping behind: %+v", got)
+	}
+	// The task is the same task, not a second one for the same issue.
+	if got.ID != before.ID || got.Source != before.Source || got.ExternalID != before.ExternalID {
+		t.Fatalf("identity changed: %s/%s/%s", got.ID, got.Source, got.ExternalID)
+	}
+	if got.Title != before.Title || got.Description != before.Description || len(got.Acceptance) != len(before.Acceptance) {
+		t.Fatalf("goal or acceptance changed: %+v", got)
+	}
+	if got.WorkPhase != before.WorkPhase || got.ResearchRef != before.ResearchRef {
+		t.Fatalf("phase artifacts changed: phase=%q research=%q", got.WorkPhase, got.ResearchRef)
+	}
+	// History is appended to, never rewritten.
+	var failures, releases int
+	for _, e := range s.Events(0) {
+		switch e.Type {
+		case "TaskFailed":
+			failures++
+		case "TaskReleased":
+			releases++
+		}
+	}
+	if failures != 1 || releases != 3 {
+		t.Fatalf("history lost: %d failures, %d releases", failures, releases)
+	}
+}
+
+func TestRetryTaskRefusesATaskThatIsNotTerminal(t *testing.T) {
+	s := exhausted(t)
+	if _, err := RetryTask(s, authz.TUI, "task", "op-1"); err != nil {
+		t.Fatal(err)
+	}
+	// Queued: nothing to recover.
+	if _, err := RetryTask(s, authz.TUI, "task", "op-2"); !errors.Is(err, ErrNotRetryable) {
+		t.Fatalf("err=%v, want ErrNotRetryable for a queued task", err)
+	}
+	lease(t, s, "task")
+	if _, err := RetryTask(s, authz.TUI, "task", "op-3"); !errors.Is(err, ErrNotRetryable) {
+		t.Fatalf("err=%v, want ErrNotRetryable for a leased task", err)
+	}
+}
+
+// A failure that was not the retry budget running out needs a decision about
+// the failure, not another identical attempt.
+func TestRetryTaskRefusesANonRetryFailure(t *testing.T) {
+	s, err := store.Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	created := mustJSONBytes(t, map[string]any{"source": "test", "external_id": "1", "project": "p", "title": "x"})
+	if err := s.Append(domain.Event{ID: "task", TaskID: "task", Type: "TaskCreated", Version: 1, Payload: created, Surface: string(authz.System)}); err != nil {
+		t.Fatal(err)
+	}
+	task, _ := s.Task("task")
+	p := mustJSONBytes(t, map[string]any{"reason": "unrecoverable"})
+	if err := s.Append(domain.Event{ID: "failed", TaskID: "task", Type: "TaskFailed", Version: task.Version + 1, Payload: p, Surface: string(authz.System)}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := RetryTask(s, authz.TUI, "task", "op-1"); !errors.Is(err, ErrNotRetryable) {
+		t.Fatalf("err=%v, want ErrNotRetryable", err)
+	}
+}
+
+// A gated surface holds an agent credential. Reviving a terminal task is an
+// operator decision.
+func TestRetryTaskRefusesAGatedSurface(t *testing.T) {
+	s := exhausted(t)
+	for _, surface := range []authz.Surface{authz.Agent, authz.MCP, authz.Maven, authz.Ntfy} {
+		if _, err := RetryTask(s, surface, "task", "op-1"); err == nil {
+			t.Fatalf("surface %s retried a terminal task", surface)
+		}
+	}
+	if got, _ := s.Task("task"); got.State != domain.StateFailed {
+		t.Fatalf("state=%s, want the task untouched", got.State)
+	}
+}
+
+// Idempotency: a repeated request under the same operation id must not reset
+// an attempt that has since started running.
+func TestRetryTaskIsIdempotentPerOperation(t *testing.T) {
+	s := exhausted(t)
+	first, err := RetryTask(s, authz.TUI, "task", "op-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	lease(t, s, "task")
+	again, err := RetryTask(s, authz.TUI, "task", "op-1")
+	if err != nil {
+		t.Fatalf("duplicate retry returned %v, want the recorded event", err)
+	}
+	if again.ID != first.ID {
+		t.Fatalf("event %s, want the original %s", again.ID, first.ID)
+	}
+	got, _ := s.Task("task")
+	if got.State != domain.StateLeased || got.Lease == nil {
+		t.Fatalf("duplicate retry disturbed a running attempt: %+v", got)
+	}
+}
+
+// Exhausting the budget a second time is retryable again, explicitly.
+func TestRetryTaskWorksAfterASecondExhaustion(t *testing.T) {
+	s := exhausted(t)
+	if _, err := RetryTask(s, authz.TUI, "task", "op-1"); err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 3; i++ {
+		lease(t, s, "task")
+		release(t, s)
+	}
+	task, _ := s.Task("task")
+	failed := mustJSONBytes(t, map[string]any{"reason": "retry_limit", "attempts": task.Attempt})
+	if err := s.Append(domain.Event{ID: "failed-2", TaskID: "task", Type: "TaskFailed", Version: task.Version + 1, Payload: failed, Surface: string(authz.System)}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := RetryTask(s, authz.TUI, "task", "op-2"); err != nil {
+		t.Fatal(err)
+	}
+	got, _ := s.Task("task")
+	if got.State != domain.StateQueued || got.Attempt != 0 {
+		t.Fatalf("state=%s attempt=%d, want queued at 0", got.State, got.Attempt)
+	}
+	// The correction names the second failure, not the first.
+	var corrects string
+	for _, e := range s.Events(0) {
+		if e.Type != "TaskCorrected" {
+			continue
+		}
+		var p struct {
+			Corrects    string `json:"corrects"`
+			OperationID string `json:"operation_id"`
+		}
+		if json.Unmarshal(e.Payload, &p) == nil && p.OperationID == "op-2" {
+			corrects = p.Corrects
+		}
+	}
+	if corrects != "failed-2" {
+		t.Fatalf("corrects=%q, want failed-2", corrects)
+	}
+}

internal/router/router_test.go

@@ -490,3 +490,70 @@ func backoffPayload(at time.Time) []byte {
 	b, _ := json.Marshal(map[string]any{"next_retry_at": at.UTC().Format(time.RFC3339Nano)})
 	return b
 }
+
+// F19: a retry-exhausted task the operator revived must be leasable again.
+// Before RetryTask nothing lowered Attempt, so the router refailed the task on
+// sight and an exhausted issue could never be worked again.
+func TestOperatorRetryMakesAnExhaustedTaskLeasableAgain(t *testing.T) {
+	s, err := store.Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	r, err := registry.New(registry.Config{
+		Projects: []registry.Project{{ID: "p", MachineAffinity: []string{"m"}}},
+		Machines: []registry.Machine{{ID: "m", Address: "unused"}},
+		Herdrs:   []registry.Herdr{{ID: "h", MachineID: "m", Concurrency: 1}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, _ := json.Marshal(map[string]any{"source": "test", "external_id": "a", "project": "p"})
+	if err := s.Append(domain.Event{ID: "a", TaskID: "a", Type: "TaskCreated", Version: 1, Payload: b, Surface: string(authz.System)}); err != nil {
+		t.Fatal(err)
+	}
+	// Each reclaim schedules a backoff, and the retry-limit check sits behind
+	// it, so the clock has to move for the router to reach the task at all.
+	clock := time.Now()
+	rt := Router{Store: s, Registry: r, Reachability: reachable{}, Retry: RetryPolicy{MaxAttempts: 3}, Now: func() time.Time { return clock }}
+	for i := 0; i < 3; i++ {
+		clock = clock.Add(time.Hour)
+		if _, err := rt.AssignPending(); err != nil {
+			t.Fatal(err)
+		}
+		task, _ := s.Task("a")
+		if task.Lease == nil {
+			t.Fatalf("round %d: task not leased", i)
+		}
+		p, _ := json.Marshal(map[string]any{
+			"reason": "lease_expired", "failure_class": "prompt_not_submitted",
+			"harness_id": task.Lease.HarnessID, "lease_epoch": task.Lease.Epoch,
+		})
+		if err := s.Append(domain.Event{ID: domain.NewID(), TaskID: "a", Type: "TaskReleased", Version: task.Version + 1, Payload: p, Surface: string(authz.System)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	clock = clock.Add(time.Hour)
+	if _, err := rt.AssignPending(); err != nil {
+		t.Fatal(err)
+	}
+	task, _ := s.Task("a")
+	if task.State != domain.StateFailed {
+		t.Fatalf("state=%s, want failed at the retry limit", task.State)
+	}
+
+	// The correction operations.RetryTask appends. Asserted here against the
+	// router, because the router is what refused the revived task.
+	fix, _ := json.Marshal(map[string]any{
+		"corrects": "a", "state": string(domain.StateQueued),
+		"attempt": 0, "next_retry_at": "", "failure_class": "",
+	})
+	if err := s.Append(domain.Event{ID: domain.NewID(), TaskID: "a", Type: "TaskCorrected", Version: task.Version + 1, Payload: fix, Surface: string(authz.TUI)}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := rt.AssignPending(); err != nil {
+		t.Fatal(err)
+	}
+	if got, _ := s.Task("a"); got.State != domain.StateLeased {
+		t.Fatalf("state=%s, want the revived task leased", got.State)
+	}
+}

internal/store/store.go

@@ -505,6 +505,22 @@ func (s *Store) apply(e domain.Event) error {
 				t.Lease = nil
 			}
 		}
+		// Retry recovery: the router treats Attempt as terminal once it
+		// reaches MaxAttempts, and no other event lowers it. See
+		// operations.RetryTask, which is the only intended producer.
+		if v, ok := p["attempt"].(float64); ok {
+			t.Attempt = int(v)
+		}
+		if v, ok := p["next_retry_at"].(string); ok {
+			if v == "" {
+				t.NextRetryAt = time.Time{}
+			} else if d, err := time.Parse(time.RFC3339, v); err == nil {
+				t.NextRetryAt = d
+			}
+		}
+		if v, ok := p["failure_class"].(string); ok {
+			t.FailureClass = v
+		}
 	}
 	// A question only stands while the task is blocked on it. Afterwards the
 	// answer is an ordinary standing decision and the log still holds the