diff --git a/docs/logs/engineering-log.md b/docs/logs/engineering-log.md index a2ca2ea9..7828a5ae 100644 --- a/docs/logs/engineering-log.md +++ b/docs/logs/engineering-log.md @@ -66,6 +66,148 @@ `./scripts/test-regression.sh` passed at 85.6% total coverage with zero uncovered functions. +## 2026-07-31 (Terminal Status/Event Atomicity — Issue #1067) + +- Symptom: aggregate race load exposed `RunStatusFailed` from `GetRun` while + the same run's immediate replay ended at `llm.turn.requested` without + `run.failed`; code inspection found the same status-first window on completed + and cancelled paths. +- Cause: every terminal helper called `setStatus` before `emit`, splitting the + public run record from the event journal's ledger, bounded store append, + subscriber fanout, and recorder drain. +- Deterministic red: a no-sleep transition barrier reproduced all three states. + Completed replay lacked `run.completed`, failed replay contained the required + `error.context` but lacked `run.failed`, and cancelled replay lacked + `run.cancelled` while `GetRun` already returned each terminal status. +- Fix: one `transitionTerminal` seam now lets the winning terminal emit seal and + append the matching event, completes bounded store append and ordered recorder + dispatch/drain, conditionally persists the matching status, commits in-memory + status, then fans out. Every status transition shares a per-run mutex, so a + delayed running/waiting snapshot cannot overwrite terminal state. +- Preserved reliability: terminal store I/O remains outside `Runner.mu`, and + status-store I/O remains outside the global conversation journal lock; + a refcounted per-conversation sequence guard prevents same-conversation + overtaking while unrelated `GetRun` and unrelated event journals stay + responsive, then reclaims idle keys. Terminal redaction sealing, event IDs, + causal/error snapshot order, recorder order, backup, and pruning contracts + remain intact. +- Failure policy: retained terminal status persistence is never attempted when + terminal append reports failure. If append succeeds but final status update + errors or reaches its context deadline, durable status may remain + non-terminal while the durable event, in-memory terminal state, and subscriber + fanout proceed. This is the strongest one-way guarantee available without a + store transaction and does not claim two-way atomicity. +- Explicit exception: existing terminal `StorageModeNone` configurations still + suppress the matching replay event while sealing and publishing status, as + pinned by terminal-redaction tests. The stronger replay implication applies + to terminal events retained by policy. +- Review regressions: append failure cannot persist terminal status; status + update failure/timeout still completes live publication; unrelated + conversations progress during terminal I/O while target events cannot + overtake; delayed non-terminal status cannot overwrite terminal; explicit + terminal redaction waits for recorder drain; and contended/distinct keyed + locks reclaim to zero. +- Exact-head retention cause: the terminal event success marker made a run + eligible for pruning even when the matching final `UpdateRun` failed, because + that return value was discarded. The truthful in-memory terminal state could + be evicted while the durable row remained running. Both append- and + status-failure exceptions could also accumulate without an admission bound. +- Retention/admission fix: terminal event resolution and terminal status + persistence are tracked separately. Store-backed pruning requires both; + `StorageModeNone` is explicit event suppression plus durable status, while + no-store runs remain process-local. Both unresolved append and status states + count toward `MaxCompletedRetention`. At the cap, Start/Continue retry only + status gaps under one shared deadline of at most 250 ms and otherwise return + `TerminalDurabilityBackpressureError`; their HTTP routes map it to 503 + `terminal_durability_unavailable`. +- Recovery and boundedness: no recovery store I/O holds Runner, status, + event-journal, or conversation locks. Status overwrite retries are safe and + immediately restore the completed-retention bound before reopening admission + once acknowledged. Ambiguous failed appends are protected but never retried + because a third-party store may have applied the append. + Already-admitted work finishes and remains visible; admission closes at the + cap so permanent failure growth stops at that finite admitted population. +- Exact-head regressions: retention 1 preserves several already-admitted + UpdateRun-failed completions while durable rows stay non-terminal; concurrent + admissions reject during outage and recover under race; one blocked retry + proves the shared deadline and unlocked state/journal access; append failure, + StorageModeNone, no-store, Continue error precedence, and both HTTP 503 routes + are pinned. +- Concurrent review red: a phase hook paused Continue immediately after its + completed-source validation. A concurrent Start recovered three pending + statuses, pruned the oldest validated source at retention 1, and the resumed + Continue deterministically failed with `run not found`. +- Concurrent review fix: validation now increments an in-state continuation + reservation under `Runner.mu`. The one shared completed-run prune candidate + filter excludes reserved sources for every caller. A defer releases on every + success/error path and immediately re-prunes; the existing later write-lock + `continued` check still chooses exactly one continuation winner. Recovery + store I/O remains outside Runner/status/journal/conversation locks. +- Gate-test correction: full race exposed one test treating terminal replay as + proof status had already committed. The one-way contract is the reverse: + terminal status implies replay, while replay may lead status. The test now + waits independently for failed status and retains both replay/status checks. +- Verification: the concurrent red is green with cleanup and single-winner + controls normal/race at `-count=100`; the expanded durability harness and HTTP + suites pass normal/race at `-count=100`; complete `internal/harness` plus + `internal/server` normal/race and affected `go vet` pass. The final direct + foreground non-TTY `./scripts/test-regression.sh` passes normal, full race, + and `coveragegate: PASS (total=85.7%, min=80.0%, zero-functions=0)`. +- Hosted settled-helper symptom: exact rebased race run `30656467482` failed + `TestRunnerHookErrorFailOpen` because `collectRunEvents` returned terminal + replay history before the valid later status commit; the immediate `GetRun` + still reported `running`. +- Audit/cause: 215 `collectRunEvents` references, its sole configurable-timeout + caller, and 79 `collectEvents` snapshot references were reviewed. No shared + collector caller intentionally observes the event-leading-status window. + Exact ordering regressions use direct phase hooks/subscriptions; snapshot + consumers either wait separately or intentionally inspect nonterminal state. +- Deterministic red/fix: a pre-status terminal barrier made replay visible and + failed the old collector immediately with `returned before terminal status + commit`. Both collector variants now preserve event assertions and then poll + for any terminal status within the same total deadline. A missing status is a + timeout failure rather than a false settled result; production order is + unchanged. +- Hosted-equivalent verification: the collector, all hook scenarios, and the + configurable-timeout caller pass normal/race at `-count=100`; `make + test-race` passes. The final outside-sandbox foreground + `TMPDIR=/private/tmp GOCACHE=/private/tmp/gocode-go-cache + ./scripts/test-regression.sh` passes normal, full race, and + `coveragegate: PASS (total=85.7%, min=80.0%, zero-functions=0)`. GitHub + comment publication was blocked by external-write safety review and was not + retried; this repository evidence records the run. +- Exact-head helper review: commit `8757e8a3` still let settlement succeed when + a closed stream had no terminal event, or when the sole terminal event did + not match the later terminal status. This was a P1 test-integrity defect, not + a production-path defect: it could mask the exact #1067 invariant across the + shared collector callers. +- Review TDD reds: a completed run plus closed stream containing only + `run.started` returned success, and failed status plus `run.completed` + returned success. Both failures were immediate and deterministic. +- Review fix: the test-only subscribed-stream core now requires exactly one + terminal event and requires its completed/failed/cancelled meaning to match + the observed terminal status. Event slices remain unchanged on success and + error, and both event collection and status settlement consume the original + single deadline. The phase regression now waits for an explicit settlement + entry signal before proving the collector cannot return ahead of status. +- Review verification: the two regressions, explicit settlement barrier, hook + family, and configurable-timeout caller pass normal/race at `-count=100`. + Outside-sandbox `make test-race` passes. The authoritative foreground + `TMPDIR=/private/tmp GOCACHE=/private/tmp/gocode-go-cache + ./scripts/test-regression.sh` passes normal, full race, and + `coveragegate: PASS (total=85.7%, min=80.0%, zero-functions=0)` on this exact + follow-up diff. +- Promotion integration regression: the semantic merge with #1054 initially + persisted every non-terminal status before committing it in memory. A failed + best-effort `UpdateRun` therefore left an executing run visibly stale; the + AskUser broker could own a live pending question while API, TUI, and GUI + still saw `running` instead of `waiting_for_user`. A deterministic failing + store test reproduced `queued` after a requested transition to `running`. + The unified per-run status lock now commits live non-terminal state first, + retaining the persistence attempt and false return so strict + waiting-status/event publication can retry without making the pending prompt + invisible. Terminal transitions retain their event-before-status contract. + ## 2026-07-31 (Provider-Key Matrix Health Wait — Issue #1062) - Symptom: hosted race run `30583930460` failed diff --git a/docs/logs/long-term-thinking-log.md b/docs/logs/long-term-thinking-log.md index 8457c4e8..d8b34bf7 100644 --- a/docs/logs/long-term-thinking-log.md +++ b/docs/logs/long-term-thinking-log.md @@ -56,6 +56,64 @@ repository regression gate pass. The local commit is ready for parent promotion; PR #1069 remains open and unmerged pending hosted reruns. +## 2026-07-31 (Terminal Status/Event Atomicity — Issue #1067) + +- Command intent: repair the aggregate-load Runner race where `GetRun` exposes + completed, failed, or cancelled before the matching terminal event and + required causal evidence are replayable. +- User intent: external monitoring must never report a terminal result from an + incomplete transcript, and this engine repair must remain isolated from PR + #1060/#1055. +- Success definition: deterministic no-sleep red evidence covers all terminal + statuses; one shared lifecycle seam makes terminal ledger/store publication + precede status visibility while preserving recorder order; immediate + Subscribe and HTTP SSE replay agree; focused stress, affected race/vet, full + regression, and hosted checks pass on one unmerged closing PR. Concurrent + Start recovery/pruning must not evict a completed source after Continue has + validated it but before that continuation performs its single-winner commit. + Test helpers that promise a collected terminal run must preserve event + assertions while independently requiring the later terminal status, without + weakening the production event-first order or direct barrier regressions. +- Non-goals: conversation cursor redesign, cron/callback behavior, client UI, + provider routing, schemas, or workflow timing changes. +- Guardrails: preserve out-of-lock bounded store writes, terminal sealing, + recorder drain order, cleanup order, causal/error snapshots, SSE IDs, and + unrelated conversation responsiveness. Do not claim a cross-record store + transaction the interface cannot provide. Never evict truthful terminal state + without both event and status durability, and never trade that protection for + unbounded memory growth during permanent store failure. +- Outcome: one winner-only transition now publishes terminal ledger history + before matching in-memory status and subscriber fanout. Retained terminal + status persistence is attempted only after event append reports success; + append/status failures have explicit bounded live-availability behavior. + Shared per-run status serialization prevents a delayed non-terminal overwrite, + and refcounted per-conversation sequencing avoids both overtaking and lock-map + growth. Exact-head retention hardening now tracks event and status durability + separately, requires both before pruning, and closes new admission when the + combined unresolved backlog reaches `MaxCompletedRetention`. Status recovery + uses one unlocked deadline capped at 250 ms; Start/Continue expose typed HTTP + 503 while no-store and intentional StorageModeNone policies remain distinct. + Successful status recovery immediately restores the retention bound before + reopening admission. A temporary in-state reservation now protects a + validated Continue source from every prune caller across unlocked recovery; + release on all exits restores the retention policy without weakening the + existing single-winner check. Focused normal/race and real HTTP mapping tests + plus affected normal/race/vet are green. The unchanged foreground repository + gate passes normal, full race, and 85.7% coverage with zero uncovered + production functions on the prior follow-up diff. After hosted race run + `30656467482`, the shared test collector now treats terminal history plus the + later terminal status as its settled boundary under one total deadline. The + event-first production contract and direct phase tests remain unchanged; + affected normal/race x100 and hosted-equivalent `make test-race` pass. The + final outside-sandbox foreground repository gate passes normal, full race, + and 85.7% coverage with zero uncovered production functions. Exact-head + review then hardened the helper boundary again: settlement now requires + exactly one terminal event whose meaning matches terminal status, and the + phase regression proves it entered settlement before asserting non-return. + Focused normal/race x100, hosted-equivalent race, and the final foreground + repository gate pass; coverage remains 85.7% with zero uncovered production + functions on this follow-up diff. + ## 2026-07-31 (Provider-Key Matrix Health Wait — Issue #1062) - Command intent: isolate and clear the hosted race blocker first observed on diff --git a/docs/logs/observational-log.md b/docs/logs/observational-log.md index e751c84a..a506c86b 100644 --- a/docs/logs/observational-log.md +++ b/docs/logs/observational-log.md @@ -45,6 +45,70 @@ Use this file for observations about system behavior without immediately prescri calls before invoking `Shutdown`; otherwise cleanup can wait on the very run the fixture still holds blocked. +## 2026-07-31 (Terminal Run Publication Window) + +- Concurrency observation: a terminal event can be prepared under the Runner + lock and persisted outside it without blocking unrelated run queries, but the + matching status needs one explicit commit point between persistence and + subscriber fanout. Status before preparation yields incomplete replay; + status after fanout lets terminal-event consumers briefly read `running`. +- Testing observation: a phase channel at the terminal transition boundary + deterministically exposes the forbidden state for completed, failed, and + cancelled paths without relying on aggregate load or fixed sleeps. +- Replay observation: after an HTTP status poll returns terminal, reconnecting + run SSE from the first event ID must replay exactly one matching terminal + event and no terminal event of another status. +- Conversation-stream observation: terminal persistence and terminal fanout + must share the same per-conversation sequence; the global journal lock can be + released for slow recorder/status persistence only if later events and + subscriptions on that conversation cannot overtake the terminal event. +- Durability observation: the current store interface has separate + `AppendEvent` and `UpdateRun` calls, so it cannot promise two-way atomicity. + The enforceable direction is: never attempt retained terminal status + persistence after terminal append reports failure. If status update then + fails or times out, durable event may lead durable status while bounded + in-memory status and fanout still complete. +- Resource observation: a keyed sequence lock needs waiter-inclusive reference + accounting. Deleting only after owners and queued waiters release prevents a + second lock generation for the same key and avoids an unbounded conversation + map. +- Retention observation: event append success alone is insufficient proof that + a terminal state can fall back to the store. Pruning also needs acknowledged + terminal status persistence; otherwise fallback resurrects a non-terminal + durable row after evicting the only truthful process-local result. +- Outage observation: protecting unpersisted truth and bounding memory require + one admission boundary. Both ambiguous event appends and failed status + updates consume it; already-admitted runs may finish above the numeric cap, + but later admissions stop growth once the outage is observed. +- Recovery observation: `UpdateRun` is an idempotent overwrite and can be + retried under one shared short context. `AppendEvent` is not safe to retry + after an ambiguous third-party error because the append may already exist. + Once status retries succeed, pruning newly durable candidates immediately + restores the retention bound before another admission is accepted. +- Policy observation: no-store and `StorageModeNone` are distinct. No-store has + no durable fallback and stays process-local; StorageModeNone intentionally + resolves the event side while its final status can still make safe pruning + possible. +- Continuation observation: preserving a source only in Continue's own recovery + prune is insufficient because concurrent Start recovery calls the same prune + policy without that local argument. A reservation stored on the source and + checked by the shared candidate filter protects it across every prune caller. +- Contract observation: terminal replay can lead the later status commit during + event-first publication. Tests that assert both must wait independently for + status; only terminal status is guaranteed to imply matching replay. +- Helper-audit observation: aggregate race load repeatedly finds stale tests + when a helper named as event collection is treated implicitly as run + settlement. Shared callers all want settlement, while intentional window + probes use direct `Subscribe`; encoding the distinction once in the test + helper prevents the next immediate-`GetRun` variant without changing replay. +- Settlement observation: waiting for any terminal status is insufficient if + the collected transcript is absent or contradictory. A settled test result + requires exactly one terminal event whose completed/failed/cancelled meaning + matches status; stream closure is not evidence of transcript completeness. +- Synchronization observation: a timed non-return assertion is meaningful only + after the tested goroutine proves it reached the intended blocking phase. An + explicit settlement-entry handshake removes scheduler delay as a false pass. + ## 2026-07-31 (Source-Workflow Dual-Error Arbitration) - Process observation: a child can exit non-zero while closing its stdin also diff --git a/docs/logs/system-log.md b/docs/logs/system-log.md index 8f6b6680..436b72aa 100644 --- a/docs/logs/system-log.md +++ b/docs/logs/system-log.md @@ -44,6 +44,66 @@ or tool contract changes; existing queue-drain, timeout, and idempotency behavior remains the rollback boundary. +## 2026-07-31 (Terminal Run Transition Publication — Issue #1067) + +- System/component: `Runner.transitionTerminal`, `Runner.emit`, and + `eventJournal` terminal persistence/fanout in `internal/harness`. +- Responsibilities: the event journal remains the only event-ledger writer and + terminal-seal owner; the transition seam binds the winning terminal event to + its completed, failed, or cancelled `Run` record. +- Order: prior causal/error events -> terminal ledger append/seal -> bounded + event-store append -> ordered terminal recorder dispatch/drain -> matching + conditional status persistence -> matching in-memory status -> subscriber + fanout -> backup/pruning lifecycle. +- Concurrency boundary: store and recorder I/O remain outside `Runner.mu`. + A per-conversation sequence guard preserves replay-to-live ordering across + the whole transition. The global `conversationEventMu` is released around + recorder and status-store I/O, so unrelated conversation journals progress; + the in-memory status commit briefly reacquires only the Runner state lock. +- Consumers: `GetRun`, run summary, run SSE replay/live delivery, + conversation replay, CLI/TUI exit handling, and macOS transcript state keep + existing schemas and event IDs. +- Failure boundary: retained terminal `UpdateRun` is attempted only after + `AppendEvent` reports success. Append failure leaves durable status + non-terminal; later status-update failure/timeout may leave durable terminal + event ahead of durable status. Both remain non-fatal to bounded in-memory + status/fanout, so this is explicitly one-way rather than transactional + two-way atomicity. Terminal redaction drops remain the explicit no-event + exception and now drain the recorder before publishing status. +- Status/resource boundary: every status transition shares a per-run mutex, so + delayed non-terminal writes cannot overwrite terminal state. The + per-conversation sequence lock counts queued waiters and deletes idle keys; + external terminal I/O never holds the global conversation journal mutex. +- Retention/admission boundary: store-backed terminal pruning requires event + persistence (or intentional StorageModeNone suppression) plus acknowledged + final status persistence. Both unresolved event and status records consume + the `MaxCompletedRetention` durability backlog. Once full, StartRun and + ContinueRun retry status-only gaps under one shared deadline capped at 250 ms + with no store I/O under Runner/status/journal/conversation locks, then return + typed fail-closed backpressure if unresolved. +- API/recovery boundary: the two run-admission HTTP routes map typed durability + backpressure to 503 `terminal_durability_unavailable`; Continue validates + missing/non-completed sources first and revalidates before its single-winner + mutation. Successful status retry immediately prunes newly durable candidates + back to the configured retention limit before reopening admission. Ambiguous + append errors remain blocked until process/operator recovery rather than + risking a duplicate forensic event. No-store runs intentionally remain + process-local and ungated. +- Continuation reservation boundary: source validation and reservation are one + `Runner.mu` mutation; unlocked durability recovery follows; the existing + write-lock revalidation remains the only single-winner mutation. The shared + prune candidate filter excludes nonzero reservations regardless of caller. + Deferred release decrements the in-state counter and immediately invokes the + shared lock-held prune policy, so no keyed side map or stale reservation entry + survives success, backpressure, revalidation loss, or dispatch failure. +- Test settlement boundary: `collectRunEvents` and its configurable-timeout + variant retain terminal history exactly, require exactly one terminal event, + then use the remaining shared deadline to require the matching terminal + `GetRun` status. Closed streams without a terminal event and contradictory + event/status pairs fail explicitly. This is test-only and does not move the + production commit or fanout boundaries. Direct phase tests bypass the settled + helper and continue observing the intentional publication window. + ## 2026-07-31 (Source-Workflow Terminal Error Arbitration) - System/component: `internal/workflow.SourceManager.runSourceWorkflow` and its diff --git a/docs/plans/2026-07-31-issue-1067-terminal-status-event-atomicity-impact-map.md b/docs/plans/2026-07-31-issue-1067-terminal-status-event-atomicity-impact-map.md new file mode 100644 index 00000000..2249a954 --- /dev/null +++ b/docs/plans/2026-07-31-issue-1067-terminal-status-event-atomicity-impact-map.md @@ -0,0 +1,172 @@ +# Cross-Surface Impact Map: Issue #1067 Terminal Publication Atomicity + +## Task + +- Task / issue: #1067, terminal status visible before matching replay event. +- Plan: `2026-07-31-issue-1067-terminal-status-event-atomicity-plan.md`. +- Owner: Codex. +- Status: exact-head durability-retention hardening implemented and verified + locally through focused stress, affected normal/race/vet, and the unchanged + full repository regression gate; hosted checks remain pending promotion. + +## Current Ownership, Callers, and Data Flow + +- Entry points: `completeRun`, `failRun`, `failRunMaxSteps`, + `failRunMaxTurns`, and `cancelledRun`. +- Source of truth: `Runner` owns `runState.run.Status`, `runState.events`, + `runState.terminated`, terminal event/status durability markers, recorder + channels, and run subscribers; + `eventJournal` owns append/store/fanout ordering. +- Callers/consumers: step-engine completion/provider/tool/budget/cancellation + paths; `GetRun`; `Subscribe`; run HTTP/SSE routes; CLI, TUI, and macOS + transcript/lifecycle consumers. +- Similar abstractions searched: `rg -n + "setStatus|completeRun|failRun|cancelledRun|publishTerminal|GetRun|Subscribe" + internal/harness internal/server`. No second terminal lifecycle owner exists. +- Duplication conclusion: repair the shared Runner transition; do not add + provider-, server-, or client-specific compensation. + +## Config, API, CLI, and Tools + +- Config/env/defaults: `MaxCompletedRetention` retains its value/default and now + also bounds the unresolved store-backed terminal durability backlog. +- Endpoints/request/response/wire formats: unchanged `Run`, `Event`, event IDs, + SSE names, payload schema, and HTTP routes. When the durability backlog is at + its cap after bounded recovery, `POST /v1/runs` and + `POST /v1/runs/{id}/continue` return HTTP 503 with + `terminal_durability_unavailable`. +- CLI/tools/integrations: no command changes; terminal polling and stream + consumers gain a stronger ordering guarantee. +- Error states: completed/failed/cancelled values and payloads are unchanged; + `TerminalDurabilityBackpressureError` is the typed degraded-admission error. + +## Persistence and Compatibility + +- Schemas/migrations/caches/generated data: none. +- Store order: matching retained terminal `AppendEvent` is bounded and precedes + recorder dispatch, conditional terminal status `UpdateRun`, in-memory status + publication, and subscriber fanout. `UpdateRun` is attempted only after + `AppendEvent` reports success. An append error leaves durable status + non-terminal; an update error can leave a durable terminal event with a + non-terminal durable run row. In either case, bounded in-memory terminal + publication/fanout completes. This is a one-way invariant, not a two-record + transaction; third-party stores must return errors without partial writes if + they need the same durable-read characterization as the built-in stores. +- Recorder: terminal JSONL remains queued after all prior events, closed once, + and drained before terminal transition returns. An explicitly suppressed + `StorageModeNone` terminal also closes and drains the recorder before status + visibility even though no terminal event is appended. +- Retention: a store-backed terminal state is prunable only after event + persistence (or explicit `StorageModeNone` suppression) and final status + persistence are both acknowledged. No-store states remain process-local and + outside durable pruning/backpressure. +- Compatibility: event/status values and replay IDs remain stable. The new 503 + is an intentional fail-closed availability change during persistence outage. +- Mixed-version behavior: process-local; older daemons retain the race until + upgraded, with no data migration. + +## Lifecycle, Security, and Reliability + +- Concurrency: the winning terminal event seals the ledger before status is + updated; competing terminal helpers cannot overwrite it with a mismatched + status. Every status snapshot/persist/commit sequence shares a per-run mutex, + so a delayed running/waiting write cannot overwrite terminal state. A + validated Continue source requires a temporary in-state reservation across + unlocked durability recovery so a concurrent Start prune cannot remove it + before the existing single-winner continuation commit. +- Cancellation/retries/cleanup: cooperative cancellation and idempotency stay + unchanged; workspace/tool/MCP cleanup remains before terminal publication. + Status-only durability gaps retry safely as idempotent `UpdateRun` overwrites + under one total deadline of at most 250 ms. Ambiguous failed event appends are + counted but not retried because a third-party store may have applied them. +- Locks/resources: terminal store/recorder waits remain outside `Runner.mu` and + the global conversation journal lock. A refcounted target-conversation lock + prevents same-conversation overtaking while unrelated journals progress, and + deletes itself when owners and queued waiters drain. +- Auth/permissions/privacy/secrets: no boundary change after searches through + run routes and redaction/audit paths; terminal payload redaction remains + owned by the event journal. Explicit terminal `StorageModeNone` remains the + documented exception: it seals and publishes status without replaying the + intentionally suppressed event. +- Failure/recovery: append and status writes use bounded contexts and failures + remain non-fatal to already-admitted work. Both failure classes count toward + the same finite admission gate. At the configured cap, Start/Continue retries + recoverable status gaps without holding Runner/status/conversation locks, + then rejects admission if the backlog remains full. Already-admitted work may + temporarily exceed the numeric cap but the excess is bounded by the finite + active/queued population admitted before outage detection. No two-way durable + atomicity is claimed without a transactional store API. A successful status + retry prunes newly durable candidates immediately, restoring the configured + completed-retention bound before admission reopens. + +## Product and Integration Surfaces + +- Server/runtime: `GetRun` terminal now implies immediate `Subscribe` replay + contains the matching event. Start/Continue expose the explicit degraded 503; + Continue preserves not-found/non-completed source error precedence. +- TUI/web/macOS/other clients: terminal badges, failure text, exit codes, and + transcript state no longer disagree during the publication window; no client + code changes. +- Provider/model/tool catalogs/routing: none; provider failure is only a caller. +- External systems/automation: internal StartRun callers inherit the typed + fail-closed error through their existing error paths; cron/callback/workflow + request schemas and successful semantics are unchanged. +- UX/accessibility/focus/motion: no visual or interaction change. + +## Deployment and Operations + +- Deployment/migrations/flags: ordinary daemon rollout; no migration or flag. +- Observability: deterministic regression records transition phases without + logging prompts, event payload secrets, or credentials. Operators can + correlate store append/update errors with 503 + `terminal_durability_unavailable`; successful status retry reopens admission. +- Rollback: revert if healthy stores spuriously return 503, the total retry + exceeds 250 ms, source error precedence changes, terminal fanout deadlocks, + or unrelated Runner/conversation work blocks. During a real store outage, + stop new producers and preserve the process before rollback because removing + the gate reintroduces truthful-state eviction or unbounded protected memory. +- Runbooks/operator docs: no command changes; the issue and implementation logs + carry the degraded-mode recovery policy. + +## Regression Tests + +- First red: phase barrier proves old completed/failed/cancelled status can win + before terminal history. +- Acceptance: terminal status implies one matching replay event; failed causal + snapshot precedes `run.failed`; competing terminal transitions match the + winning sealed event; later same-conversation events cannot overtake terminal + fanout to an existing conversation subscriber. +- Store/recorder: blocked append allows unrelated conversations but not target + overtaking; append error prevents terminal status persistence; status update + error/timeout preserves live publication; retained and suppressed terminal + recorder paths drain before status visibility. +- Retention/admission: several already-admitted completions remain visible at + retention 1 when final status updates fail and their durable rows remain + non-terminal; append- and status-pending runs both close admission at the cap; + concurrent callers reject during outage and recover after status persistence; + the retry uses one unlocked deadline; concurrent Start recovery cannot prune + a reserved Continue source; reservation cleanup and single-winner behavior + remain bounded; StorageModeNone and no-store policies remain explicit. +- Integration: HTTP poll immediately followed by run SSE replay for all three + statuses. +- Test helpers: terminal event/history assertions remain unchanged; shared + settled-run collectors require exactly one terminal event and independently + await its matching status under their existing total deadline. Stream close + without terminal history and event/status mismatch are explicit failures. + Direct transition-window tests keep raw Subscribe and phase barriers; their + non-return proof uses an explicit settlement-entry handshake. +- Exact gates: focused normal/race stress `-count=100`; harness/server + normal/race/vet; unchanged foreground non-TTY regression; hosted checks. + +## Documentation and Handoff + +- Plans/specs: issue-specific plan and this map. +- Logs/indexes: engineering, observational, system, long-term, plans index, and + active plan. +- Public/training/release docs: none because no new route, schema, or command is + introduced. + +## Warning Check + +- Every cross-surface heading is resolved. Unaffected surfaces are explicitly + named with search and data-flow rationale above. diff --git a/docs/plans/2026-07-31-issue-1067-terminal-status-event-atomicity-plan.md b/docs/plans/2026-07-31-issue-1067-terminal-status-event-atomicity-plan.md new file mode 100644 index 00000000..c5a7b16d --- /dev/null +++ b/docs/plans/2026-07-31-issue-1067-terminal-status-event-atomicity-plan.md @@ -0,0 +1,247 @@ +# Plan: Make Terminal Status and Event Publication Atomic + +## Context + +- Governing GitHub issue: #1067. +- Problem: terminal helpers publish `Run.Status` before the event journal has + prepared, persisted, and exposed the matching terminal event. A concurrent + `GetRun` can therefore return completed, failed, or cancelled while an + immediate `Subscribe` replay still ends at a non-terminal event. +- User impact: API polling, SSE reconnect, CLI/TUI, and macOS clients can + briefly render terminal state without the authoritative terminal transcript + or its preceding causal/error evidence. +- Constraints: preserve terminal sealing, recorder drain ordering, bounded + store writes, run-independent availability, current event/status schemas, + cleanup ordering, and the explicit `StorageModeNone` terminal-redaction + policy. The store API has no cross-record transaction, so promise the + testable one-way invariant rather than two-way event/status atomicity. +- Exact-head review finding: terminal event success was tracked for pruning but + terminal status update success was discarded. A failed `UpdateRun` could + therefore make pruning evict truthful live state while the durable row stayed + non-terminal; permanent append/update failures could also grow protected + memory without an admission bound. +- Exact-head concurrent review finding: Continue validated its completed source, + released `Runner.mu`, and performed bounded durability recovery before its + single-winner mutation. A concurrent Start recovery could prune that source + through the shared prune path, turning a valid continuation into a spurious + `ErrRunNotFound`. +- Hosted aggregate review finding: race run `30656467482` exposed another stale + test assumption in `TestRunnerHookErrorFailOpen`: `collectRunEvents` returned + terminal replay history while the valid later completed-status commit was + still in progress. The helper's 215 references and its configurable-timeout + equivalent were audited; no caller intentionally observes that window. +- Exact-head helper review finding: commit `8757e8a3` accepted any terminal + status after collection, even if stream closure supplied no terminal event or + the collected terminal event contradicted status. Its phase regression also + used a timed non-return window without proving the goroutine entered + settlement first. + +## Scope + +- In scope: one shared Runner terminal-transition seam for completed, failed, + cancelled, max-step failed, and max-turn failed paths; deterministic + concurrency and replay regressions; event/status-aware safe pruning; bounded + degraded admission and recovery; explicit Start/Continue HTTP 503 mapping; + real HTTP poll-then-replay proof; test-helper settlement semantics that keep + terminal event collection and terminal status waiting distinct. +- Out of scope: PR #1060/#1055 changes, cron/callback behavior, conversation + cursor redesign, GUI visual changes, provider routing, schemas, and workflow + timing issue #1049. + +## Documentation Contract + +- Feature status: durability-retention hardening remains implemented; the + exact-head settled-helper validation repair passes focused normal/race stress, + hosted-equivalent race, and the full repository regression gate. +- Public API behavior: event/status wire formats stay unchanged. During a full + terminal durability backlog, Start/Continue now return documented HTTP 503 + `terminal_durability_unavailable` after bounded recovery fails. +- Spec docs before code: this plan and its linked impact map. +- Implementation notes after code: engineering, observational, system, and + long-term logs plus the plans index and active plan. + +## Test Plan (TDD) + +- First red: a deterministic phase barrier pauses each completed, failed, and + cancelled helper after the old status write but before terminal event + publication. Concurrent `GetRun` plus `Subscribe` must never observe that + forbidden state. +- Causal control: on an error-chain-enabled failure, the required + `error.context` snapshot must precede `run.failed` before failed status is + observable. +- Store/recorder controls: block terminal append and prove unrelated + conversations progress while later events in the target conversation cannot + overtake. On append error, never attempt a durable terminal status update but + still complete bounded in-memory terminal publication/fanout. On status + update error or context timeout after a successful append, keep the durable + run non-terminal while the live Runner and subscribers complete. For an + explicit `StorageModeNone` drop, drain the recorder before status visibility. +- Retention/admission controls: with retention 1, pre-admit several runs and + fail terminal `UpdateRun`; every truthful terminal state must remain in + memory while durable rows remain non-terminal. Count both append- and + status-pending states toward one gate; reject concurrent Start admissions and + Continue admission with a typed error/HTTP 503 at the cap; recover concurrent + callers after status persistence returns; preserve source error precedence, + intentional StorageModeNone suppression, and no-store behavior. Pin one + shared retry deadline of at most 250 ms with no store I/O under Runner, + status, event-journal, or conversation locks. +- Continuation reservation control: deterministically block Continue's first + recovery write after source validation, let a concurrent Start recover and + prune, then require Continue to retain its source through the single-winner + handoff. The reservation must apply to every completed-run prune caller and + release on all success/error paths without holding locks across store I/O. +- Settled collection control: pause terminal publication after replay history is + visible but before status commit, call the shared event collector, and prove + it reaches settlement and does not return until the matching terminal status + is independently observable. Reject closed streams without a terminal event + and terminal event/status mismatch. Keep one total bounded deadline and + retain the exact collected event assertions. +- Concurrency control: race competing terminal transitions and require the + winning status to match the single sealed terminal event; hold a terminal at + the pre-fanout boundary and prove a later same-conversation event cannot + overtake it for an existing conversation subscriber. +- Real path: HTTP `GET /v1/runs/{id}` followed immediately by run-event SSE + replay contains the matching terminal event for all three statuses. +- Focused stress: normal and race at `-count=100`. +- Affected gates: `internal/harness` and `internal/server` normal/race and vet. +- Repository gate: unchanged foreground non-TTY + `./scripts/test-regression.sh`. +- Hosted gates: required PR checks, including `test-fast` and `test-race`. + +## Cross-Surface Impact Map + +- See `2026-07-31-issue-1067-terminal-status-event-atomicity-impact-map.md`. + +## Implementation Checklist + +- [x] Link contract-complete bug #1067 before implementation. +- [x] Record current ownership, callers, sources of truth, and search evidence. +- [x] Write this plan and impact map before code. +- [x] Add and confirm the deterministic failing regressions. +- [x] Implement the smallest shared terminal-transition repair. +- [x] Confirm focused stress and affected normal/race/vet gates. +- [x] Confirm the unchanged repository regression gate on the final diff. +- [x] Prove the HTTP poll-then-replay path. +- [x] Update all required logs and documentation status. +- [x] Track terminal status durability separately and require event plus status + resolution before store-backed pruning. +- [x] Add finite append/status backlog admission, bounded status recovery, typed + errors, and Start/Continue HTTP 503 mappings. +- [x] Prove concurrent outage/recovery, unlocked deadline, StorageModeNone, and + no-store behavior. +- [ ] Open one closing PR, push its exact head, and request `@codex` review. +- [ ] Confirm hosted checks are green; do not merge. + +## Risks and Mitigations + +- Risk: holding `Runner.mu` across persistence would block unrelated queries. +- Mitigation: retain out-of-lock store/recorder I/O, serialize only the target + conversation across the terminal sequence, and test unrelated `GetRun` and + unrelated conversation-journal responsiveness while persistence is blocked. +- Risk: concurrent terminal helpers could publish one event and a different + status. +- Mitigation: serialize each run's complete terminal helper lifecycle, make the + shared transition return whether it won terminal sealing, and update status + only for that winner. +- Risk: moving status later could reorder cleanup, causal snapshots, audit, + profile persistence, backup, or pruning. +- Mitigation: pin causal/event order and keep cleanup before terminal + transition, operational side effects after the matching status/event pair. +- Risk: the store API cannot atomically append an event and update the run row. +- Mitigation: enforce the one-way invariant that retained terminal status is + never attempted unless terminal `AppendEvent` reported success. If append + fails, durable status stays non-terminal; if final `UpdateRun` fails after a + successful append, the durable event may lead the durable status. Both + failures remain bounded and non-fatal to in-memory status/fanout. This does + not claim two-way transactional atomicity or infer whether a third-party store + applied a write before returning an error. +- Risk: recorder or store delay could weaken availability or lifecycle order. +- Mitigation: use context-bounded terminal store calls, keep external I/O out + of the global conversation mutex, preserve target-conversation sequencing, + drain retained and suppressed terminal recorders, and reclaim idle keyed + sequence locks. +- Risk: protecting event/status-pending terminal states from pruning could make + a permanent store outage consume memory without bound. +- Mitigation: both failure classes consume the `MaxCompletedRetention` backlog. + At the cap, new admissions retry status-only gaps under one shared deadline + and otherwise fail closed. Already-admitted work may finish and remain visible, + but growth stops at the finite population admitted before outage detection. + Ambiguous append failures are never replayed in-process. +- Risk: degraded admission could mask normal Continue errors or block unrelated + state while retrying. +- Mitigation: validate unknown/non-completed sources before the global gate, + revalidate for the single-winner mutation, map only the typed error to 503, + and prove the retry holds no Runner/status/conversation lock. +- Risk: validation followed by unlocked recovery lets another admission's prune + delete the continuation source before revalidation. +- Mitigation: reserve the validated source under `Runner.mu`, exclude reserved + sources in the shared prune implementation used by every caller, and release + plus re-prune on every return path. The reservation does not grant the + continuation winner status; the existing write-lock revalidation still does. +- Risk: an explicit terminal `StorageModeNone` policy intentionally removes the + matching event from replay. +- Mitigation: preserve and test the existing redaction exception and scope the + stronger replay implication to terminal events retained by policy. + +## Verification Evidence + +- Semantic red: + `go test ./internal/harness -run + '^TestTerminalStatusNeverPrecedesTerminalReplayEvent$' -count=1` failed all + three cases: completed before `run.completed`, failed after `error.context` + but before `run.failed`, and cancelled before `run.cancelled`. +- Review reds: blocked terminal append serialized an unrelated conversation; + append failure still persisted terminal status; delayed non-terminal status + overwrote terminal status; context-blocking final status persistence stranded + the transition; an explicit terminal redaction drop exposed status before + recorder drain; and keyed sequence entries were never reclaimed. +- Exact-head P1 reds: at retention 1, append-pending admission remained open; + status-update-failed terminal runs were pruned despite non-terminal durable + rows; and both Start/Continue mapped the new typed degraded state to 400. +- Exact-head concurrent P1 red: a phase hook paused Continue after source + validation, concurrent Start recovery pruned that oldest source at retention + 1, and resumed Continue failed deterministically with `run not found`. +- Exact-head focused green: append/status pending runs remain visible, both + close admission at the cap, 16 concurrent callers reject then recover under + race, successful status recovery immediately restores the retention bound, + status recovery uses one unlocked total deadline, Start/Continue return the + explicit 503, and StorageModeNone/no-store controls pass. +- Concurrent focused green: the source reservation is respected by the shared + prune path, releases after backpressure, and preserves exactly one continuation + winner. The new regression, cleanup control, and existing winner control pass + normal/race at `-count=100`. +- Focused current green: all terminal publication/failure-policy regressions + and HTTP replay passed normal and race at `-count=100`. +- Affected current green: complete harness/server normal and race passed; + affected `go vet` passed. +- Real path: HTTP terminal polling followed immediately by Last-Event-ID run + SSE replay passed for completed, failed, and cancelled. +- Gate-test review: the complete race gate exposed a replay-first test that read + status before the later commit. Its independent failed-status wait now matches + the one-way contract, and the affected race gate passes. +- Hosted settled-helper red: race run `30656467482` reproduced the next stale + immediate-status assumption in `TestRunnerHookErrorFailOpen`. A deterministic + pre-status barrier then proved the shared collector returned terminal history + while status remained running. +- Helper audit/green: all 215 shared collector references, the configurable + timeout variant, and the separate snapshot helper family were classified. + No shared collector consumer requires the transition window. Settlement now + preserves event history and independently requires terminal status within one + total deadline; the deterministic test, hook family, and timeout caller pass + normal/race at `-count=100`, and hosted-equivalent `make test-race` passes. +- Exact-head review reds: a completed run plus closed stream containing only + `run.started` returned success, and failed status plus `run.completed` + returned success. The phase regression's prior timed non-return check did not + prove its collector goroutine had entered settlement. +- Exact-head review green: the subscribed-stream core requires exactly one + terminal event and matches its meaning to status without changing the + collected slice or original total deadline. The phase test uses an explicit + settlement-entry signal. Both regressions, the phase test, hook family, and + configurable-timeout caller pass normal/race at `-count=100`; + hosted-equivalent `make test-race` and the authoritative full repository gate + pass on this follow-up diff. +- Repository: the final direct foreground non-TTY + `TMPDIR=/private/tmp GOCACHE=/private/tmp/gocode-go-cache + ./scripts/test-regression.sh` passed normal, full race, and coverage with + `coveragegate: PASS (total=85.7%, min=80.0%, zero-functions=0)` on the + settled-helper diff. diff --git a/docs/plans/INDEX.md b/docs/plans/INDEX.md index 8c8d2b9a..7abf5c9c 100644 --- a/docs/plans/INDEX.md +++ b/docs/plans/INDEX.md @@ -6,6 +6,8 @@ - `2026-07-31-issue-1076-workflow-initial-write-exit-impact-map.md` — Cross-surface impact map for Issue #1076 lifecycle and error arbitration. - `2026-07-31-issue-1068-dispatcher-shutdown-isolation-plan.md` — Issue #1068 plan for deterministic instance-scoped Runner dispatcher shutdown verification. - `2026-07-31-issue-1068-dispatcher-shutdown-isolation-impact-map.md` — Cross-surface impact map for Issue #1068 lifecycle isolation. +- `2026-07-31-issue-1067-terminal-status-event-atomicity-plan.md` — Issue #1067 plan for linearizable terminal publication, safe durability-aware pruning, and bounded degraded admission. +- `2026-07-31-issue-1067-terminal-status-event-atomicity-impact-map.md` — Cross-surface impact map for Issue #1067 Runner lifecycle, persistence, retention, and 503 recovery behavior. - `2026-07-31-issue-1062-provider-key-matrix-health-wait-plan.md` — Issue #1062 plan for a contention-tolerant provider API-key matrix startup wait. - `2026-07-31-issue-1062-provider-key-matrix-health-wait-impact-map.md` — Cross-surface impact map for Issue #1062. - `2026-07-31-issue-1064-workflow-exit-precedence-plan.md` — Issue #1064 deterministic source-workflow process-exit diagnostic precedence repair. diff --git a/docs/plans/active-plan.md b/docs/plans/active-plan.md index a5bf68b1..0e741c90 100644 --- a/docs/plans/active-plan.md +++ b/docs/plans/active-plan.md @@ -1,5 +1,28 @@ # Active Plan +Current status: Issue #1067 terminal publication atomicity is implemented on +isolated branch `codex/issue-1067-terminal-status-event-atomicity`. Post-review +hardening now defines and tests the one-way durable failure contract, bounded +store waits, per-run status serialization, per-conversation lock reclamation, +recorder drain for suppressed terminal events, event-plus-status safe pruning, +finite fail-closed admission during terminal persistence outage, and a +prune-wide reservation for validated continuation sources. The concurrent red, +durability-retention focused suite, affected-package normal/race/vet, and full +foreground repository gate pass at 85.7% coverage with zero uncovered +production functions on the prior follow-up. PR #1070 reached rebased head +`5f106bef`, where hosted race run `30656467482` exposed a stale settled-test +assumption; the audited shared event +collector now independently awaits terminal status under the same deadline. +Affected normal/race x100, hosted-equivalent `make test-race`, and the final +outside-sandbox foreground repository gate pass at 85.7% coverage with zero +uncovered production functions. Exact-head review of local commit `8757e8a3` +then found the settled helper accepted missing or status-mismatched terminal +events and its phase test lacked an entry handshake. The test-only correction +requires exactly one matching terminal event and passes focused normal/race +x100, hosted-equivalent race, and the full repository gate at 85.7% coverage +with zero uncovered production functions. The repair remains local and +unpushed; PR #1060 remains excluded and no merge is authorized. + Current status: Issue #1076 isolates the source-workflow lifecycle race where a child exits before the initial `start` write and the early EPIPE return skips wait plus bounded stderr arbitration. Natural-exit and live-child cleanup reds @@ -43,6 +66,7 @@ Remaining work before merge is final verification and any requested review/cleanup. Current active plans: +- `2026-07-31-issue-1067-terminal-status-event-atomicity-plan.md` - `2026-07-31-issue-1076-workflow-initial-write-exit-plan.md` - `2026-07-31-issue-1068-dispatcher-shutdown-isolation-plan.md` - `2026-07-30-issue-1052-provider-key-capture-sync-plan.md` diff --git a/internal/harness/job_bridge.go b/internal/harness/job_bridge.go index 1067ed8a..3c6199f1 100644 --- a/internal/harness/job_bridge.go +++ b/internal/harness/job_bridge.go @@ -184,6 +184,8 @@ func (r *Runner) emitToConversation(convID, originRunID string, eventType EventT if convID == "" { return } + unlockSequence := r.lockConversationSequence(convID) + defer unlockSequence() r.conversationEventMu.Lock() defer r.conversationEventMu.Unlock() diff --git a/internal/harness/runner.go b/internal/harness/runner.go index 734f2b49..0f814c02 100644 --- a/internal/harness/runner.go +++ b/internal/harness/runner.go @@ -36,7 +36,6 @@ import ( type runState struct { run Run - statusPersist contextMutex planMode PlanModeState planFile string staticSystemPrompt string @@ -98,16 +97,29 @@ type runState struct { // continued is set to true once ContinueRun has been called on this run, // preventing a second continuation without mutating the run's terminal Status. continued bool + // continuationReservations protects a validated completed source from every + // completed-run prune path while ContinueRun performs unlocked durability + // recovery and its later single-winner mutation. It is guarded by Runner.mu. + continuationReservations int // snapshotBuilder collects a rolling window of tool calls and messages for // error context snapshots. Non-nil only when ErrorChainEnabled is set in // RunnerConfig. snapshotBuilder *errorchain.SnapshotBuilder - // terminated is set to true once the terminal event (run.completed or - // run.failed) has been emitted. Any subsequent emit() call returns - // immediately to prevent post-terminal streaming callbacks from appending - // events after the forensic record is closed. - terminated bool - terminalEventPersisted bool + // terminated is set to true once a terminal event (run.completed, + // run.failed, or run.cancelled) has been emitted. Any subsequent emit() call + // returns immediately to prevent post-terminal streaming callbacks from + // appending events after the forensic record is closed. + terminated bool + terminalEventPersisted bool + terminalEventSuppressed bool + terminalStatusPersisted bool + // terminalMu serializes the complete terminal-helper lifecycle so only the + // event winner may run terminal audit/profile/cleanup side effects. + terminalMu sync.Mutex + // statusMu serializes every status snapshot/persist/commit sequence with the + // terminal transition. A delayed waiting/running write therefore cannot + // overwrite a terminal result. + statusMu contextMutex // compactMu serializes auto-compact and manual CompactRun calls. compactMu sync.RWMutex // resetIndex increments each time the agent calls reset_context. @@ -180,6 +192,27 @@ func (m *contextMutex) unlock() { m.token <- struct{}{} } +func (m *contextMutex) Lock() { + _ = m.lock(context.Background()) +} + +func (m *contextMutex) Unlock() { + m.unlock() +} + +func (m *contextMutex) TryLock() bool { + m.once.Do(func() { + m.token = make(chan struct{}, 1) + m.token <- struct{}{} + }) + select { + case <-m.token: + return true + default: + return false + } +} + // Runner concurrency/lifecycle invariants // // Event ledger: @@ -189,6 +222,9 @@ func (m *contextMutex) unlock() { // mirror that must drain exactly that ledger before a terminal emit returns. // 3. state.terminated is armed before terminal redaction/fanout so no // post-terminal goroutine can append to the sealed forensic record. +// 4. For terminal events retained by redaction policy, status is published +// only after the winning event reaches replay/store/recorder history; +// GetRun therefore cannot expose terminal state ahead of replay. // // Message lifecycle: // 1. state.messages is the only source of truth for run context. @@ -240,6 +276,24 @@ var ( ErrRunnerClosed = errors.New("runner is closed") ) +// TerminalDurabilityBackpressureError is returned by StartRun and ContinueRun +// when terminal runs whose event or final status has not been durably +// acknowledged reach MaxCompletedRetention. New admissions fail closed so +// already-admitted runs can finish without allowing an unbounded in-memory +// durability backlog. +type TerminalDurabilityBackpressureError struct { + Pending int + Limit int +} + +func (e *TerminalDurabilityBackpressureError) Error() string { + return fmt.Sprintf( + "terminal durability backlog reached retention limit: %d pending, limit %d", + e.Pending, + e.Limit, + ) +} + // steeringBufferSize is the capacity of the per-run steering message channel. const steeringBufferSize = 10 @@ -259,7 +313,10 @@ const recorderDrainTimeout = 30 * time.Second // the model returns 0 completion_tokens with empty content. const maxEmptyRetries = 3 -const terminalEventStoreTimeout = 5 * time.Second +const ( + terminalEventStoreTimeout = 5 * time.Second + terminalDurabilityAdmissionMaxDuration = 250 * time.Millisecond +) const ( defaultMaxCompletedRetention = 32 @@ -347,6 +404,12 @@ type Runner struct { // boundary with no event prepared before registration and persisted after // the snapshot. conversationEventMu sync.Mutex + // conversationSequenceMu guards the per-conversation publication locks. + // A terminal transition may release conversationEventMu around status-store + // I/O while retaining its conversation lock, so unrelated conversations can + // progress without allowing same-conversation replay/live delivery to pass it. + conversationSequenceMu sync.Mutex + conversationSequence map[string]*conversationSequenceLock // conversationEvents is the bounded no-run-store replay fallback. Built-in // stores provide durable replay; third-party/no-store configurations retain // this process-local window instead. @@ -370,6 +433,21 @@ type Runner struct { // immediately before it marks dispatcherWG done. It lets lifecycle tests // identify one Runner without scanning process-global goroutine stacks. poolDispatcherExitHook func() + // terminalTransitionHook is a test seam invoked at the publication boundary + // between terminal lifecycle preparation and terminal event emission. + terminalTransitionHook func(string, RunStatus, EventType) + // terminalBeforeDispatchHook is a test seam invoked after terminal replay + // persistence and before recorder/status/subscriber publication. + terminalBeforeDispatchHook func(string, EventType) + // statusBeforeCommitHook is a test seam invoked after a non-terminal status + // snapshot is prepared and before it is committed. + statusBeforeCommitHook func(string, RunStatus) + // continuationAfterValidationHook is a test seam invoked after ContinueRun + // validates its source and releases Runner.mu, before durability recovery. + continuationAfterValidationHook func(string) + // terminalStoreTimeout overrides the bounded terminal store timeout in + // deterministic tests. Zero uses terminalEventStoreTimeout. + terminalStoreTimeout time.Duration // runQueue is a FIFO channel of pending (runID, req) pairs waiting for a // worker slot. It is only used when workerSem is non-nil. runQueue chan queuedRun @@ -542,29 +620,136 @@ type retainedRunCandidate struct { updatedAt time.Time } +type terminalStatusRetry struct { + run Run +} + +func completedRetentionLimit(rc RunnerConfig) int { + if rc.MaxCompletedRetention > 0 { + return rc.MaxCompletedRetention + } + return defaultMaxCompletedRetention +} + +func runStateHasPersistentStore(state *runState, fallback RunnerConfig) bool { + if state != nil && state.config != nil { + return state.config.Store != nil + } + return fallback.Store != nil +} + +func terminalEventDurabilityResolved(state *runState) bool { + return state != nil && (state.terminalEventPersisted || state.terminalEventSuppressed) +} + +func terminalDurabilityComplete(state *runState) bool { + return terminalEventDurabilityResolved(state) && state.terminalStatusPersisted +} + +// terminalDurabilityBacklog snapshots the number of store-backed terminal runs +// which cannot yet be safely evicted. Status-only gaps are safe to retry with +// the same terminal Run snapshot because UpdateRun is an idempotent overwrite. +// Unacknowledged event appends are counted for backpressure but are not retried: +// a third-party store may have applied an append before returning an error, so +// retrying could duplicate the forensic event. +func (r *Runner) terminalDurabilityBacklog(rc RunnerConfig) (int, []terminalStatusRetry) { + r.mu.RLock() + defer r.mu.RUnlock() + + pending := 0 + retries := make([]terminalStatusRetry, 0) + for _, state := range r.runs { + if state == nil || !isTerminalRunStatus(state.run.Status) || !runStateHasPersistentStore(state, rc) { + continue + } + if terminalDurabilityComplete(state) { + continue + } + pending++ + if terminalEventDurabilityResolved(state) && !state.terminalStatusPersisted { + retries = append(retries, terminalStatusRetry{run: state.run}) + } + } + return pending, retries +} + +// ensureTerminalDurabilityCapacity retries recoverable status-only gaps under +// one small total deadline, then rejects admission if the unresolved backlog +// still reaches the retention cap. Store I/O occurs after terminalDurabilityBacklog +// releases Runner.mu and without status, conversation, or journal locks. +func (r *Runner) ensureTerminalDurabilityCapacity(preserveRunID string) error { + rc := r.snapshotConfig() + if rc.Store == nil { + return nil + } + limit := completedRetentionLimit(rc) + pending, retries := r.terminalDurabilityBacklog(rc) + if pending < limit { + return nil + } + + recovered := false + if len(retries) > 0 { + timeout := r.terminalStoreTimeoutDuration() + if timeout > terminalDurabilityAdmissionMaxDuration { + timeout = terminalDurabilityAdmissionMaxDuration + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + for _, retry := range retries { + if ctx.Err() != nil { + break + } + if r.storeUpdateRunSnapshotContext(ctx, retry.run) { + r.markTerminalStatusPersisted(retry.run) + recovered = true + } + } + cancel() + } + if recovered { + // Recovery turns protected terminal states into safe durable candidates. + // Restore the retention window immediately, without store I/O under the + // lock. ContinueRun preserves its source while still counting it toward + // the quota so the source survives long enough for the revalidated handoff. + r.pruneCompletedRunsPreserving(preserveRunID) + } + + pending, _ = r.terminalDurabilityBacklog(rc) + if pending >= limit { + return &TerminalDurabilityBackpressureError{Pending: pending, Limit: limit} + } + return nil +} + func (r *Runner) pruneCompletedRuns() { + r.pruneCompletedRunsPreserving("") +} + +func (r *Runner) pruneCompletedRunsPreserving(preserveRunID string) { r.mu.Lock() defer r.mu.Unlock() - r.pruneCompletedRunsLocked() + r.pruneCompletedRunsLockedPreserving(preserveRunID) } func (r *Runner) pruneCompletedRunsLocked() { + r.pruneCompletedRunsLockedPreserving("") +} + +func (r *Runner) pruneCompletedRunsLockedPreserving(preserveRunID string) { rc := r.snapshotConfig() if rc.Store == nil { return } - limit := rc.MaxCompletedRetention - if limit <= 0 { - limit = defaultMaxCompletedRetention - } + limit := completedRetentionLimit(rc) candidates := make([]retainedRunCandidate, 0) for runID, state := range r.runs { - if state == nil || !isTerminalRunStatus(state.run.Status) || !state.terminalEventPersisted { + if state == nil || !isTerminalRunStatus(state.run.Status) || + !runStateHasPersistentStore(state, rc) || !terminalDurabilityComplete(state) { continue } - if len(state.subscribers) == 0 { + if len(state.subscribers) == 0 && state.continuationReservations == 0 { candidates = append(candidates, retainedRunCandidate{ id: runID, updatedAt: state.run.UpdatedAt, @@ -586,8 +771,15 @@ func (r *Runner) pruneCompletedRunsLocked() { }) toDelete := len(candidates) - limit - for i := 0; i < toDelete; i++ { - delete(r.runs, candidates[i].id) + for _, candidate := range candidates { + if toDelete == 0 { + break + } + if candidate.id == preserveRunID { + continue + } + delete(r.runs, candidate.id) + toDelete-- } } @@ -1011,6 +1203,13 @@ func (r *Runner) StartRun(req RunRequest) (Run, error) { } } + // A terminal persistence outage must not turn the retention exceptions into + // unbounded memory growth. This check runs after request validation and + // ownership checks but before recorder creation or run-state admission. + if err := r.ensureTerminalDurabilityCapacity(""); err != nil { + return Run{}, err + } + // Create rollout recorder before acquiring the run lock so that any // filesystem error is surfaced at start time rather than mid-run. var rec *rollout.Recorder @@ -1590,10 +1789,12 @@ func (r *Runner) GetRun(runID string) (Run, bool) { // transcript. // // Errors: -// - ErrRunNotFound — the source run does not exist. -// - ErrRunNotCompleted — the source run has not reached RunStatusCompleted +// - ErrRunNotFound — the source run does not exist. +// - ErrRunNotCompleted — the source run has not reached RunStatusCompleted // (it is still running, queued, waiting for user, or has failed). -// - validation error — message is empty. +// - TerminalDurabilityBackpressureError — unresolved terminal persistence +// reached the configured in-memory retention cap. +// - validation error — message is empty. // // The method is safe for concurrent use. Only one goroutine can successfully // continue a given completed run: the first to acquire the lock transitions @@ -1629,6 +1830,23 @@ func (r *Runner) ContinueRunWithOptions(runID string, req ContinueRunRequest) (R } } + // Preserve source error precedence while degraded and reserve the validated + // source before releasing Runner.mu. The reservation is not the continuation + // winner decision: concurrent continuations still revalidate and compete on + // state.continued under the write lock below. It only prevents every shared + // prune path from deleting the source during unlocked durability recovery. + if err := r.reserveContinuationSource(runID); err != nil { + return Run{}, err + } + defer r.releaseContinuationSource(runID) + if r.continuationAfterValidationHook != nil { + r.continuationAfterValidationHook(runID) + } + + if err := r.ensureTerminalDurabilityCapacity(runID); err != nil { + return Run{}, err + } + // Atomically check that the run exists and is completed, then immediately // stamp it with RunStatusRunning to prevent any other goroutine from also // starting a continuation. All snapshot values are read under the same @@ -1808,6 +2026,39 @@ func (r *Runner) ContinueRunWithOptions(runID string, req ContinueRunRequest) (R return newRun, nil } +func (r *Runner) reserveContinuationSource(runID string) error { + r.mu.Lock() + defer r.mu.Unlock() + + state, ok := r.runs[runID] + if !ok { + return ErrRunNotFound + } + if state.run.Status != RunStatusCompleted { + return ErrRunNotCompleted + } + if state.continued { + return fmt.Errorf("run %q has already been continued", runID) + } + state.continuationReservations++ + return nil +} + +func (r *Runner) releaseContinuationSource(runID string) { + r.mu.Lock() + defer r.mu.Unlock() + + state := r.runs[runID] + if state == nil || state.continuationReservations == 0 { + return + } + state.continuationReservations-- + // A reservation is a temporary pruning exception. Re-run the shared prune + // policy immediately on release so success, backpressure, validation races, + // and dispatch failure cannot leave the retention window inflated. + r.pruneCompletedRunsLocked() +} + // GetRunSummary computes a telemetry summary for a completed (or failed) run // by scanning the run's event history. Returns ErrRunNotFound if the run does // not exist, or an error if the run is still in progress. @@ -2033,6 +2284,8 @@ func (r *Runner) SubscribeConversationFrom( return nil, nil, nil, ConversationReplayInfo{}, fmt.Errorf("conversation %q not found", convID) } + unlockSequence := r.lockConversationSequence(convID) + defer unlockSequence() r.conversationEventMu.Lock() defer r.conversationEventMu.Unlock() @@ -2066,6 +2319,42 @@ func (r *Runner) SubscribeConversationFrom( return history, ch, cancel, replay, nil } +func (r *Runner) lockConversationSequence(convID string) func() { + key := strings.TrimSpace(convID) + if key == "" { + key = "__no_conversation__" + } + r.conversationSequenceMu.Lock() + if r.conversationSequence == nil { + r.conversationSequence = make(map[string]*conversationSequenceLock) + } + sequence := r.conversationSequence[key] + if sequence == nil { + sequence = &conversationSequenceLock{} + r.conversationSequence[key] = sequence + } + sequence.refs++ + r.conversationSequenceMu.Unlock() + sequence.mu.Lock() + var unlockOnce sync.Once + return func() { + unlockOnce.Do(func() { + sequence.mu.Unlock() + r.conversationSequenceMu.Lock() + sequence.refs-- + if sequence.refs == 0 && r.conversationSequence[key] == sequence { + delete(r.conversationSequence, key) + } + r.conversationSequenceMu.Unlock() + }) + } +} + +type conversationSequenceLock struct { + mu sync.Mutex + refs int +} + func (r *Runner) conversationReplay( convID, tenantID, lastEventID string, ) ([]Event, ConversationReplayInfo) { @@ -3237,6 +3526,12 @@ func (r *Runner) drainSteering(runID string, messages *[]Message) { } func (r *Runner) completeRun(runID, output string) { + unlockTerminal, ok := r.lockTerminalTransition(runID) + if !ok { + return + } + defer unlockTerminal() + rc := r.configForRun(runID) // Clean up per-run workspace before terminal event (issue #324). r.runWorkspaceCleanup(runID) @@ -3316,8 +3611,6 @@ func (r *Runner) completeRun(runID, output string) { r.closeAuditWriter(runID) } - r.setStatus(runID, RunStatusCompleted, output, "") - usageTotals, costTotals := r.accountingTotals(runID) // Efficiency suggestion: if the run used a named profile and the @@ -3328,7 +3621,7 @@ func (r *Runner) completeRun(runID, output string) { // Profile run history: persist completion record for analysis. r.persistProfileRun(runID, "completed", costTotals.CostUSDTotal) - r.emit(runID, EventRunCompleted, map[string]any{ + r.transitionTerminal(runID, RunStatusCompleted, output, "", EventRunCompleted, map[string]any{ "output": output, "usage_totals": usageTotals, "cost_totals": costTotals, @@ -3649,6 +3942,12 @@ func (r *Runner) emitContextWindowSnapshot( } func (r *Runner) failRun(runID string, err error) { + unlockTerminal, ok := r.lockTerminalTransition(runID) + if !ok { + return + } + defer unlockTerminal() + rc := r.configForRun(runID) if err == nil { err = errors.New("run failed") @@ -3692,14 +3991,12 @@ func (r *Runner) failRun(runID string, err error) { r.closeAuditWriter(runID) } - r.setStatus(runID, RunStatusFailed, "", err.Error()) - usageTotals, costTotals := r.accountingTotals(runID) // Profile run history: persist failure record for analysis. r.persistProfileRun(runID, "failed", costTotals.CostUSDTotal) - r.emit(runID, EventRunFailed, map[string]any{ + r.transitionTerminal(runID, RunStatusFailed, "", err.Error(), EventRunFailed, map[string]any{ "error": err.Error(), "usage_totals": usageTotals, "cost_totals": costTotals, @@ -3716,6 +4013,12 @@ func (r *Runner) failRun(runID string, err error) { // reason="max_steps_reached" and max_steps field so clients can distinguish // this terminal state from other failures without parsing the error string. func (r *Runner) failRunMaxSteps(runID string, maxSteps int) { + unlockTerminal, ok := r.lockTerminalTransition(runID) + if !ok { + return + } + defer unlockTerminal() + rc := r.configForRun(runID) err := fmt.Errorf("max steps (%d) reached", maxSteps) @@ -3744,14 +4047,12 @@ func (r *Runner) failRunMaxSteps(runID string, maxSteps int) { r.closeAuditWriter(runID) } - r.setStatus(runID, RunStatusFailed, "", err.Error()) - usageTotals, costTotals := r.accountingTotals(runID) // Profile run history: persist partial record (max steps reached) for analysis. r.persistProfileRun(runID, "partial", costTotals.CostUSDTotal) - r.emit(runID, EventRunFailed, map[string]any{ + r.transitionTerminal(runID, RunStatusFailed, "", err.Error(), EventRunFailed, map[string]any{ "error": err.Error(), "reason": "max_steps_reached", "max_steps": maxSteps, @@ -3770,6 +4071,12 @@ func (r *Runner) failRunMaxSteps(runID string, maxSteps int) { // reason="max_turns_exhausted" and max_turns field so clients can distinguish // this terminal state from other failures without parsing the error string. func (r *Runner) failRunMaxTurns(runID string, maxTurns int) { + unlockTerminal, ok := r.lockTerminalTransition(runID) + if !ok { + return + } + defer unlockTerminal() + rc := r.configForRun(runID) err := fmt.Errorf("max turns (%d) reached", maxTurns) @@ -3798,14 +4105,12 @@ func (r *Runner) failRunMaxTurns(runID string, maxTurns int) { r.closeAuditWriter(runID) } - r.setStatus(runID, RunStatusFailed, "", err.Error()) - usageTotals, costTotals := r.accountingTotals(runID) // Profile run history: persist partial record (max turns exhausted) for analysis. r.persistProfileRun(runID, "partial", costTotals.CostUSDTotal) - r.emit(runID, EventRunFailed, map[string]any{ + r.transitionTerminal(runID, RunStatusFailed, "", err.Error(), EventRunFailed, map[string]any{ "error": err.Error(), "reason": "max_turns_exhausted", "max_turns": maxTurns, @@ -3823,6 +4128,12 @@ func (r *Runner) failRunMaxTurns(runID string, maxTurns int) { // to RunStatusCancelled. It mirrors the structure of failRun but uses the // dedicated cancelled event and status rather than failed. func (r *Runner) cancelledRun(runID string) { + unlockTerminal, ok := r.lockTerminalTransition(runID) + if !ok { + return + } + defer unlockTerminal() + rc := r.configForRun(runID) // Clean up per-run workspace before terminal event (issue #324). r.runWorkspaceCleanup(runID) @@ -3846,10 +4157,8 @@ func (r *Runner) cancelledRun(runID string) { r.closeAuditWriter(runID) } - r.setStatus(runID, RunStatusCancelled, "", "") - usageTotals, costTotals := r.accountingTotals(runID) - r.emit(runID, EventRunCancelled, map[string]any{ + r.transitionTerminal(runID, RunStatusCancelled, "", "", EventRunCancelled, map[string]any{ "usage_totals": usageTotals, "cost_totals": costTotals, }) @@ -4252,62 +4561,90 @@ func (r *Runner) updateStatusContext( if !ok { return false } - if err := state.statusPersist.lock(ctx); err != nil { + if err := state.statusMu.lock(ctx); err != nil { return false } - defer state.statusPersist.unlock() + defer state.statusMu.unlock() - r.mu.Lock() + r.mu.RLock() current, ok := r.runs[runID] - if !ok || current != state || ctx.Err() != nil { - r.mu.Unlock() + available := ok && current == state && !state.terminated && !isTerminalRunStatus(state.run.Status) + r.mu.RUnlock() + if !available || ctx.Err() != nil { return false } - // Terminal state is monotonic. In particular, a delayed pending notifier - // must never downgrade a run that already failed, completed, or cancelled. - if isTerminalRunStatus(state.run.Status) && !isTerminalRunStatus(status) { - r.mu.Unlock() + finalRun, ok := r.statusRunSnapshot(runID, status, output, runErr) + if !ok { return false } - state.run.Status = status - state.run.Output = output - state.run.Error = runErr - state.run.UpdatedAt = time.Now().UTC() - if shouldPersistWorkflowRecap(status) { - state.run.Recap = buildWorkflowRecap(state.run, state.messages, state.events) - } else { - state.run.Recap = nil + if r.statusBeforeCommitHook != nil { + r.statusBeforeCommitHook(runID, status) } - run := state.run - r.mu.Unlock() - - rc := r.configForRun(runID) - if rc.Store == nil { - if ctx.Err() != nil { - return false - } - if afterPersist != nil { - return afterPersist() - } - return ctx.Err() == nil + if ctx.Err() != nil { + return false } - if err := rc.Store.UpdateRun(ctx, runToStoreRun(run)); err != nil { - if rc.Logger != nil { - rc.Logger.Error("store: UpdateRun failed", "run_id", runID, "error", err) - } + if !r.commitStatusSnapshot(runID, finalRun) { return false } - if ctx.Err() != nil { + persistCtx, cancel := context.WithTimeout(ctx, r.terminalStoreTimeoutDuration()) + persisted := r.storeUpdateRunSnapshotContext(persistCtx, finalRun) + cancel() + if !persisted { return false } if afterPersist != nil { - // Keep statusPersist held through its corresponding lifecycle event so + // Keep statusMu held through its corresponding lifecycle event so // terminal mutation/publication cannot overtake a waiting transition. return afterPersist() } return ctx.Err() == nil } +func (r *Runner) statusRunSnapshot(runID string, status RunStatus, output, runErr string) (Run, bool) { + r.mu.RLock() + state, ok := r.runs[runID] + if !ok { + r.mu.RUnlock() + return Run{}, false + } + finalRun := state.run + finalRun.Status = status + finalRun.Output = output + finalRun.Error = runErr + finalRun.UpdatedAt = time.Now().UTC() + if shouldPersistWorkflowRecap(status) { + finalRun.Recap = buildWorkflowRecap(finalRun, state.messages, state.events) + } else { + finalRun.Recap = nil + } + r.mu.RUnlock() + return finalRun, true +} + +func (r *Runner) commitStatusSnapshot(runID string, finalRun Run) bool { + r.mu.Lock() + state, ok := r.runs[runID] + if !ok { + r.mu.Unlock() + return false + } + if state.terminated && !isTerminalRunStatus(finalRun.Status) { + r.mu.Unlock() + return false + } + if isTerminalRunStatus(state.run.Status) && !isTerminalRunStatus(finalRun.Status) { + r.mu.Unlock() + return false + } + state.run.Status = finalRun.Status + state.run.Output = finalRun.Output + state.run.Error = finalRun.Error + state.run.UpdatedAt = finalRun.UpdatedAt + state.run.Recap = finalRun.Recap + r.mu.Unlock() + return true +} + func (r *Runner) setMessages(runID string, messages []Message) { r.mu.Lock() state, ok := r.runs[runID] @@ -5441,8 +5778,8 @@ func (r runTranscriptReader) Snapshot(limit int, includeTools bool) htools.Trans // emit appends one event to the canonical in-memory ledger and mirrors that // same event to subscribers and the optional JSONL recorder. -func (r *Runner) emit(runID string, eventType EventType, payload map[string]any) { - r.emitWithPersistence(context.Background(), runID, eventType, payload, false) +func (r *Runner) emit(runID string, eventType EventType, payload map[string]any) bool { + return r.emitWithPersistence(context.Background(), runID, eventType, payload, false) } func (r *Runner) emitWithPersistence( @@ -5451,6 +5788,50 @@ func (r *Runner) emitWithPersistence( eventType EventType, payload map[string]any, requirePersistence bool, +) bool { + return r.emitWithTerminalCommitContext( + ctx, + runID, + eventType, + payload, + requirePersistence, + nil, + nil, + ) +} + +// emitWithTerminalCommit runs terminalPersist and terminalCommit after the +// terminal event is in replay/store/recorder history but before subscribers +// receive it. The global conversationEventMu is released around recorder and +// status-store I/O while a per-conversation sequence lock prevents overtaking; +// unrelated conversation journals remain available. Non-terminal callers pass +// nil callbacks. +func (r *Runner) emitWithTerminalCommit( + runID string, + eventType EventType, + payload map[string]any, + terminalPersist func(bool), + terminalCommit func(), +) bool { + return r.emitWithTerminalCommitContext( + context.Background(), + runID, + eventType, + payload, + false, + terminalPersist, + terminalCommit, + ) +} + +func (r *Runner) emitWithTerminalCommitContext( + ctx context.Context, + runID string, + eventType EventType, + payload map[string]any, + requirePersistence bool, + terminalPersist func(bool), + terminalCommit func(), ) bool { if ctx == nil { ctx = context.Background() @@ -5458,6 +5839,18 @@ func (r *Runner) emitWithPersistence( if ctx.Err() != nil { return false } + r.mu.RLock() + initialState := r.runs[runID] + conversationID := "" + if initialState != nil { + conversationID = initialState.run.ConversationID + } + r.mu.RUnlock() + if initialState == nil { + return false + } + unlockSequence := r.lockConversationSequence(conversationID) + defer unlockSequence() r.conversationEventMu.Lock() conversationLocked := true defer func() { @@ -5489,17 +5882,50 @@ func (r *Runner) emitWithPersistence( return false } if publishTerminal { - persisted := journal.publishTerminalContext(ctx, delivery) r.conversationEventMu.Unlock() conversationLocked = false - r.pruneCompletedRuns() - journal.dispatchContext(ctx, delivery, requirePersistence) - return persisted + eventPersisted := journal.persistTerminalEventContext(ctx, delivery) + r.conversationEventMu.Lock() + conversationLocked = true + journal.recordTerminalConversation(delivery) + r.conversationEventMu.Unlock() + conversationLocked = false + if r.terminalBeforeDispatchHook != nil { + r.terminalBeforeDispatchHook(runID, eventType) + } + journal.dispatchContext(context.Background(), delivery, false) + if terminalPersist != nil { + terminalPersist(eventPersisted) + } + r.conversationEventMu.Lock() + conversationLocked = true + if terminalCommit != nil { + terminalCommit() + } + journal.fanoutTerminal(delivery) + r.conversationEventMu.Unlock() + conversationLocked = false + return true } if delivery.dropped { r.conversationEventMu.Unlock() conversationLocked = false - return journal.dispatchContext(ctx, delivery, requirePersistence) + dispatched := journal.dispatchContext(ctx, delivery, requirePersistence) + if IsTerminalEvent(eventType) && terminalPersist != nil { + // StorageModeNone intentionally suppresses the terminal event; its + // status remains persistable by explicit policy rather than being + // classified as an AppendEvent failure. + r.markTerminalEventSuppressed(runID) + terminalPersist(true) + } + r.conversationEventMu.Lock() + conversationLocked = true + if IsTerminalEvent(eventType) && terminalCommit != nil { + terminalCommit() + } + r.conversationEventMu.Unlock() + conversationLocked = false + return dispatched } dispatched := journal.dispatchContext(ctx, delivery, requirePersistence) r.conversationEventMu.Unlock() @@ -5507,6 +5933,81 @@ func (r *Runner) emitWithPersistence( return dispatched } +// lockTerminalTransition serializes complete terminal helper lifecycles. The +// second check happens after acquiring terminalMu so a waiter cannot run audit, +// profile, cleanup, backup, or pruning side effects after another helper seals +// the run. +func (r *Runner) lockTerminalTransition(runID string) (func(), bool) { + r.mu.RLock() + state := r.runs[runID] + r.mu.RUnlock() + if state == nil { + return nil, false + } + + state.terminalMu.Lock() + r.mu.RLock() + current := r.runs[runID] + available := current == state && !state.terminated + r.mu.RUnlock() + if !available { + state.terminalMu.Unlock() + return nil, false + } + return state.terminalMu.Unlock, true +} + +// transitionTerminal publishes exactly one matching terminal event before it +// makes the terminal status visible. emit returns false for a run already +// sealed by a competing terminal path, so only the event winner may publish +// status and persist the matching final run record. +func (r *Runner) transitionTerminal( + runID string, + status RunStatus, + output, runErr string, + eventType EventType, + payload map[string]any, +) bool { + if r.terminalTransitionHook != nil { + r.terminalTransitionHook(runID, status, eventType) + } + r.mu.RLock() + state := r.runs[runID] + r.mu.RUnlock() + if state == nil { + return false + } + state.statusMu.Lock() + defer state.statusMu.Unlock() + r.mu.RLock() + current := r.runs[runID] + available := current == state && !state.terminated + r.mu.RUnlock() + if !available { + return false + } + committed := false + var finalRun Run + prepared := false + statusPersisted := false + if !r.emitWithTerminalCommit(runID, eventType, payload, func(eventPersisted bool) { + finalRun, prepared = r.statusRunSnapshot(runID, status, output, runErr) + if prepared && eventPersisted { + statusPersisted = r.storeUpdateRunSnapshot(finalRun) + } + }, func() { + if prepared { + committed = r.commitStatusSnapshot(runID, finalRun) + if committed && statusPersisted { + r.markTerminalStatusPersisted(finalRun) + } + } + }) { + return false + } + return committed +} + // EmitEvent publishes an additive adapter-originated event through the run's // canonical event journal. It is a no-op for unknown or terminal runs. func (r *Runner) EmitEvent(runID string, eventType EventType, payload map[string]any) { @@ -5697,6 +6198,34 @@ func (r *Runner) storeCreateRun(run Run) { } } +func (r *Runner) storeUpdateRunSnapshot(run Run) bool { + ctx, cancel := context.WithTimeout(context.Background(), r.terminalStoreTimeoutDuration()) + defer cancel() + return r.storeUpdateRunSnapshotContext(ctx, run) +} + +func (r *Runner) storeUpdateRunSnapshotContext(ctx context.Context, run Run) bool { + rc := r.configForRun(run.ID) + if rc.Store == nil { + return true + } + sr := runToStoreRun(run) + if err := rc.Store.UpdateRun(ctx, sr); err != nil { + if rc.Logger != nil { + rc.Logger.Error("store: UpdateRun failed", "run_id", run.ID, "error", err) + } + return false + } + return true +} + +func (r *Runner) terminalStoreTimeoutDuration() time.Duration { + if r.terminalStoreTimeout > 0 { + return r.terminalStoreTimeout + } + return terminalEventStoreTimeout +} + func shouldPersistWorkflowRecap(status RunStatus) bool { return status == RunStatusCompleted || status == RunStatusFailed || status == RunStatusCancelled } @@ -5718,7 +6247,7 @@ func (r *Runner) storeAppendEventContext(parent context.Context, ev Event, seq u Payload: string(payloadJSON), Timestamp: ev.Timestamp, } - ctx, cancel := context.WithTimeout(parent, terminalEventStoreTimeout) + ctx, cancel := context.WithTimeout(parent, r.terminalStoreTimeoutDuration()) defer cancel() if err := rc.Store.AppendEvent(ctx, se); err != nil { if rc.Logger != nil { @@ -5738,6 +6267,25 @@ func (r *Runner) markTerminalEventPersisted(runID string) { } } +func (r *Runner) markTerminalEventSuppressed(runID string) { + r.mu.Lock() + defer r.mu.Unlock() + if state := r.runs[runID]; state != nil { + state.terminalEventSuppressed = true + } +} + +func (r *Runner) markTerminalStatusPersisted(run Run) { + r.mu.Lock() + defer r.mu.Unlock() + state := r.runs[run.ID] + if state == nil || !isTerminalRunStatus(state.run.Status) || + state.run.Status != run.Status || !state.run.UpdatedAt.Equal(run.UpdatedAt) { + return + } + state.terminalStatusPersisted = true +} + // storeAppendNewMessages appends any messages in the current run state that // have not yet been persisted to the store. It tracks how many messages have // already been stored via state.storedMsgCount and appends only the new tail. diff --git a/internal/harness/runner_event_journal.go b/internal/harness/runner_event_journal.go index c4049506..17d858cd 100644 --- a/internal/harness/runner_event_journal.go +++ b/internal/harness/runner_event_journal.go @@ -148,21 +148,32 @@ func (j *eventJournal) prepareLocked(state *runState, runID string, eventType Ev return delivery, true } -func (j *eventJournal) publishTerminal(delivery eventDispatch) { - j.publishTerminalContext(context.Background(), delivery) +func (j *eventJournal) persistTerminalEvent(delivery eventDispatch) bool { + return j.persistTerminalEventContext(context.Background(), delivery) } -func (j *eventJournal) publishTerminalContext(ctx context.Context, delivery eventDispatch) bool { +func (j *eventJournal) persistTerminalEventContext(ctx context.Context, delivery eventDispatch) bool { persisted := j.runner.storeAppendEventContext(ctx, delivery.event, delivery.eventSeq) if persisted { j.runner.markTerminalEventPersisted(delivery.runID) } + return persisted +} + +func (j *eventJournal) recordTerminalConversation(delivery eventDispatch) { j.runner.recordConversationEvent(delivery.conversationID, delivery.event) +} +func (j *eventJournal) fanoutTerminal(delivery eventDispatch) { for _, sub := range delivery.subscribers { j.runner.sendTerminalSubscriberEvent(sub.ch, sub.event) } - return persisted +} + +func (j *eventJournal) publishTerminal(delivery eventDispatch) { + j.persistTerminalEvent(delivery) + j.recordTerminalConversation(delivery) + j.fanoutTerminal(delivery) } func (j *eventJournal) dispatch(delivery eventDispatch) { @@ -173,6 +184,7 @@ func (j *eventJournal) dispatchContext(ctx context.Context, delivery eventDispat if delivery.dropped { if delivery.closeRecorder != nil { delivery.closeRecorder() + j.waitForRecorderDrain(delivery, j.runner.configForRun(delivery.runID)) } // StorageModeNone is an intentional policy outcome, not a transient // publication failure. Treat it as consumed so strict publishers do not @@ -240,16 +252,7 @@ func (j *eventJournal) dispatchContext(ctx context.Context, delivery eventDispat } } delivery.closeRecorder() - drainTimer := time.NewTimer(recorderDrainTimeout) - defer drainTimer.Stop() - select { - case <-delivery.recorderDone: - case <-drainTimer.C: - if rc.Logger != nil { - rc.Logger.Error("rollout recorder: drain timeout exceeded, JSONL may be incomplete", - "run_id", delivery.runID, "timeout", recorderDrainTimeout) - } - } + j.waitForRecorderDrain(delivery, rc) } return true } @@ -281,3 +284,19 @@ func (j *eventJournal) discardPreparedEvent(delivery eventDispatch) { state.events = state.events[:last] state.nextEventSeq = delivery.eventSeq } + +func (j *eventJournal) waitForRecorderDrain(delivery eventDispatch, rc RunnerConfig) { + if delivery.recorderDone == nil { + return + } + drainTimer := time.NewTimer(recorderDrainTimeout) + defer drainTimer.Stop() + select { + case <-delivery.recorderDone: + case <-drainTimer.C: + if rc.Logger != nil { + rc.Logger.Error("rollout recorder: drain timeout exceeded, JSONL may be incomplete", + "run_id", delivery.runID, "timeout", recorderDrainTimeout) + } + } +} diff --git a/internal/harness/runner_meta_test.go b/internal/harness/runner_meta_test.go index 9c65a6f9..db9ae1c9 100644 --- a/internal/harness/runner_meta_test.go +++ b/internal/harness/runner_meta_test.go @@ -585,30 +585,12 @@ func TestRunnerConcurrentMetaMessageInjection(t *testing.T) { // collectRunEventsWithTimeout is like collectRunEvents but with configurable timeout. func collectRunEventsWithTimeout(runner *Runner, runID string, timeout time.Duration) ([]Event, error) { + deadline := time.Now().Add(timeout) history, stream, cancel, err := runner.Subscribe(runID) if err != nil { return nil, err } defer cancel() - events := append([]Event(nil), history...) - if hasTerminalEvent(events) { - return events, nil - } - - timer := time.After(timeout) - for { - select { - case ev, ok := <-stream: - if !ok { - return events, nil - } - events = append(events, ev) - if IsTerminalEvent(ev.Type) { - return events, nil - } - case <-timer: - return nil, context.DeadlineExceeded - } - } + return collectSubscribedRunEvents(runner, runID, history, stream, deadline, nil) } diff --git a/internal/harness/runner_prune_test.go b/internal/harness/runner_prune_test.go index 43291ec6..e4c9e1ab 100644 --- a/internal/harness/runner_prune_test.go +++ b/internal/harness/runner_prune_test.go @@ -5,9 +5,11 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "testing" "time" + "go-agent-harness/internal/forensics/redaction" runstore "go-agent-harness/internal/store" ) @@ -39,26 +41,497 @@ func TestRunner_PruneCompletedRunsFromMemory(t *testing.T) { } func TestRunner_PruneWaitsForTerminalEventPersistence(t *testing.T) { - runner := NewRunner(staticContentProvider{content: "done"}, NewRegistry(), RunnerConfig{ + release := make(chan struct{}) + runner := NewRunner(&blockingProvider{blocker: release}, NewRegistry(), RunnerConfig{ DefaultModel: "test-model", MaxSteps: 1, MaxCompletedRetention: 1, Store: &terminalAppendFailStore{Store: runstore.NewMemoryStore()}, }) + runIDs := make([]string, 0, 3) for i := 0; i < 3; i++ { run, err := runner.StartRun(RunRequest{Prompt: fmt.Sprintf("unpersisted %d", i)}) if err != nil { t.Fatalf("StartRun: %v", err) } - waitForStatus(t, runner, run.ID, RunStatusCompleted) + runIDs = append(runIDs, run.ID) + } + close(release) + for _, runID := range runIDs { + waitForStatus(t, runner, runID, RunStatusCompleted) } runner.mu.RLock() - defer runner.mu.RUnlock() if got := len(runner.runs); got != 3 { + runner.mu.RUnlock() t.Fatalf("pruned terminal runs before terminal events persisted: got %d, want 3", got) } + runner.mu.RUnlock() + + _, err := runner.StartRun(RunRequest{Prompt: "must fail closed"}) + requireTerminalDurabilityBackpressure(t, err, 3, 1) + _, err = runner.ContinueRun(runIDs[0], "valid source must also fail closed") + requireTerminalDurabilityBackpressure(t, err, 3, 1) + runner.mu.RLock() + reservations := runner.runs[runIDs[0]].continuationReservations + runner.mu.RUnlock() + if reservations != 0 { + t.Fatalf("continuation reservations after backpressure=%d, want 0", reservations) + } + + runner.mu.Lock() + runner.runs["noncompleted-source"] = &runState{ + run: Run{ID: "noncompleted-source", Status: RunStatusRunning}, + subscribers: make(map[chan Event]struct{}), + } + runner.mu.Unlock() + if _, err := runner.ContinueRun("missing-source", "continue"); !errors.Is(err, ErrRunNotFound) { + t.Fatalf("ContinueRun missing source error=%v, want ErrRunNotFound before backpressure", err) + } + if _, err := runner.ContinueRun("noncompleted-source", "continue"); !errors.Is(err, ErrRunNotCompleted) { + t.Fatalf("ContinueRun noncompleted source error=%v, want ErrRunNotCompleted before backpressure", err) + } +} + +func TestRunner_PruneWaitsForTerminalStatusPersistence(t *testing.T) { + release := make(chan struct{}) + store := &terminalFailureStore{ + Store: runstore.NewMemoryStore(), + failUpdateRun: true, + } + runner := NewRunner(&blockingProvider{blocker: release}, NewRegistry(), RunnerConfig{ + DefaultModel: "test-model", + MaxSteps: 1, + MaxCompletedRetention: 1, + Store: store, + }) + + runIDs := make([]string, 0, 3) + for i := 0; i < 3; i++ { + run, err := runner.StartRun(RunRequest{Prompt: fmt.Sprintf("status pending %d", i)}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + runIDs = append(runIDs, run.ID) + } + close(release) + for _, runID := range runIDs { + waitForStatus(t, runner, runID, RunStatusCompleted) + } + + runner.mu.RLock() + if got := len(runner.runs); got != 3 { + runner.mu.RUnlock() + t.Fatalf("pruned terminal runs before terminal statuses persisted: got %d, want 3", got) + } + runner.mu.RUnlock() + + for _, runID := range runIDs { + stored, err := store.Store.GetRun(context.Background(), runID) + if err != nil { + t.Fatalf("GetRun(%s): %v", runID, err) + } + if isTerminalStoreStatus(stored.Status) { + t.Fatalf("durable run %s status=%s, want non-terminal after UpdateRun failure", runID, stored.Status) + } + } + + _, err := runner.StartRun(RunRequest{Prompt: "must fail closed"}) + requireTerminalDurabilityBackpressure(t, err, 3, 1) +} + +func TestRunner_TerminalStatusPersistenceRecoveryAllowsConcurrentAdmissions(t *testing.T) { + store := &recoveringTerminalStatusStore{Store: runstore.NewMemoryStore()} + store.fail.Store(true) + runner := NewRunner(staticContentProvider{content: "done"}, NewRegistry(), RunnerConfig{ + DefaultModel: "test-model", + MaxSteps: 1, + MaxCompletedRetention: 1, + Store: store, + }) + + first, err := runner.StartRun(RunRequest{Prompt: "create pending terminal status"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + waitForStatus(t, runner, first.ID, RunStatusCompleted) + + const callers = 16 + rejectStart := make(chan struct{}) + rejectErrs := make(chan error, callers) + for i := 0; i < callers; i++ { + go func(i int) { + <-rejectStart + _, err := runner.StartRun(RunRequest{Prompt: fmt.Sprintf("still unavailable %d", i)}) + rejectErrs <- err + }(i) + } + close(rejectStart) + for i := 0; i < callers; i++ { + requireTerminalDurabilityBackpressure(t, <-rejectErrs, 1, 1) + } + + store.fail.Store(false) + start := make(chan struct{}) + errs := make(chan error, callers) + for i := 0; i < callers; i++ { + go func(i int) { + <-start + _, err := runner.StartRun(RunRequest{Prompt: fmt.Sprintf("recovered %d", i)}) + errs <- err + }(i) + } + close(start) + for i := 0; i < callers; i++ { + if err := <-errs; err != nil { + t.Fatalf("concurrent recovered StartRun %d: %v", i, err) + } + } + + ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second) + defer cancel() + if err := runner.Shutdown(ctx); err != nil { + t.Fatalf("Shutdown: %v", err) + } + if got := store.successfulTerminalUpdates.Load(); got == 0 { + t.Fatal("recovery never persisted a pending terminal status") + } +} + +func TestRunner_TerminalStatusRecoveryImmediatelyRestoresRetentionWindow(t *testing.T) { + release := make(chan struct{}) + store := &recoveringTerminalStatusStore{Store: runstore.NewMemoryStore()} + store.fail.Store(true) + runner := NewRunner(&blockingProvider{blocker: release}, NewRegistry(), RunnerConfig{ + DefaultModel: "test-model", + MaxSteps: 1, + MaxCompletedRetention: 1, + Store: store, + }) + + runIDs := make([]string, 0, 3) + for i := 0; i < 3; i++ { + run, err := runner.StartRun(RunRequest{Prompt: fmt.Sprintf("recover and prune %d", i)}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + runIDs = append(runIDs, run.ID) + } + close(release) + for _, runID := range runIDs { + waitForStatus(t, runner, runID, RunStatusCompleted) + } + + store.fail.Store(false) + if err := runner.ensureTerminalDurabilityCapacity(""); err != nil { + t.Fatalf("ensureTerminalDurabilityCapacity after recovery: %v", err) + } + runner.mu.RLock() + retained := len(runner.runs) + runner.mu.RUnlock() + if retained != 1 { + t.Fatalf("recovered terminal states retained=%d, want exact retention window 1", retained) + } + for _, runID := range runIDs { + stored, err := store.GetRun(context.Background(), runID) + if err != nil { + t.Fatalf("store.GetRun(%s): %v", runID, err) + } + if stored.Status != runstore.RunStatusCompleted { + t.Fatalf("stored run %s status=%s, want completed after recovery", runID, stored.Status) + } + } + + if _, err := runner.StartRun(RunRequest{Prompt: "admitted after bounded recovery"}); err != nil { + t.Fatalf("StartRun after recovery: %v", err) + } +} + +func TestRunner_TerminalStatusRecoveryPreservesContinuationSource(t *testing.T) { + release := make(chan struct{}) + store := &recoveringTerminalStatusStore{Store: runstore.NewMemoryStore()} + store.fail.Store(true) + runner := NewRunner(&blockingProvider{blocker: release}, NewRegistry(), RunnerConfig{ + DefaultModel: "test-model", + MaxSteps: 1, + MaxCompletedRetention: 1, + Store: store, + }) + + runIDs := make([]string, 0, 3) + for i := 0; i < 3; i++ { + run, err := runner.StartRun(RunRequest{Prompt: fmt.Sprintf("continuation recovery %d", i)}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + runIDs = append(runIDs, run.ID) + } + close(release) + for _, runID := range runIDs { + waitForStatus(t, runner, runID, RunStatusCompleted) + } + + store.fail.Store(false) + continued, err := runner.ContinueRun(runIDs[0], "continue after persistence recovery") + if err != nil { + t.Fatalf("ContinueRun after recovery: %v", err) + } + if continued.ID == "" { + t.Fatal("ContinueRun returned an empty run ID") + } +} + +func TestRunner_ConcurrentStartRecoveryCannotPruneValidatedContinuationSource(t *testing.T) { + releaseProvider := make(chan struct{}) + store := &recoveringTerminalStatusStore{Store: runstore.NewMemoryStore()} + store.fail.Store(true) + runner := NewRunner(&blockingProvider{blocker: releaseProvider}, NewRegistry(), RunnerConfig{ + DefaultModel: "test-model", + MaxSteps: 1, + MaxCompletedRetention: 1, + Store: store, + }) + + runIDs := make([]string, 0, 3) + for i := 0; i < 3; i++ { + run, err := runner.StartRun(RunRequest{Prompt: fmt.Sprintf("concurrent continuation recovery %d", i)}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + runIDs = append(runIDs, run.ID) + } + close(releaseProvider) + for _, runID := range runIDs { + waitForStatus(t, runner, runID, RunStatusCompleted) + } + + // Make the continuation source the deterministic first prune candidate. + base := time.Now().UTC().Add(-time.Hour) + runner.mu.Lock() + for i, runID := range runIDs { + runner.runs[runID].run.UpdatedAt = base.Add(time.Duration(i) * time.Minute) + } + runner.mu.Unlock() + + validated := make(chan struct{}) + releaseContinue := make(chan struct{}) + runner.continuationAfterValidationHook = func(gotRunID string) { + if gotRunID != runIDs[0] { + return + } + close(validated) + <-releaseContinue + } + store.fail.Store(false) + + type continueResult struct { + run Run + err error + } + continueDone := make(chan continueResult, 1) + go func() { + run, err := runner.ContinueRun(runIDs[0], "continue after concurrent recovery") + continueDone <- continueResult{run: run, err: err} + }() + select { + case <-validated: + case <-time.After(2 * time.Second): + t.Fatal("ContinueRun did not reach the post-validation boundary") + } + + if _, err := runner.StartRun(RunRequest{Prompt: "concurrent recovery admission"}); err != nil { + close(releaseContinue) + t.Fatalf("concurrent StartRun recovery: %v", err) + } + close(releaseContinue) + + select { + case result := <-continueDone: + if result.err != nil { + t.Fatalf("ContinueRun after concurrent Start recovery: %v", result.err) + } + if result.run.ID == "" { + t.Fatal("ContinueRun returned an empty run ID") + } + case <-time.After(2 * time.Second): + t.Fatal("ContinueRun did not finish after concurrent recovery") + } +} + +func TestRunner_TerminalDurabilityAdmissionRetryUsesSingleUnlockedDeadline(t *testing.T) { + release := make(chan struct{}) + store := &blockingAdmissionRecoveryStore{ + Store: runstore.NewMemoryStore(), + started: make(chan struct{}), + } + runner := NewRunner(&blockingProvider{blocker: release}, NewRegistry(), RunnerConfig{ + DefaultModel: "test-model", + MaxSteps: 1, + MaxCompletedRetention: 1, + Store: store, + }) + runner.terminalStoreTimeout = 200 * time.Millisecond + + runIDs := make([]string, 0, 3) + for i := 0; i < 3; i++ { + run, err := runner.StartRun(RunRequest{Prompt: fmt.Sprintf("deadline pending %d", i)}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + runIDs = append(runIDs, run.ID) + } + close(release) + for _, runID := range runIDs { + waitForStatus(t, runner, runID, RunStatusCompleted) + } + + const unrelatedID = "admission-lock-control" + runner.mu.Lock() + runner.runs[unrelatedID] = &runState{ + run: Run{ + ID: unrelatedID, + ConversationID: "admission-lock-control-conversation", + Status: RunStatusRunning, + }, + subscribers: make(map[chan Event]struct{}), + } + runner.mu.Unlock() + runner.storeCreateRun(Run{ID: unrelatedID, ConversationID: "admission-lock-control-conversation", Status: RunStatusRunning}) + + store.block.Store(true) + admissionDone := make(chan error, 1) + startedAt := time.Now() + go func() { + _, err := runner.StartRun(RunRequest{Prompt: "bounded blocked retry"}) + admissionDone <- err + }() + select { + case <-store.started: + case <-time.After(2 * time.Second): + t.Fatal("admission recovery did not reach UpdateRun") + } + + runner.mu.RLock() + pendingStates := make([]*runState, 0, len(runIDs)) + for _, runID := range runIDs { + pendingStates = append(pendingStates, runner.runs[runID]) + } + runner.mu.RUnlock() + for i, state := range pendingStates { + if state == nil || !state.statusMu.TryLock() { + t.Fatalf("pending status lock %d held during admission store I/O", i) + } + state.statusMu.Unlock() + } + + getDone := make(chan struct{}) + go func() { + _, _ = runner.GetRun(unrelatedID) + close(getDone) + }() + select { + case <-getDone: + case <-time.After(100 * time.Millisecond): + t.Fatal("GetRun blocked behind admission recovery store I/O") + } + + emitDone := make(chan struct{}) + go func() { + runner.emit(unrelatedID, EventAssistantMessage, map[string]any{"content": "unrelated"}) + close(emitDone) + }() + select { + case <-emitDone: + case <-time.After(100 * time.Millisecond): + t.Fatal("conversation event journal blocked behind admission recovery store I/O") + } + + select { + case err := <-admissionDone: + requireTerminalDurabilityBackpressure(t, err, 3, 1) + if elapsed := time.Since(startedAt); elapsed >= 500*time.Millisecond { + t.Fatalf("admission retry took %s, want one shared 200ms deadline rather than per-run waits", elapsed) + } + case <-time.After(time.Second): + t.Fatal("admission retry exceeded its shared deadline") + } +} + +func TestRunner_PruneTreatsStorageModeNoneAsIntentionalEventSuppression(t *testing.T) { + pipeline := redaction.NewPipeline( + redaction.NewRedactor(nil), + redaction.EventClassConfig{string(EventRunCompleted): redaction.StorageModeNone}, + ) + store := runstore.NewMemoryStore() + runner := NewRunner(staticContentProvider{content: "done"}, NewRegistry(), RunnerConfig{ + DefaultModel: "test-model", + MaxSteps: 1, + MaxCompletedRetention: 1, + RedactionPipeline: pipeline, + Store: store, + }) + + for i := 0; i < 3; i++ { + run, err := runner.StartRun(RunRequest{Prompt: fmt.Sprintf("suppressed %d", i)}) + if err != nil { + t.Fatalf("StartRun %d: %v", i, err) + } + waitForStatus(t, runner, run.ID, RunStatusCompleted) + stored, err := store.GetRun(context.Background(), run.ID) + if err != nil { + t.Fatalf("store.GetRun(%s): %v", run.ID, err) + } + if stored.Status != runstore.RunStatusCompleted { + t.Fatalf("stored status=%s, want completed", stored.Status) + } + } + + waitForRunnerPrune(t, runner, func() bool { + runner.mu.RLock() + defer runner.mu.RUnlock() + return len(runner.runs) <= 1 + }) +} + +func TestRunner_NoStorePreservesInMemoryTerminalRunsWithoutBackpressure(t *testing.T) { + runner := NewRunner(staticContentProvider{content: "done"}, NewRegistry(), RunnerConfig{ + DefaultModel: "test-model", + MaxSteps: 1, + MaxCompletedRetention: 1, + }) + + for i := 0; i < 3; i++ { + run, err := runner.StartRun(RunRequest{Prompt: fmt.Sprintf("memory only %d", i)}) + if err != nil { + t.Fatalf("StartRun %d: %v", i, err) + } + waitForStatus(t, runner, run.ID, RunStatusCompleted) + } + runner.mu.RLock() + retained := len(runner.runs) + runner.mu.RUnlock() + if retained != 3 { + t.Fatalf("no-store retained runs=%d, want 3", retained) + } + if _, err := runner.StartRun(RunRequest{Prompt: "no-store remains available"}); err != nil { + t.Fatalf("no-store StartRun unexpectedly backpressured: %v", err) + } +} + +func requireTerminalDurabilityBackpressure( + t *testing.T, + err error, + wantPending, wantLimit int, +) { + t.Helper() + var backpressure *TerminalDurabilityBackpressureError + if !errors.As(err, &backpressure) { + t.Fatalf("error=%v, want TerminalDurabilityBackpressureError", err) + } + if backpressure.Pending != wantPending || backpressure.Limit != wantLimit { + t.Fatalf("backpressure=%+v, want pending=%d limit=%d", backpressure, wantPending, wantLimit) + } } type terminalAppendFailStore struct{ runstore.Store } @@ -70,6 +543,41 @@ func (s *terminalAppendFailStore) AppendEvent(_ context.Context, event *runstore return s.Store.AppendEvent(context.Background(), event) } +type recoveringTerminalStatusStore struct { + runstore.Store + fail atomic.Bool + successfulTerminalUpdates atomic.Int64 +} + +func (s *recoveringTerminalStatusStore) UpdateRun(ctx context.Context, run *runstore.Run) error { + if isTerminalStoreStatus(run.Status) { + if s.fail.Load() { + return errors.New("terminal status store unavailable") + } + s.successfulTerminalUpdates.Add(1) + } + return s.Store.UpdateRun(ctx, run) +} + +type blockingAdmissionRecoveryStore struct { + runstore.Store + block atomic.Bool + started chan struct{} + startedOnce sync.Once +} + +func (s *blockingAdmissionRecoveryStore) UpdateRun(ctx context.Context, run *runstore.Run) error { + if !isTerminalStoreStatus(run.Status) { + return s.Store.UpdateRun(ctx, run) + } + if s.block.Load() { + s.startedOnce.Do(func() { close(s.started) }) + <-ctx.Done() + return ctx.Err() + } + return errors.New("terminal status store unavailable") +} + func TestRunner_PruneKeepsCompletedRunWithActiveSubscriber(t *testing.T) { t.Parallel() diff --git a/internal/harness/runner_terminal_atomicity_test.go b/internal/harness/runner_terminal_atomicity_test.go new file mode 100644 index 00000000..001472fb --- /dev/null +++ b/internal/harness/runner_terminal_atomicity_test.go @@ -0,0 +1,496 @@ +package harness + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" +) + +func TestTerminalStatusNeverPrecedesTerminalReplayEvent(t *testing.T) { + tests := []struct { + name string + provider Provider + wantStatus RunStatus + wantEvent EventType + cancelStarted <-chan struct{} + }{ + { + name: "completed", + provider: staticContentProvider{content: "done"}, + wantStatus: RunStatusCompleted, + wantEvent: EventRunCompleted, + }, + { + name: "failed_with_error_context", + provider: &errorProvider{err: errors.New("provider unavailable")}, + wantStatus: RunStatusFailed, + wantEvent: EventRunFailed, + }, + } + + cancelProvider := &terminalCancellationProvider{started: make(chan struct{})} + tests = append(tests, struct { + name string + provider Provider + wantStatus RunStatus + wantEvent EventType + cancelStarted <-chan struct{} + }{ + name: "cancelled", + provider: cancelProvider, + wantStatus: RunStatusCancelled, + wantEvent: EventRunCancelled, + cancelStarted: cancelProvider.started, + }) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + reached := make(chan struct{}) + release := make(chan struct{}) + t.Cleanup(func() { + select { + case <-release: + default: + close(release) + } + }) + + runner := NewRunner(tt.provider, NewRegistry(), RunnerConfig{ + ErrorChainEnabled: true, + CausalGraphEnabled: true, + }) + runner.terminalTransitionHook = func(_ string, status RunStatus, eventType EventType) { + if status != tt.wantStatus || eventType != tt.wantEvent { + return + } + close(reached) + <-release + } + + run, err := runner.StartRun(RunRequest{Prompt: "terminal atomicity " + tt.name}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + if tt.cancelStarted != nil { + select { + case <-tt.cancelStarted: + case <-time.After(2 * time.Second): + t.Fatal("provider did not start") + } + if err := runner.CancelRun(run.ID); err != nil { + t.Fatalf("CancelRun: %v", err) + } + } + + select { + case <-reached: + case <-time.After(2 * time.Second): + t.Fatalf("terminal transition did not reach %s barrier", tt.wantStatus) + } + + current, ok := runner.GetRun(run.ID) + if !ok { + t.Fatalf("GetRun(%q) returned not found", run.ID) + } + history, _, unsubscribe, err := runner.Subscribe(run.ID) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + unsubscribe() + + if isTerminalRunStatus(current.Status) && !containsEventType(history, tt.wantEvent) { + t.Fatalf("GetRun returned %s before %s was replayable; history=%v", + current.Status, tt.wantEvent, terminalAtomicityEventTypes(history)) + } + if tt.wantStatus == RunStatusFailed && + containsEventType(history, EventRunFailed) && + !eventPrecedes(history, EventErrorContext, EventRunFailed) { + t.Fatalf("failed status became observable without error.context before run.failed; history=%v", + terminalAtomicityEventTypes(history)) + } + + close(release) + waitForStatus(t, runner, run.ID, tt.wantStatus) + + finalHistory, _, finalUnsubscribe, err := runner.Subscribe(run.ID) + if err != nil { + t.Fatalf("Subscribe after terminal status: %v", err) + } + finalUnsubscribe() + if !containsEventType(finalHistory, tt.wantEvent) { + t.Fatalf("terminal status %s missing matching replay event %s; history=%v", + tt.wantStatus, tt.wantEvent, terminalAtomicityEventTypes(finalHistory)) + } + switch tt.wantStatus { + case RunStatusCompleted: + if !eventPrecedes(finalHistory, EventCausalGraphSnapshot, EventRunCompleted) { + t.Fatalf("completed status missing causal snapshot before run.completed; history=%v", + terminalAtomicityEventTypes(finalHistory)) + } + case RunStatusFailed: + if !eventPrecedes(finalHistory, EventErrorContext, EventRunFailed) { + t.Fatalf("failed status missing error.context before run.failed; history=%v", + terminalAtomicityEventTypes(finalHistory)) + } + } + }) + } +} + +func TestCompetingTerminalTransitionsPublishMatchingStatusAndEvent(t *testing.T) { + const iterations = 100 + for iteration := 0; iteration < iterations; iteration++ { + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{}) + runID := fmt.Sprintf("terminal-race-%d", iteration) + runner.runs[runID] = &runState{ + run: Run{ + ID: runID, + ConversationID: "terminal-race-conversation", + Status: RunStatusRunning, + }, + subscribers: make(map[chan Event]struct{}), + } + + start := make(chan struct{}) + var transitions sync.WaitGroup + transitions.Add(3) + go func() { + defer transitions.Done() + <-start + runner.completeRun(runID, "done") + }() + go func() { + defer transitions.Done() + <-start + runner.failRun(runID, errors.New("failed")) + }() + go func() { + defer transitions.Done() + <-start + runner.cancelledRun(runID) + }() + close(start) + transitions.Wait() + + current, ok := runner.GetRun(runID) + if !ok { + t.Fatalf("iteration %d: GetRun returned not found", iteration) + } + history, _, unsubscribe, err := runner.Subscribe(runID) + if err != nil { + t.Fatalf("iteration %d: Subscribe: %v", iteration, err) + } + unsubscribe() + + terminalEvents := make([]EventType, 0, 1) + for _, event := range history { + if IsTerminalEvent(event.Type) { + terminalEvents = append(terminalEvents, event.Type) + } + } + if len(terminalEvents) != 1 { + t.Fatalf("iteration %d: terminal events=%v, want exactly one; history=%v", + iteration, terminalEvents, terminalAtomicityEventTypes(history)) + } + if wantStatus := statusForTerminalEvent(terminalEvents[0]); current.Status != wantStatus { + t.Fatalf("iteration %d: status=%s does not match sealed event=%s (want %s)", + iteration, current.Status, terminalEvents[0], wantStatus) + } + } +} + +func TestTerminalConversationFanoutCannotBeOvertaken(t *testing.T) { + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{}) + const ( + conversationID = "terminal-conversation-order" + terminalRunID = "terminal-conversation-order-terminal" + laterRunID = "terminal-conversation-order-later" + ) + for _, runID := range []string{terminalRunID, laterRunID} { + runner.runs[runID] = &runState{ + run: Run{ + ID: runID, + ConversationID: conversationID, + Status: RunStatusRunning, + }, + subscribers: make(map[chan Event]struct{}), + } + } + + history, stream, unsubscribe, err := runner.SubscribeConversation(conversationID) + if err != nil { + t.Fatalf("SubscribeConversation: %v", err) + } + defer unsubscribe() + if len(history) != 0 { + t.Fatalf("initial conversation history = %v, want empty", terminalAtomicityEventTypes(history)) + } + + reachedDispatch := make(chan struct{}) + releaseDispatch := make(chan struct{}) + runner.terminalBeforeDispatchHook = func(runID string, eventType EventType) { + if runID != terminalRunID || eventType != EventRunCompleted { + return + } + close(reachedDispatch) + <-releaseDispatch + } + + terminalDone := make(chan bool, 1) + go func() { + terminalDone <- runner.transitionTerminal( + terminalRunID, + RunStatusCompleted, + "done", + "", + EventRunCompleted, + map[string]any{"output": "done"}, + ) + }() + select { + case <-reachedDispatch: + case <-time.After(2 * time.Second): + t.Fatal("terminal transition did not reach pre-dispatch barrier") + } + + // Slow terminal persistence/recorder/status work must not monopolize the + // global replay-to-live mutex. Same-conversation ordering is now carried by + // the narrower keyed sequence lock for the complete terminal transition. + if !runner.conversationEventMu.TryLock() { + close(releaseDispatch) + <-terminalDone + t.Fatal("global conversation event lock held during terminal I/O") + } + runner.conversationEventMu.Unlock() + + laterStarted := make(chan struct{}) + laterDone := make(chan struct{}) + go func() { + close(laterStarted) + runner.emit(laterRunID, EventAssistantMessage, map[string]any{"content": "later"}) + close(laterDone) + }() + <-laterStarted + select { + case <-laterDone: + close(releaseDispatch) + <-terminalDone + t.Fatal("later same-conversation event overtook terminal fanout") + default: + } + close(releaseDispatch) + if won := <-terminalDone; !won { + t.Fatal("terminal transition did not win") + } + select { + case <-laterDone: + case <-time.After(2 * time.Second): + t.Fatal("later conversation event remained blocked") + } + + got := make([]Event, 0, 2) + for len(got) < 2 { + select { + case event := <-stream: + got = append(got, event) + case <-time.After(2 * time.Second): + t.Fatalf("conversation subscriber received %v, want terminal then later event", + terminalAtomicityEventTypes(got)) + } + } + if got[0].RunID != terminalRunID || got[0].Type != EventRunCompleted || + got[1].RunID != laterRunID || got[1].Type != EventAssistantMessage { + t.Fatalf("conversation subscriber order = %v, want %s/%s then %s/%s", + terminalAtomicityEventTypes(got), terminalRunID, EventRunCompleted, + laterRunID, EventAssistantMessage) + } +} + +func TestCollectSubscribedRunEventsRejectsClosedStreamWithoutTerminalEvent(t *testing.T) { + const runID = "closed-before-terminal-event" + runner := NewRunner(staticContentProvider{content: "unused"}, NewRegistry(), RunnerConfig{}) + runner.mu.Lock() + runner.runs[runID] = &runState{ + run: Run{ID: runID, Status: RunStatusCompleted}, + subscribers: make(map[chan Event]struct{}), + } + runner.mu.Unlock() + + stream := make(chan Event) + close(stream) + history := []Event{{RunID: runID, Type: EventRunStarted}} + events, err := collectSubscribedRunEvents( + runner, runID, history, stream, time.Now().Add(time.Second), nil, + ) + if err == nil { + t.Fatalf("closed stream without terminal event returned success: events=%v", eventTypes(events)) + } + if len(events) != 1 || events[0].Type != EventRunStarted { + t.Fatalf("closed-stream events = %v, want preserved history", eventTypes(events)) + } +} + +func TestCollectSubscribedRunEventsRejectsMismatchedTerminalEventAndStatus(t *testing.T) { + const runID = "mismatched-terminal-event-status" + runner := NewRunner(staticContentProvider{content: "unused"}, NewRegistry(), RunnerConfig{}) + runner.mu.Lock() + runner.runs[runID] = &runState{ + run: Run{ID: runID, Status: RunStatusFailed}, + subscribers: make(map[chan Event]struct{}), + } + runner.mu.Unlock() + + history := []Event{{RunID: runID, Type: EventRunCompleted}} + events, err := collectSubscribedRunEvents( + runner, runID, history, nil, time.Now().Add(time.Second), nil, + ) + if err == nil { + t.Fatalf("completed event with failed status returned success: events=%v", eventTypes(events)) + } + if len(events) != 1 || events[0].Type != EventRunCompleted { + t.Fatalf("mismatched-status events = %v, want preserved history", eventTypes(events)) + } +} + +func TestCollectRunEventsWaitsForTerminalStatusAfterReplay(t *testing.T) { + reached := make(chan struct{}) + release := make(chan struct{}) + t.Cleanup(func() { + select { + case <-release: + default: + close(release) + } + }) + + runner := NewRunner(staticContentProvider{content: "done"}, NewRegistry(), RunnerConfig{}) + runner.terminalBeforeDispatchHook = func(_ string, eventType EventType) { + if eventType != EventRunCompleted { + return + } + close(reached) + <-release + } + run, err := runner.StartRun(RunRequest{Prompt: "settled event collection"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + select { + case <-reached: + case <-time.After(2 * time.Second): + t.Fatal("terminal publication did not reach the pre-status barrier") + } + + history, stream, cancel, err := runner.Subscribe(run.ID) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer cancel() + settlementEntered := make(chan struct{}) + collected := make(chan error, 1) + go func() { + events, err := collectSubscribedRunEvents( + runner, + run.ID, + history, + stream, + time.Now().Add(4*time.Second), + func() { close(settlementEntered) }, + ) + if err == nil && !containsEventType(events, EventRunCompleted) { + err = fmt.Errorf("collected events missing %s", EventRunCompleted) + } + collected <- err + }() + select { + case <-settlementEntered: + case <-time.After(2 * time.Second): + t.Fatal("collector did not enter terminal settlement") + } + + current, ok := runner.GetRun(run.ID) + if !ok { + t.Fatalf("GetRun(%q) returned not found", run.ID) + } + if isTerminalRunStatus(current.Status) { + t.Fatalf("status=%s before releasing terminal commit, want non-terminal", current.Status) + } + select { + case err := <-collected: + t.Fatalf("collector returned after settlement entry but before terminal status commit: %v", err) + default: + } + close(release) + select { + case err := <-collected: + if err != nil { + t.Fatalf("collectRunEvents after terminal status commit: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("collectRunEvents did not return after terminal status commit") + } +} + +type terminalCancellationProvider struct { + started chan struct{} +} + +func (p *terminalCancellationProvider) Complete(ctx context.Context, _ CompletionRequest) (CompletionResult, error) { + select { + case <-p.started: + default: + close(p.started) + } + <-ctx.Done() + return CompletionResult{}, ctx.Err() +} + +func containsEventType(events []Event, want EventType) bool { + for _, event := range events { + if event.Type == want { + return true + } + } + return false +} + +func eventPrecedes(events []Event, first, second EventType) bool { + firstIndex, secondIndex := -1, -1 + for i, event := range events { + switch event.Type { + case first: + if firstIndex == -1 { + firstIndex = i + } + case second: + if secondIndex == -1 { + secondIndex = i + } + } + } + return firstIndex >= 0 && secondIndex > firstIndex +} + +func terminalAtomicityEventTypes(events []Event) []string { + types := make([]string, 0, len(events)) + for _, event := range events { + types = append(types, fmt.Sprint(event.Type)) + } + return types +} + +func statusForTerminalEvent(eventType EventType) RunStatus { + switch eventType { + case EventRunCompleted: + return RunStatusCompleted + case EventRunFailed: + return RunStatusFailed + case EventRunCancelled: + return RunStatusCancelled + default: + return "" + } +} diff --git a/internal/harness/runner_terminal_failure_policy_test.go b/internal/harness/runner_terminal_failure_policy_test.go new file mode 100644 index 00000000..5662a5bc --- /dev/null +++ b/internal/harness/runner_terminal_failure_policy_test.go @@ -0,0 +1,527 @@ +package harness + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "go-agent-harness/internal/forensics/redaction" + "go-agent-harness/internal/rollout" + runstore "go-agent-harness/internal/store" +) + +type failingNonTerminalStatusStore struct { + *runstore.MemoryStore +} + +func (s *failingNonTerminalStatusStore) UpdateRun(ctx context.Context, run *runstore.Run) error { + if run.Status == runstore.RunStatusRunning { + return errors.New("nonterminal status persistence failed") + } + return s.MemoryStore.UpdateRun(ctx, run) +} + +func TestNonTerminalStatusPersistenceFailureKeepsLiveStateMoving(t *testing.T) { + store := &failingNonTerminalStatusStore{MemoryStore: runstore.NewMemoryStore()} + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{Store: store}) + t.Cleanup(func() { _ = runner.Shutdown(context.Background()) }) + const runID = "nonterminal-status-write-failure" + run := Run{ID: runID, ConversationID: "nonterminal-status-write-failure-conv", Status: RunStatusQueued} + runner.runs[runID] = &runState{run: run, subscribers: make(map[chan Event]struct{})} + if err := store.CreateRun(context.Background(), runToStoreRun(run)); err != nil { + t.Fatalf("CreateRun: %v", err) + } + + runner.setStatus(runID, RunStatusRunning, "", "") + + current, ok := runner.GetRun(runID) + if !ok { + t.Fatal("run disappeared") + } + if current.Status != RunStatusRunning { + t.Fatalf("in-memory status = %q, want running despite best-effort store failure", current.Status) + } + stored, err := store.GetRun(context.Background(), runID) + if err != nil { + t.Fatalf("GetRun: %v", err) + } + if stored.Status != runstore.RunStatusQueued { + t.Fatalf("durable status = %q, want queued after injected write failure", stored.Status) + } +} + +func TestTerminalAppendDoesNotBlockUnrelatedConversationOrAllowSameConversationOvertake(t *testing.T) { + store := &blockingTerminalAppendStore{ + Store: runstore.NewMemoryStore(), + started: make(chan struct{}), + release: make(chan struct{}), + } + var releaseOnce sync.Once + releaseStore := func() { releaseOnce.Do(func() { close(store.release) }) } + t.Cleanup(releaseStore) + + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{Store: store}) + const ( + terminalRunID = "blocked-terminal-append" + sameRunID = "same-conversation-later" + otherRunID = "unrelated-conversation" + terminalConv = "terminal-conversation" + ) + for runID, convID := range map[string]string{ + terminalRunID: terminalConv, + sameRunID: terminalConv, + otherRunID: "other-conversation", + } { + run := Run{ID: runID, ConversationID: convID, Status: RunStatusRunning} + runner.runs[runID] = &runState{run: run, subscribers: make(map[chan Event]struct{})} + runner.storeCreateRun(run) + } + + history, stream, unsubscribe, err := runner.SubscribeConversation(terminalConv) + if err != nil { + t.Fatalf("SubscribeConversation: %v", err) + } + defer unsubscribe() + if len(history) != 0 { + t.Fatalf("initial history=%v, want empty", terminalAtomicityEventTypes(history)) + } + + store.setBlockRunID(terminalRunID) + terminalDone := make(chan bool, 1) + go func() { + terminalDone <- runner.transitionTerminal( + terminalRunID, RunStatusCompleted, "done", "", + EventRunCompleted, map[string]any{"output": "done"}, + ) + }() + select { + case <-store.started: + case <-time.After(2 * time.Second): + t.Fatal("terminal AppendEvent did not block") + } + + otherDone := make(chan struct{}) + go func() { + runner.emit(otherRunID, EventAssistantMessage, map[string]any{"content": "other"}) + close(otherDone) + }() + select { + case <-otherDone: + case <-time.After(200 * time.Millisecond): + t.Fatal("unrelated conversation emit blocked behind terminal AppendEvent") + } + + sameStarted := make(chan struct{}) + sameDone := make(chan struct{}) + go func() { + close(sameStarted) + runner.emit(sameRunID, EventAssistantMessage, map[string]any{"content": "later"}) + close(sameDone) + }() + <-sameStarted + select { + case <-sameDone: + t.Fatal("same-conversation event overtook blocked terminal AppendEvent") + default: + } + + releaseStore() + if won := <-terminalDone; !won { + t.Fatal("terminal transition did not commit") + } + select { + case <-sameDone: + case <-time.After(2 * time.Second): + t.Fatal("same-conversation event stayed blocked after terminal publication") + } + + got := make([]Event, 0, 2) + for len(got) < 2 { + select { + case event := <-stream: + got = append(got, event) + case <-time.After(2 * time.Second): + t.Fatalf("conversation stream=%v, want terminal then later", terminalAtomicityEventTypes(got)) + } + } + if got[0].Type != EventRunCompleted || got[1].Type != EventAssistantMessage { + t.Fatalf("conversation order=%v, want run.completed then assistant.message", + terminalAtomicityEventTypes(got)) + } +} + +func TestTerminalAppendFailureNeverPersistsTerminalStatusButStillPublishesInMemory(t *testing.T) { + store := &terminalFailureStore{ + Store: runstore.NewMemoryStore(), + failAppendEvent: true, + } + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{Store: store}) + run := Run{ID: "append-failure", ConversationID: "append-failure-conv", Status: RunStatusRunning} + runner.runs[run.ID] = &runState{run: run, subscribers: make(map[chan Event]struct{})} + runner.storeCreateRun(run) + history, stream, unsubscribe, err := runner.Subscribe(run.ID) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsubscribe() + if len(history) != 0 { + t.Fatalf("initial history=%v, want empty", terminalAtomicityEventTypes(history)) + } + if won := runner.transitionTerminal( + run.ID, RunStatusCompleted, "done", "", + EventRunCompleted, map[string]any{"output": "done"}, + ); !won { + t.Fatal("terminal transition did not commit after append failure") + } + requireTerminalSubscriberEvent(t, stream, EventRunCompleted) + current, _ := runner.GetRun(run.ID) + if current.Status != RunStatusCompleted { + t.Fatalf("in-memory status=%s, want completed after append failure", current.Status) + } + + stored, err := store.Store.GetRun(context.Background(), run.ID) + if err != nil { + t.Fatalf("store.GetRun: %v", err) + } + if stored.Status == runstore.RunStatusCompleted { + t.Fatal("durable terminal status was written after terminal AppendEvent failed") + } + replayHistory, _, replayUnsubscribe, err := runner.Subscribe(run.ID) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + replayUnsubscribe() + if !containsEventType(replayHistory, EventRunCompleted) { + t.Fatal("in-memory replay did not publish run.completed after bounded append failure") + } +} + +func TestTerminalStatusUpdateFailureLeavesDurableRunNonTerminalAndPublishesInMemory(t *testing.T) { + store := &terminalFailureStore{ + Store: runstore.NewMemoryStore(), + failUpdateRun: true, + } + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{Store: store}) + run := Run{ID: "status-failure", ConversationID: "status-failure-conv", Status: RunStatusRunning} + runner.runs[run.ID] = &runState{run: run, subscribers: make(map[chan Event]struct{})} + runner.storeCreateRun(run) + _, stream, unsubscribe, err := runner.Subscribe(run.ID) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsubscribe() + if won := runner.transitionTerminal( + run.ID, RunStatusCompleted, "done", "", + EventRunCompleted, map[string]any{"output": "done"}, + ); !won { + t.Fatal("terminal transition did not commit after status update failure") + } + requireTerminalSubscriberEvent(t, stream, EventRunCompleted) + current, _ := runner.GetRun(run.ID) + if current.Status != RunStatusCompleted { + t.Fatalf("in-memory status=%s, want completed after status update failure", current.Status) + } + + stored, err := store.Store.GetRun(context.Background(), run.ID) + if err != nil { + t.Fatalf("store.GetRun: %v", err) + } + if stored.Status == runstore.RunStatusCompleted { + t.Fatal("durable run unexpectedly became terminal after UpdateRun failure") + } + events, err := store.Store.GetEvents(context.Background(), run.ID, -1) + if err != nil { + t.Fatalf("store.GetEvents: %v", err) + } + foundTerminal := false + for _, event := range events { + if event.EventType == string(EventRunCompleted) { + foundTerminal = true + } + } + if !foundTerminal { + t.Fatal("durable terminal event missing when only UpdateRun failed") + } +} + +func TestDelayedNonTerminalStatusCannotOverwriteTerminalTransition(t *testing.T) { + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{}) + const runID = "status-vs-terminal" + runner.runs[runID] = &runState{ + run: Run{ID: runID, ConversationID: "status-vs-terminal-conv", Status: RunStatusRunning}, + subscribers: make(map[chan Event]struct{}), + } + + waitingPrepared := make(chan struct{}) + releaseWaiting := make(chan struct{}) + runner.statusBeforeCommitHook = func(gotRunID string, status RunStatus) { + if gotRunID == runID && status == RunStatusWaitingForUser { + close(waitingPrepared) + <-releaseWaiting + } + } + waitingDone := make(chan struct{}) + go func() { + runner.setStatus(runID, RunStatusWaitingForUser, "", "") + close(waitingDone) + }() + <-waitingPrepared + + terminalAttempted := make(chan struct{}) + runner.terminalTransitionHook = func(gotRunID string, status RunStatus, _ EventType) { + if gotRunID == runID && status == RunStatusCompleted { + close(terminalAttempted) + } + } + terminalDone := make(chan bool, 1) + go func() { + terminalDone <- runner.transitionTerminal( + runID, RunStatusCompleted, "done", "", EventRunCompleted, map[string]any{"output": "done"}, + ) + }() + <-terminalAttempted + close(releaseWaiting) + <-waitingDone + if won := <-terminalDone; !won { + t.Fatal("terminal transition did not commit") + } + + current, ok := runner.GetRun(runID) + if !ok { + t.Fatal("run disappeared") + } + if current.Status != RunStatusCompleted { + t.Fatalf("delayed non-terminal status overwrote terminal status: got %s", current.Status) + } +} + +func TestTerminalStatusUpdateTimeoutStillPublishesAndDoesNotBlockUnrelatedConversation(t *testing.T) { + store := &contextBlockingTerminalStatusStore{ + Store: runstore.NewMemoryStore(), + started: make(chan struct{}), + forceRelease: make(chan struct{}), + } + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{Store: store}) + runner.terminalStoreTimeout = 25 * time.Millisecond + const ( + terminalRunID = "terminal-status-timeout" + otherRunID = "terminal-status-timeout-other" + ) + for runID, convID := range map[string]string{ + terminalRunID: "terminal-status-timeout-conv", + otherRunID: "terminal-status-timeout-other-conv", + } { + run := Run{ID: runID, ConversationID: convID, Status: RunStatusRunning} + runner.runs[runID] = &runState{run: run, subscribers: make(map[chan Event]struct{})} + runner.storeCreateRun(run) + } + store.blockRunID = terminalRunID + _, terminalStream, unsubscribe, err := runner.Subscribe(terminalRunID) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsubscribe() + + terminalDone := make(chan bool, 1) + go func() { + terminalDone <- runner.transitionTerminal( + terminalRunID, RunStatusCompleted, "done", "", + EventRunCompleted, map[string]any{"output": "done"}, + ) + }() + select { + case <-store.started: + case <-time.After(2 * time.Second): + t.Fatal("terminal UpdateRun did not block") + } + + otherDone := make(chan struct{}) + go func() { + runner.emit(otherRunID, EventAssistantMessage, map[string]any{"content": "other"}) + close(otherDone) + }() + select { + case <-otherDone: + case <-time.After(200 * time.Millisecond): + t.Fatal("unrelated conversation blocked behind terminal UpdateRun") + } + + select { + case won := <-terminalDone: + if !won { + t.Fatal("terminal transition did not publish after UpdateRun timeout") + } + case <-time.After(200 * time.Millisecond): + close(store.forceRelease) + t.Fatal("terminal transition did not recover from bounded UpdateRun timeout") + } + current, _ := runner.GetRun(terminalRunID) + if current.Status != RunStatusCompleted { + t.Fatalf("in-memory status=%s, want completed after UpdateRun timeout", current.Status) + } + requireTerminalSubscriberEvent(t, terminalStream, EventRunCompleted) +} + +func TestRedactedTerminalWaitsForRecorderDrainBeforeStatus(t *testing.T) { + pipeline := redaction.NewPipeline( + redaction.NewRedactor(nil), + redaction.EventClassConfig{string(EventRunCompleted): redaction.StorageModeNone}, + ) + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{RedactionPipeline: pipeline}) + const runID = "redacted-recorder-drain" + recorderDone := make(chan struct{}) + recorderClosed := make(chan struct{}) + var closeOnce sync.Once + runner.runs[runID] = &runState{ + run: Run{ID: runID, ConversationID: "redacted-recorder-drain-conv", Status: RunStatusRunning}, + subscribers: make(map[chan Event]struct{}), + recorderCh: make(chan rollout.RecordableEvent, 1), + recorderDone: recorderDone, + closeRecorderOnce: func() { + closeOnce.Do(func() { + close(recorderClosed) + }) + }, + } + + terminalDone := make(chan bool, 1) + go func() { + terminalDone <- runner.transitionTerminal( + runID, RunStatusCompleted, "done", "", EventRunCompleted, map[string]any{"output": "done"}, + ) + }() + select { + case <-recorderClosed: + case <-time.After(2 * time.Second): + t.Fatal("redacted terminal did not close recorder") + } + current, _ := runner.GetRun(runID) + if current.Status == RunStatusCompleted { + close(recorderDone) + <-terminalDone + t.Fatal("redacted terminal status became visible before recorder drain") + } + close(recorderDone) + if won := <-terminalDone; !won { + t.Fatal("redacted terminal transition did not commit") + } +} + +func TestConversationSequenceLockReclaimedAfterContention(t *testing.T) { + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{}) + unlockFirst := runner.lockConversationSequence("reclaim") + waiterDone := make(chan struct{}) + go func() { + unlockWaiter := runner.lockConversationSequence("reclaim") + unlockWaiter() + close(waiterDone) + }() + waitForConversationSequenceRefs(t, runner, "reclaim", 2) + unlockFirst() + select { + case <-waiterDone: + case <-time.After(2 * time.Second): + t.Fatal("conversation sequence waiter did not finish") + } + + runner.conversationSequenceMu.Lock() + remaining := len(runner.conversationSequence) + runner.conversationSequenceMu.Unlock() + if remaining != 0 { + t.Fatalf("conversation sequence entries=%d, want 0 after owners and waiters release", remaining) + } + + for i := 0; i < 100; i++ { + unlock := runner.lockConversationSequence(fmt.Sprintf("distinct-%d", i)) + unlock() + } + runner.conversationSequenceMu.Lock() + remaining = len(runner.conversationSequence) + runner.conversationSequenceMu.Unlock() + if remaining != 0 { + t.Fatalf("conversation sequence entries=%d, want 0 after distinct-key releases", remaining) + } +} + +func waitForConversationSequenceRefs(t *testing.T, runner *Runner, key string, want int) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for { + runner.conversationSequenceMu.Lock() + entry := runner.conversationSequence[key] + got := 0 + if entry != nil { + got = entry.refs + } + runner.conversationSequenceMu.Unlock() + if got == want { + return + } + if time.Now().After(deadline) { + t.Fatalf("conversation sequence refs=%d, want %d", got, want) + } + time.Sleep(time.Millisecond) + } +} + +func requireTerminalSubscriberEvent(t *testing.T, stream <-chan Event, want EventType) { + t.Helper() + select { + case event := <-stream: + if event.Type != want { + t.Fatalf("subscriber event=%s, want %s", event.Type, want) + } + case <-time.After(2 * time.Second): + t.Fatalf("timed out waiting for subscriber event %s", want) + } +} + +type terminalFailureStore struct { + runstore.Store + failAppendEvent bool + failUpdateRun bool +} + +func (s *terminalFailureStore) AppendEvent(ctx context.Context, event *runstore.Event) error { + if s.failAppendEvent && IsTerminalEvent(EventType(event.EventType)) { + return errors.New("terminal append failed") + } + return s.Store.AppendEvent(ctx, event) +} + +func (s *terminalFailureStore) UpdateRun(ctx context.Context, run *runstore.Run) error { + if s.failUpdateRun && isTerminalStoreStatus(run.Status) { + return errors.New("terminal status update failed") + } + return s.Store.UpdateRun(ctx, run) +} + +type contextBlockingTerminalStatusStore struct { + runstore.Store + blockRunID string + started chan struct{} + startedOnce sync.Once + forceRelease chan struct{} +} + +func (s *contextBlockingTerminalStatusStore) UpdateRun(ctx context.Context, run *runstore.Run) error { + if run.ID == s.blockRunID && isTerminalStoreStatus(run.Status) { + s.startedOnce.Do(func() { close(s.started) }) + select { + case <-ctx.Done(): + return ctx.Err() + case <-s.forceRelease: + return errors.New("forced release") + } + } + return s.Store.UpdateRun(ctx, run) +} + +func isTerminalStoreStatus(status runstore.RunStatus) bool { + return status == runstore.RunStatusCompleted || + status == runstore.RunStatusFailed || + status == runstore.RunStatus("cancelled") +} diff --git a/internal/harness/runner_terminal_store_test.go b/internal/harness/runner_terminal_store_test.go index ca4cbbe2..cedfbec2 100644 --- a/internal/harness/runner_terminal_store_test.go +++ b/internal/harness/runner_terminal_store_test.go @@ -48,6 +48,16 @@ func TestTerminalStoreAppendDoesNotBlockRunnerQueries(t *testing.T) { t.Fatal("timed out waiting for terminal event store append to block") } + blockedRun, ok := runner.GetRun(doneRun.ID) + if !ok { + t.Fatalf("GetRun(%q) returned false during terminal append", doneRun.ID) + } + if isTerminalRunStatus(blockedRun.Status) { + close(backingStore.release) + close(holdProviderRelease) + t.Fatalf("GetRun exposed %s before terminal event store append completed", blockedRun.Status) + } + queryDone := make(chan struct{}) var found bool go func() { @@ -95,6 +105,113 @@ func TestTerminalStoreAppendUsesBoundedContext(t *testing.T) { } } +func TestTerminalStatusStoreUpdateDoesNotBlockOtherEventJournals(t *testing.T) { + holdProviderRelease := make(chan struct{}) + doneProviderRelease := make(chan struct{}) + provider := &promptGateProvider{ + gates: map[string]<-chan struct{}{ + "hold": holdProviderRelease, + "done": doneProviderRelease, + }, + } + backingStore := &blockingTerminalStatusStore{ + Store: runstore.NewMemoryStore(), + started: make(chan struct{}), + release: make(chan struct{}), + } + runner := NewRunner(provider, NewRegistry(), RunnerConfig{Store: backingStore}) + + holdRun, err := runner.StartRun(RunRequest{Prompt: "hold"}) + if err != nil { + t.Fatalf("StartRun hold: %v", err) + } + waitForStatus(t, runner, holdRun.ID, RunStatusRunning) + + doneRun, err := runner.StartRun(RunRequest{Prompt: "done"}) + if err != nil { + t.Fatalf("StartRun done: %v", err) + } + history, terminalStream, unsubscribe, err := runner.Subscribe(doneRun.ID) + if err != nil { + t.Fatalf("Subscribe done: %v", err) + } + defer unsubscribe() + for _, event := range history { + if IsTerminalEvent(event.Type) { + t.Fatalf("done run unexpectedly terminal before provider release: %s", event.Type) + } + } + backingStore.setBlockRunID(doneRun.ID) + close(doneProviderRelease) + + select { + case <-backingStore.started: + case <-time.After(2 * time.Second): + close(holdProviderRelease) + t.Fatal("timed out waiting for terminal status store update to block") + } + + blockedRun, ok := runner.GetRun(doneRun.ID) + if !ok { + t.Fatalf("GetRun(%q) returned false during terminal status persistence", doneRun.ID) + } + if isTerminalRunStatus(blockedRun.Status) { + t.Fatalf("GetRun exposed %s before terminal status persistence completed", blockedRun.Status) + } + for { + select { + case event := <-terminalStream: + if IsTerminalEvent(event.Type) { + t.Fatalf("terminal event %s fanned out before final status persistence completed", event.Type) + } + default: + goto streamDrained + } + } + +streamDrained: + + emitDone := make(chan struct{}) + go func() { + runner.emit(holdRun.ID, EventAssistantMessage, map[string]any{"content": "still responsive"}) + close(emitDone) + }() + select { + case <-emitDone: + case <-time.After(200 * time.Millisecond): + close(backingStore.release) + close(holdProviderRelease) + <-emitDone + t.Fatal("unrelated event journal blocked behind terminal status store update") + } + + close(backingStore.release) + deadline := time.After(2 * time.Second) + for { + select { + case event := <-terminalStream: + if event.Type == EventRunCompleted { + goto terminalDelivered + } + case <-deadline: + t.Fatal("terminal event was not delivered after final status persistence completed") + } + } + +terminalDelivered: + storedRun, err := backingStore.Store.GetRun(context.Background(), doneRun.ID) + if err != nil { + t.Fatalf("store.GetRun(%q): %v", doneRun.ID, err) + } + if storedRun.Status != runstore.RunStatusCompleted || storedRun.Output != "ok: done" { + t.Fatalf("stored run after terminal fanout = status %q output %q, want completed/ok: done", + storedRun.Status, storedRun.Output) + } + close(holdProviderRelease) + waitForStatus(t, runner, doneRun.ID, RunStatusCompleted) + waitForStatus(t, runner, holdRun.ID, RunStatusCompleted) +} + type promptGateProvider struct { gates map[string]<-chan struct{} } @@ -135,6 +252,37 @@ type deadlineRecordingStore struct { deadline chan time.Duration } +type blockingTerminalStatusStore struct { + runstore.Store + + mu sync.Mutex + blockRunID string + once sync.Once + started chan struct{} + release chan struct{} +} + +func (s *blockingTerminalStatusStore) setBlockRunID(runID string) { + s.mu.Lock() + defer s.mu.Unlock() + s.blockRunID = runID +} + +func (s *blockingTerminalStatusStore) UpdateRun(ctx context.Context, run *runstore.Run) error { + s.mu.Lock() + blockRunID := s.blockRunID + s.mu.Unlock() + if run.ID == blockRunID && run.Status == runstore.RunStatusCompleted { + s.once.Do(func() { close(s.started) }) + select { + case <-s.release: + case <-ctx.Done(): + return ctx.Err() + } + } + return s.Store.UpdateRun(ctx, run) +} + func (s *deadlineRecordingStore) AppendEvent(ctx context.Context, event *runstore.Event) error { if event.EventType == string(EventRunCompleted) { if deadline, ok := ctx.Deadline(); ok { diff --git a/internal/harness/runner_test.go b/internal/harness/runner_test.go index ab66ca52..dc4acefc 100644 --- a/internal/harness/runner_test.go +++ b/internal/harness/runner_test.go @@ -739,8 +739,13 @@ func TestRunnerFailedRunIncludesPartialUsageTotals(t *testing.T) { } } +// collectRunEvents preserves terminal event/history collection and also waits +// for the independently published terminal status. Event-first publication may +// expose replay before status; most callers use this helper as a settled-run +// boundary. The phase-ordering tests use direct Subscribe barriers instead. func collectRunEvents(t *testing.T, runner *Runner, runID string) ([]Event, error) { t.Helper() + deadline := time.Now().Add(4 * time.Second) history, stream, cancel, err := runner.Subscribe(runID) if err != nil { @@ -748,28 +753,112 @@ func collectRunEvents(t *testing.T, runner *Runner, runID string) ([]Event, erro } defer cancel() + return collectSubscribedRunEvents(runner, runID, history, stream, deadline, nil) +} + +func collectSubscribedRunEvents( + runner *Runner, + runID string, + history []Event, + stream <-chan Event, + deadline time.Time, + settlementStarted func(), +) ([]Event, error) { events := append([]Event(nil), history...) if hasTerminalEvent(events) { - return events, nil + return settleCollectedRunEvents(runner, runID, events, deadline, settlementStarted) } - timeout := time.After(4 * time.Second) + timeout := time.NewTimer(time.Until(deadline)) + defer timeout.Stop() for { select { case ev, ok := <-stream: if !ok { - return events, nil + return settleCollectedRunEvents(runner, runID, events, deadline, settlementStarted) } events = append(events, ev) if IsTerminalEvent(ev.Type) { - return events, nil + return settleCollectedRunEvents(runner, runID, events, deadline, settlementStarted) } - case <-timeout: + case <-timeout.C: return nil, context.DeadlineExceeded } } } +func settleCollectedRunEvents( + runner *Runner, + runID string, + events []Event, + deadline time.Time, + settlementStarted func(), +) ([]Event, error) { + if settlementStarted != nil { + settlementStarted() + } + eventStatus, err := collectedTerminalRunStatus(events) + if err != nil { + return events, fmt.Errorf("run %q terminal settlement: %w", runID, err) + } + for { + run, ok := runner.GetRun(runID) + if !ok { + return events, fmt.Errorf("run %q disappeared before terminal status settled", runID) + } + if isTerminalRunStatus(run.Status) { + if run.Status != eventStatus { + return events, fmt.Errorf( + "run %q terminal status %q does not match collected terminal event status %q; events=%v", + runID, + run.Status, + eventStatus, + eventTypes(events), + ) + } + return events, nil + } + remaining := time.Until(deadline) + if remaining <= 0 { + return events, fmt.Errorf( + "run %q terminal event collected before status settled: %w", + runID, + context.DeadlineExceeded, + ) + } + wait := 5 * time.Millisecond + if remaining < wait { + wait = remaining + } + time.Sleep(wait) + } +} + +func collectedTerminalRunStatus(events []Event) (RunStatus, error) { + var terminalStatus RunStatus + for _, event := range events { + var status RunStatus + switch event.Type { + case EventRunCompleted: + status = RunStatusCompleted + case EventRunFailed: + status = RunStatusFailed + case EventRunCancelled: + status = RunStatusCancelled + default: + continue + } + if terminalStatus != "" { + return "", fmt.Errorf("collected multiple terminal events: %v", eventTypes(events)) + } + terminalStatus = status + } + if terminalStatus == "" { + return "", fmt.Errorf("collected events contain no terminal event: %v", eventTypes(events)) + } + return terminalStatus, nil +} + func hasTerminalEvent(events []Event) bool { for _, ev := range events { if IsTerminalEvent(ev.Type) { @@ -1999,6 +2088,11 @@ func TestRunnerFailsWhenClientFactoryErrors_NoFallback(t *testing.T) { if err != nil { t.Fatalf("collect events: %v", err) } + // A terminal event can be present in replay while its matching status is + // still completing the event-first publication sequence. Wait for the + // independently asserted status instead of treating replay as the inverse + // of the public status-implies-replay guarantee. + waitForStatus(t, runner, run.ID, RunStatusFailed) state, ok := runner.GetRun(run.ID) if !ok { diff --git a/internal/harness/types.go b/internal/harness/types.go index 51661e12..acb1a2de 100644 --- a/internal/harness/types.go +++ b/internal/harness/types.go @@ -623,7 +623,9 @@ type RunnerConfig struct { // immediately (the legacy unbounded behaviour). WorkerPoolSize int // MaxCompletedRetention caps completed/failed/cancelled run states retained - // in memory after terminal events are persisted and subscribers drain. + // in memory after terminal event and status durability are resolved and + // subscribers drain. When unresolved store-backed terminal durability reaches + // this cap, new run admissions fail closed until status persistence recovers. // 0 uses the default retention window. MaxCompletedRetention int // MaxConversationRetention caps the in-memory conversation transcript mirror. diff --git a/internal/server/http_runs.go b/internal/server/http_runs.go index 91326df3..d327ab30 100644 --- a/internal/server/http_runs.go +++ b/internal/server/http_runs.go @@ -64,6 +64,11 @@ func (s *Server) handlePostRun(w http.ResponseWriter, r *http.Request) { run, err := s.runner.StartRun(req) if err != nil { + var durabilityErr *harness.TerminalDurabilityBackpressureError + if errors.As(err, &durabilityErr) { + writeError(w, http.StatusServiceUnavailable, "terminal_durability_unavailable", err.Error()) + return + } writeError(w, http.StatusBadRequest, "invalid_request", err.Error()) return } @@ -703,6 +708,11 @@ func (s *Server) handleRunContinue(w http.ResponseWriter, r *http.Request, runID Permissions: req.Permissions, }) if err != nil { + var durabilityErr *harness.TerminalDurabilityBackpressureError + if errors.As(err, &durabilityErr) { + writeError(w, http.StatusServiceUnavailable, "terminal_durability_unavailable", err.Error()) + return + } if errors.Is(err, harness.ErrRunNotFound) { writeError(w, http.StatusNotFound, "not_found", fmt.Sprintf("run %q not found", runID)) return diff --git a/internal/server/http_terminal_atomicity_test.go b/internal/server/http_terminal_atomicity_test.go new file mode 100644 index 00000000..59532043 --- /dev/null +++ b/internal/server/http_terminal_atomicity_test.go @@ -0,0 +1,223 @@ +package server + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "go-agent-harness/internal/harness" + runstore "go-agent-harness/internal/store" +) + +func TestTerminalStatusPollImmediatelyReplaysMatchingTerminalEvent(t *testing.T) { + tests := []struct { + name string + provider *terminalHTTPProvider + wantStatus harness.RunStatus + wantEvent harness.EventType + cancel bool + }{ + { + name: "completed", + provider: &terminalHTTPProvider{result: harness.CompletionResult{Content: "done"}}, + wantStatus: harness.RunStatusCompleted, + wantEvent: harness.EventRunCompleted, + }, + { + name: "failed", + provider: &terminalHTTPProvider{err: errors.New("provider unavailable")}, + wantStatus: harness.RunStatusFailed, + wantEvent: harness.EventRunFailed, + }, + { + name: "cancelled", + provider: &terminalHTTPProvider{started: make(chan struct{}), hang: true}, + wantStatus: harness.RunStatusCancelled, + wantEvent: harness.EventRunCancelled, + cancel: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + runner := harness.NewRunner(tt.provider, harness.NewRegistry(), harness.RunnerConfig{ + DefaultModel: "test-model", + MaxSteps: 1, + }) + ts := httptest.NewServer(New(runner)) + t.Cleanup(func() { + ts.Close() + _ = runner.Shutdown(context.Background()) + }) + + runID := startTerminalHTTPRun(t, ts) + if tt.cancel { + select { + case <-tt.provider.started: + case <-time.After(2 * time.Second): + t.Fatal("provider did not start") + } + res, err := http.Post(ts.URL+"/v1/runs/"+runID+"/cancel", "application/json", nil) + if err != nil { + t.Fatalf("POST cancel: %v", err) + } + _ = res.Body.Close() + if res.StatusCode != http.StatusOK { + t.Fatalf("POST cancel status=%d, want %d", res.StatusCode, http.StatusOK) + } + } + + if got := waitForRunStatus(t, ts, runID, string(tt.wantStatus)); got != string(tt.wantStatus) { + t.Fatalf("terminal status=%s, want %s", got, tt.wantStatus) + } + + req, err := http.NewRequest(http.MethodGet, ts.URL+"/v1/runs/"+runID+"/events", nil) + if err != nil { + t.Fatalf("build reconnect request: %v", err) + } + req.Header.Set("Last-Event-ID", runID+":0") + res, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("GET event replay: %v", err) + } + body, readErr := io.ReadAll(res.Body) + _ = res.Body.Close() + if readErr != nil { + t.Fatalf("read event replay: %v", readErr) + } + if res.StatusCode != http.StatusOK { + t.Fatalf("GET event replay status=%d body=%s", res.StatusCode, body) + } + + replay := string(body) + matchingFrame := "event: " + string(tt.wantEvent) + "\n" + if strings.Count(replay, matchingFrame) != 1 { + t.Fatalf("terminal status %s replay count for %s=%d, want 1; replay=%s", + tt.wantStatus, tt.wantEvent, strings.Count(replay, matchingFrame), replay) + } + for _, other := range []harness.EventType{ + harness.EventRunCompleted, + harness.EventRunFailed, + harness.EventRunCancelled, + } { + if other != tt.wantEvent && strings.Contains(replay, "event: "+string(other)+"\n") { + t.Fatalf("terminal status %s replay included mismatched event %s; replay=%s", + tt.wantStatus, other, replay) + } + } + }) + } +} + +func TestTerminalDurabilityBackpressureMapsStartAndContinueToServiceUnavailable(t *testing.T) { + store := &httpTerminalStatusFailStore{Store: runstore.NewMemoryStore()} + runner := harness.NewRunner(&terminalHTTPProvider{ + result: harness.CompletionResult{Content: "done"}, + }, harness.NewRegistry(), harness.RunnerConfig{ + DefaultModel: "test-model", + MaxSteps: 1, + MaxCompletedRetention: 1, + Store: store, + }) + ts := httptest.NewServer(New(runner)) + t.Cleanup(func() { + ts.Close() + _ = runner.Shutdown(context.Background()) + }) + + runID := startTerminalHTTPRun(t, ts) + waitForRunStatus(t, ts, runID, string(harness.RunStatusCompleted)) + + tests := []struct { + name string + url string + }{ + {name: "start", url: ts.URL + "/v1/runs"}, + {name: "continue", url: ts.URL + "/v1/runs/" + runID + "/continue"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + res, err := http.Post(tt.url, "application/json", bytes.NewBufferString(`{"prompt":"blocked admission"}`)) + if err != nil { + t.Fatalf("POST: %v", err) + } + defer res.Body.Close() + var response struct { + Error struct { + Code string `json:"code"` + } `json:"error"` + } + if err := json.NewDecoder(res.Body).Decode(&response); err != nil { + t.Fatalf("decode response: %v", err) + } + if res.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("status=%d code=%q, want 503", res.StatusCode, response.Error.Code) + } + if response.Error.Code != "terminal_durability_unavailable" { + t.Fatalf("error code=%q, want terminal_durability_unavailable", response.Error.Code) + } + }) + } +} + +type terminalHTTPProvider struct { + result harness.CompletionResult + err error + started chan struct{} + hang bool +} + +type httpTerminalStatusFailStore struct{ runstore.Store } + +func (s *httpTerminalStatusFailStore) UpdateRun(ctx context.Context, run *runstore.Run) error { + if run.Status == runstore.RunStatusCompleted || run.Status == runstore.RunStatusFailed || + run.Status == runstore.RunStatus("cancelled") { + return errors.New("terminal status persistence unavailable") + } + return s.Store.UpdateRun(ctx, run) +} + +func (p *terminalHTTPProvider) Complete(ctx context.Context, _ harness.CompletionRequest) (harness.CompletionResult, error) { + if p.started != nil { + select { + case <-p.started: + default: + close(p.started) + } + } + if p.hang { + <-ctx.Done() + return harness.CompletionResult{}, ctx.Err() + } + return p.result, p.err +} + +func startTerminalHTTPRun(t *testing.T, ts *httptest.Server) string { + t.Helper() + res, err := http.Post(ts.URL+"/v1/runs", "application/json", bytes.NewBufferString(`{"prompt":"terminal atomicity"}`)) + if err != nil { + t.Fatalf("POST run: %v", err) + } + defer res.Body.Close() + if res.StatusCode != http.StatusAccepted { + body, _ := io.ReadAll(res.Body) + t.Fatalf("POST run status=%d, want %d: %s", res.StatusCode, http.StatusAccepted, body) + } + var created struct { + RunID string `json:"run_id"` + } + if err := json.NewDecoder(res.Body).Decode(&created); err != nil { + t.Fatalf("decode run response: %v", err) + } + if created.RunID == "" { + t.Fatal("POST run returned empty run_id") + } + return created.RunID +}