From edff02126508057904d6295e7f90daec7c7b4b2b Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 27 Aug 2026 13:07:48 +0400 Subject: [PATCH] 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 --- cmd/orchestra/main.go | 22 +++- internal/domain/domain.go | 17 +++ internal/operations/retry.go | 114 ++++++++++++++++ internal/operations/retry_test.go | 211 ++++++++++++++++++++++++++++++ internal/router/router_test.go | 67 ++++++++++ internal/store/store.go | 16 +++ 6 files changed, 446 insertions(+), 1 deletion(-) create mode 100644 internal/operations/retry.go create mode 100644 internal/operations/retry_test.go diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index b0db0c3..f4ab3a4 100644 --- a/cmd/orchestra/main.go +++ b/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"` diff --git a/internal/domain/domain.go b/internal/domain/domain.go index fb1acf1..d05c19e 100644 --- a/internal/domain/domain.go +++ b/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) } diff --git a/internal/operations/retry.go b/internal/operations/retry.go new file mode 100644 index 0000000..e766676 --- /dev/null +++ b/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 +} diff --git a/internal/operations/retry_test.go b/internal/operations/retry_test.go new file mode 100644 index 0000000..7da5bdc --- /dev/null +++ b/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) + } +} diff --git a/internal/router/router_test.go b/internal/router/router_test.go index 7fe2eae..0fa47ab 100644 --- a/internal/router/router_test.go +++ b/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) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index c631956..79b3d56 100644 --- a/internal/store/store.go +++ b/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