diff --git a/docs/logs/engineering-log.md b/docs/logs/engineering-log.md index 94c3c48f..a2ca2ea9 100644 --- a/docs/logs/engineering-log.md +++ b/docs/logs/engineering-log.md @@ -129,6 +129,110 @@ complete `cmd/harnessd` normal/race suites passed; the repository regression gate passed normal, race, and coverage at 85.6% with zero uncovered functions. +## 2026-07-30 — Issue #1054 wait state precedes pending input + +- Symptom: Hosted race execution observed `waiting_for_user`, then + `PendingInput` returned `no pending input`. +- Cause: `runner_step_engine.go` publishes status/event before invoking the + AskUserQuestion tool; broker registration happens later inside the handler. +- Impact: Event-driven TUI/macOS clients can render a wait state with no + question available to display or submit. +- Intended fix: Let each broker notify after its pending state is + readable/durable, forward that notification through the core tool, and + publish the runner wait state from that point. +- TDD evidence: A gated broker made registration impossible while the tool was + entered; pre-fix status was already `waiting_for_user`, proving the gap. The + test now keeps status `running` until registration, then requires both + readable pending input and `waiting_for_user`. +- Implementation: `AskUserQuestionRequest.OnPending` is a typed, + post-registration notifier. Both built-in brokers start it exactly once with + the question's deadline context; the core tool forwards it from context; the + runner uses it to publish status and the existing event without polling. +- Verification: Focused AskUser/wait suites passed 100 normal and 100 race + repetitions; complete harness normal/race suites passed; repository normal, + race, and coverage gates passed at 85.6% with zero uncovered functions. +- Review follow-up: Exact-head Codex review correctly noted that both brokers + computed `DeadlineAt` before `OnPending` but started their timeout afterward. + Regressions that held notification beyond the deadline failed on both + backends, then passed after the timer/context moved before notification. + These deadline tests passed 10 normal and 10 race repetitions; complete + harness normal/race and repository normal/race/coverage gates passed again. +- Second review follow-up: Starting the clock was insufficient because a + notifier that never returned still prevented `Ask` from selecting the + expired timer. Strengthened regressions kept both notifiers blocked while + requiring `Ask` to return its timeout. Brokers now run notification + independently with the same deadline context, while answer/cancel/timeout + selection continues immediately; the runner checks that context before + status and event publication. +- Third review follow-up: Letting answer selection race notification created + the opposite ordering bug: a quick submission could emit `run.resumed` while + waiting-state persistence was still blocked, then cancel the notifier before + `run.waiting_for_user`. A deterministic blocking-store regression reproduced + the reversed event order. Brokers now wait for notification completion before + consuming a buffered answer, while the same deadline remains independently + enforceable if notification stalls. +- Fourth review follow-up: Deadline resolution still exposed two durable-state + races. A timely checkpoint answer could be overwritten as expired, and a + blocked stale `waiting_for_user` write could land after the terminal failure + write. Checkpoint resolution is now serialized and `ExpirePending` never + replaces an accepted result; run status persistence uses a monotonic in-memory + version and rewrites the latest state after any stale write completes. + Deterministic regressions cover both races, followed by the full normal, + race, and coverage gate at 85.6% with zero uncovered functions. +- Fifth review follow-up: The in-memory broker was not symmetric with the + checkpoint broker when a timely answer was buffered while notification hit + its deadline, and a losing checkpoint resume still returned false success. + In-memory submission now publishes the buffered answer before removing the + pending entry, deadline cleanup returns any accepted answer, and checkpoint + resolution returns exported `ErrAlreadyResolved` when another terminal + transition already won. Focused normal/race stress covers both contracts. +- Sixth review follow-up: The ordinary checkpoint wait deadline still bypassed + accepted-answer recovery, and the new sentinel escaped as HTTP 500 through + run input, approval/deny, and generic checkpoint resume paths. Both AskUser + deadline branches now share one pending-only expiry/recovery function; + broker/runner boundaries normalize lost races to their existing no-pending + contracts; and generic resume returns stable `409 already_resolved` without + changing status, payload, or update time. Deterministic gates cover accepted + resume-before-notify, approval/deny expiry races, repeated resume, and both + API error shapes without unbounded channel receives. +- Seventh review follow-up: Accepted-answer recovery at the notifier deadline + returned before the blocked pending-state publication completed, so + `run.resumed` could still overtake `run.waiting_for_user`. Deterministic + regressions now require both brokers to retain the accepted answer without + returning it until pending publication finishes. Unresolved timeout paths + remain independent, while accepted answers wait on notification completion + with the parent context as the cancellation escape hatch. +- Eighth review follow-up: A broader concurrency review found four remaining + ownership gaps. The built-in notifier used a background store context; + checkpoint resolution was process-local and service-wide; stale run writes + still depended on a fallible corrective retry; and third-party brokers could + omit `OnPending`. New deterministic reds pinned each failure. Status writes + are now serialized per run and snapshot after a context-aware lock; + notification passes its deadline through status and event persistence; + checkpoint stores expose atomic pending-only resolution with per-record, + context-aware service coordination and cross-service waiter observation; and + the runner observes readable broker pending state as a callback fallback. + Both callback and fallback paths share exactly-once wait publication. + Status mutation, persistence, and its lifecycle event share the per-run lock; + terminal state rejects any delayed nonterminal downgrade, so a notifier + cannot publish stale waiting state after completion, failure, or cancellation. +- Ninth review follow-up: Pending publication still used once-on-attempt and + the fallback observer cancelled its context as soon as the tool returned. + Immediate `UpdateRun` or `AppendEvent` failures could therefore consume the + only publication attempt, while a quick accepted answer could cancel an + observer already persisting the wait. The callback and observer now share a + serialized once-on-success publisher; started observer publication drains to + success or the question deadline, and transient failures retry. Strict + durable-before-visible event behavior is limited to this waiting lifecycle; + ordinary nonterminal events preserve the existing best-effort persistence + contract. Failed strict appends roll back the final sequence allocation, so + run SSE IDs remain contiguous and `Last-Event-ID` reconnect returns only + unseen events. A redaction-policy drop counts as successful suppression and + cannot cause retries or block an accepted answer. Cross-Service checkpoint + polling is opportunistic after local waiter registration: a transient poll + read error is retried instead of unregistering the waiter or masking a later + local/remote resolution; the caller context remains the termination bound. + ## 2026-07-30 (Workflow Failure-Event Test Timeout — Issue #1049) - Symptom: the full race gate reached a stored failed workflow state but timed diff --git a/docs/logs/long-term-thinking-log.md b/docs/logs/long-term-thinking-log.md index 4abb7509..8457c4e8 100644 --- a/docs/logs/long-term-thinking-log.md +++ b/docs/logs/long-term-thinking-log.md @@ -108,6 +108,42 @@ - Next verification step: Make the direct signal expectation fail first, emit it at the provider factory boundary, then run focused and full gates. +## 2026-07-30 (Issue #1054 Waiting/Pending Atomicity) + +- Command intent: Fix the exact hosted lifecycle failure blocking the + cron/callback repair chain. +- User intent: Harness and GUI state must agree in real time, not merely pass + source-level tests. +- Success definition: + - `PendingInput` succeeds whenever `waiting_for_user` is observable. + - Both in-memory and durable checkpoint brokers uphold the invariant. + - A resume accepted before a deadline remains accepted even when persistence + or pending notification completes after that deadline. + - An accepted answer cannot become a resumed run before pending-state + publication finishes, including when the notifier deadline wins selection. + - Lost resume/approval/deny races return stable no-pending or conflict + semantics at the harness and HTTP boundaries, never false success or 500. + - Resolution remains single-winner across Service instances sharing a + durable store, without serializing unrelated checkpoints or ignoring a + waiting caller's context. + - Pending publication honors its deadline through persistence, stale run + writes cannot overwrite terminal state, and callback-omitting brokers still + produce one visible wait/resume lifecycle after exposing readable pending + input. + - Pending publication is once-on-success across callback and observer: + transient status/event failures retry, an observer already publishing is + drained, deliberate redaction suppression completes without retry, and + failed strict appends cannot create an SSE cursor gap. + - Cross-Service waiter polling tolerates transient reads after registration; + local notification, a later durable read, or caller cancellation decides + the result rather than a single opportunistic poll failure. + - Event ordering, cancellation, timeout, and denied-call restoration remain + correct and race-clean. + - Full repository verification and final native GUI conversation proof pass. +- Non-goals: Redesigning structured questions or approval behavior. +- Next verification step: Add a deterministic gated-broker regression, confirm + the ordering bug, then introduce post-registration notification. + ## 2026-07-30 (Workflow Failure-Event Test Timeout — Issue #1049) - Command intent: clear the exact full-gate timeout blocking the verified diff --git a/docs/plans/2026-07-30-issue-1054-waiting-pending-order-impact-map.md b/docs/plans/2026-07-30-issue-1054-waiting-pending-order-impact-map.md new file mode 100644 index 00000000..8b13b97d --- /dev/null +++ b/docs/plans/2026-07-30-issue-1054-waiting-pending-order-impact-map.md @@ -0,0 +1,136 @@ +# Cross-Surface Impact Map: Waiting-for-user pending-input ordering + +## Task + +- Task / issue: #1054 +- Plan link: `2026-07-30-issue-1054-waiting-pending-order-plan.md` +- Owner: Codex +- Status: Implemented; promotion pending + +## Current Ownership, Callers, and Data Flow + +- Entry points: Runner tool-call lifecycle and `AskUserQuestionTool`. +- Owning packages/types/functions and source of truth: + `runner_step_engine.go` owns run status/events; in-memory and checkpoint + brokers own pending question state. +- Callers, consumers, events, and downstream data: HTTP/TUI/macOS clients read + run status/events and then call `PendingInput`. +- Similar abstractions searched: Approval brokers, `PendingInput`, + `EventRunWaitingForUser`, AskUserQuestion broker implementations. +- Search commands/evidence: + `rg -n "RunStatusWaitingForUser|PendingInput|AskUserQuestionBroker" internal/harness`. +- Duplication/ownership conclusion: Registration stays broker-owned; status and + event publication stays runner-owned, joined by a typed post-registration + callback. + +## Config, API, CLI, and Tools + +- User-facing config added or changed: None. +- Defaults / fallbacks: None. +- Environment variables, config files, or saved settings touched: None. +- Endpoints, request fields, response fields, or server wiring affected: + Existing pending-input endpoint becomes immediately consistent with visible + wait state; wire shape unchanged. +- CLI commands, tools, wire formats, or integrations affected: + AskUserQuestion internal request gains an optional readiness callback. +- Error states / validation changes: Removes transient `ErrNoPendingInput` + after visible wait state. + +## Persistence and Compatibility + +- Schemas, migrations, caches, generated data, or ownership changes: None. +- Backward/forward compatibility and versioning: Internal Go API addition; + existing question/checkpoint records unchanged. +- Partial rollout and mixed-version behavior: Single binary; no mixed-version + protocol. + +## Lifecycle, Security, and Reliability + +- Concurrency, cancellation, retries, cleanup, and resource ownership: + Notification starts exactly once after successful registration in a + deadline-bound goroutine. The broker does not consume a buffered answer until + notification finishes, which preserves wait-before-resume ordering, while + cancellation and unresolved timeout remain independent so a stalled notifier + cannot hang `Ask`. If deadline selection recovers an already-accepted answer, + it waits for notification completion or parent cancellation before returning + so `run.resumed` cannot overtake pending-state publication. The callback must + propagate its supplied context through blocking work. Built-in status/event + persistence does so; run-status mutation, persistence, and the matching event + serialize per run after acquiring that context-aware lock. Terminal state + rejects delayed nonterminal downgrades. A runner-side pending observer invokes + the same exactly-once notifier when a broker exposes `Pending` but omits the + callback. +- Authentication, authorization, permissions, trust, privacy, and secrets: + No new data exposure; callback receives the already-public pending shape. +- Failure modes, recovery, idempotency, and data repair: Registration failure + does not publish wait state; timeout/cancel continue through existing paths. + Checkpoint expiry is conditional on unresolved state, and stale run-status + writes cannot land after newer state because persistence is serialized per + run rather than repaired afterward. + Accepted in-memory answers win deadline cleanup, and callers receive an + explicit already-resolved error when a checkpoint transition loses a race. + AskUser notification and ordinary wait deadlines share the same atomic + pending-only expiry and accepted-resume recovery path. Harness input and + approval operations translate a lost resolution to their established + no-pending result; generic checkpoint HTTP resume exposes the distinct + durable conflict as `409 already_resolved` while preserving terminal data. + Memory and SQLite checkpoint stores perform atomic pending-only resolution; + per-record service locks honor caller cancellation, unrelated records proceed + independently, and bounded store observation wakes waiters after a different + Service instance resolves the record. + +## Product and Integration Surfaces + +- Server/runtime: Strengthened lifecycle invariant. +- TUI/web/macOS/other clients: Wait event/status can be rendered without a + retry gap before the question is readable. +- Provider/model/tool catalog and routing: AskUserQuestion core tool forwards + the readiness hook; no catalog changes. +- External systems and automation: Hosted race suite becomes deterministic. +- UX states, keyboard/focus/accessibility/motion: Eliminates transient empty + input UI; no visual design change. + +## Deployment and Operations + +- Deployment/migration order and feature flags: None. +- Logs, metrics, traces, alerts, and support diagnostics: Event payload/order + retained. +- Rollback triggers and recovery steps: Revert if exact-head lifecycle tests or + manual question flow regress. +- Runbooks and operator docs: None. + +## Regression Tests + +- Characterization and first expected red test: Gated broker proves visible wait + state precedes broker registration on current code. +- New acceptance tests required: Both brokers expose `Pending` inside readiness + callback; tool forwards callback; runner invariant. +- Edge, negative, failure, lifecycle, and security tests: Registration error, + timeout, cancellation, denied tool, exactly-once notification, and accepted + answer recovery while pending publication is blocked past its deadline; + callback deadline propagation; stale-write repair failure; callback omission; + per-record cancellation; unrelated resolution; cross-Service single-winner + resolution and waiter visibility; once-on-success retry for immediate status + and event persistence failures; observer drain after accepted input; + redaction-dropped waiting events; ordinary-event best-effort persistence; and + contiguous run SSE IDs with duplicate-free `Last-Event-ID` replay after a + failed strict append; and waiter survival across a transient cross-Service + polling read failure before later resolution. +- Integration/e2e/real-path proof: Exact-head hosted normal/race checks and + final native GUI conversation test. +- Cross-surface regressions to guard: Existing event order and status + restoration suites. +- Exact targeted and full commands: + `go test ./internal/harness ./internal/harness/tools/core -run 'AskUser|WaitForUser' -count=100`; + same with `-race`; `./scripts/test-regression.sh`. + +## Documentation and Handoff + +- Specs/public docs before code: Plan and impact map. +- Implementation notes/logs/indexes after code: Plan index, active plan, + engineering log, long-term-thinking log. +- Training/onboarding/release notes: None; existing public contract is tightened. + +## Warning Check + +- Every relevant surface is mapped or explicitly unaffected. diff --git a/docs/plans/2026-07-30-issue-1054-waiting-pending-order-plan.md b/docs/plans/2026-07-30-issue-1054-waiting-pending-order-plan.md new file mode 100644 index 00000000..5787da56 --- /dev/null +++ b/docs/plans/2026-07-30-issue-1054-waiting-pending-order-plan.md @@ -0,0 +1,137 @@ +# Plan: Publish waiting-for-user only after pending input exists + +## Context + +- Governing GitHub issue: #1054 +- Problem: The runner exposes `waiting_for_user` and its event before either + built-in AskUserQuestion broker has registered pending input. +- User impact: TUI and native GUI clients can react to the wait state and + receive a transient `no pending input`, leaving the question unavailable + until a retry. +- Constraints: Preserve tool/event order, support in-memory and durable + checkpoint brokers, and keep denial/cancellation/timeout paths bounded. + +## Scope + +- In scope: A typed post-registration notification on AskUserQuestion requests, + broker implementations, tool propagation, runner status/event publication, + lifecycle regressions, and documentation. +- Out of scope: Question/answer schema, approval broker behavior, and scheduler + semantics. + +## Documentation Contract + +- Feature status: `implemented` +- Public docs affected: None; this tightens an existing lifecycle contract. +- Spec docs to update before code: This plan and impact map. +- Implementation notes to add after code: Engineering and long-term-thinking + logs with TDD and verification evidence. + +## Test Plan (TDD) + +- New failing tests to add first: Gate broker registration and prove the current + runner publishes `waiting_for_user` while `PendingInput` still fails. +- Existing tests to update: In-memory/checkpoint broker lifecycle tests, core + AskUserQuestion tool forwarding test, and runner lifecycle order test. +- Regression tests required: Denied AskUserQuestion status restoration, + cancellation, timeout, event order, repeated normal/race focused suites, and + the complete repository gate. + +## Cross-Surface Impact Map + +- See `2026-07-30-issue-1054-waiting-pending-order-impact-map.md`. + +## Implementation Checklist + +- [x] Define acceptance criteria in tests. +- [x] Link a contract-complete structured GitHub issue before implementation. +- [x] Record current architecture, callers, consumers, and source-of-truth search evidence. +- [x] Document feature status and exact contract before code. +- [x] Complete and reconcile the cross-surface impact map before implementation. +- [x] Add characterization coverage before structural refactors. +- [x] Write failing tests first. +- [x] Review ownership/copy semantics for exported or state-storing types when mutable fields cross boundaries. +- [x] Implement minimal code changes. +- [x] Refactor while tests remain green. +- [x] Update docs, status ledgers, and indexes. +- [x] Update engineering/system/observational logs as needed. +- [x] Run full test suite. +- [ ] Merge branch back to `main` after tests pass. + +## Risks and Mitigations + +- Risk: A broker invokes the readiness hook before pending state is readable. +- Mitigation: Broker-level assertions call `Pending` inside the hook. +- Risk: A hook executes more than once or after cancellation. +- Mitigation: Each broker starts exactly one context-bound notifier immediately + after successful registration. Unresolved timeout handling never waits for a + stalled notifier; an already-accepted answer waits for notification + completion or parent cancellation before it is returned. + +## Verification + +- Expected red: deterministic gated-broker test observed + `waiting_for_user` before registration while `PendingInput` failed. +- Both broker backends prove pending input is readable inside `OnPending`; the + core tool proves typed notifier propagation. +- Focused AskUser/wait lifecycle suites passed 100 normal and 100 race + repetitions. +- Complete `internal/harness/...` normal and race suites passed. +- `./scripts/test-regression.sh` passed normal, race, and coverage with 85.6% + total coverage and zero uncovered functions. +- Exact-head review found that notification time was outside the timeout + countdown. New regressions blocked both broker notifiers beyond their + deadlines and failed until timers/contexts started before notification; each + passed 10 normal and 10 race repetitions, followed by complete harness and + repository gates. +- A second exact-head review strengthened the finding: merely starting the + clock still let a never-returning notifier hang `Ask`. The regressions now + require timeout while notification remains blocked. Brokers launch the typed + notifier with the same deadline context and continue waiting for + answer/cancellation independently. +- A third exact-head review found that fully independent answer selection let + `run.resumed` overtake blocked wait-event persistence. A blocking-store + regression reproduced that reverse order. Brokers now buffer quick answers + but do not consume them until notification finishes; cancellation and timeout + can still return independently if the notifier never completes. +- A fourth exact-head review found that timeout cleanup could overwrite a + checkpoint already resumed by a timely answer, while a late stale wait-status + write could overwrite the durable terminal status. New deterministic + regressions cover both cases. Checkpoint terminal transitions are serialized, + expiry is pending-only, and status persistence repairs stale writes using a + monotonic per-run version. +- A fifth exact-head review found asymmetric in-memory answer handling and + false-success reporting when checkpoint expiry won. Regressions now require a + timely buffered in-memory answer to survive notification deadline and require + a losing checkpoint resume to return `ErrAlreadyResolved`. +- A sixth independent review traced the sentinel across every caller. The final + contract requires both checkpoint AskUser deadline branches to recover an + already-accepted answer; run input and approval/deny races to retain their + existing no-pending API semantics; and generic checkpoint resume to return + `409 already_resolved` on repeated, expired, or denied records without + mutating the durable terminal snapshot. All concurrency tests use bounded + gates and receives. +- A seventh exact-head review found that accepted-answer recovery at the + notifier deadline could return while pending-state publication was still + blocked. New deterministic regressions require both brokers to preserve the + accepted answer without returning it until notification finishes, preventing + `run.resumed` from overtaking `run.waiting_for_user`. +- An independent concurrency review then required four final hardening + contracts: every notifier persistence/publication step honors the supplied + deadline context without converting accepted input into a synthetic timeout; + checkpoint resolution uses per-record coordination plus durable pending-only + CAS across Service instances; run-status persistence serializes per run and + snapshots after acquiring that lock; and a runner-side pending observer + supplies exactly-once wait visibility when a broker omits `OnPending`. +- Final hardening changed pending notification from once-on-attempt to + serialized once-on-success. Deterministic regressions require retry after an + immediate waiting `UpdateRun` or `AppendEvent` failure, drain an observer + whose publication began before the tool returned, and preserve exactly one + visible wait/resume. Strict event persistence is scoped to waiting + publication; ordinary events retain best-effort persistence. A failed strict + append rolls back its final sequence allocation so run SSE replay remains + contiguous and `Last-Event-ID` returns only unseen events. Intentional + redaction drops complete publication without retrying to the deadline. + Cross-Service waiter polling remains registered across transient store reads + and retries until it observes resolution, receives local notification, or + the caller context ends. diff --git a/docs/plans/INDEX.md b/docs/plans/INDEX.md index dc294af5..8c8d2b9a 100644 --- a/docs/plans/INDEX.md +++ b/docs/plans/INDEX.md @@ -12,6 +12,8 @@ - `2026-07-31-issue-1064-workflow-exit-precedence-impact-map.md` — Cross-surface impact map for Issue #1064. - `2026-07-30-issue-1052-provider-key-capture-sync-plan.md` — Issue #1052 direct provider-factory synchronization for the API-key capture regression. - `2026-07-30-issue-1052-provider-key-capture-sync-impact-map.md` — Cross-surface impact map for Issue #1052. +- `2026-07-30-issue-1054-waiting-pending-order-plan.md` — Issue #1054 post-registration waiting-for-user status/event publication. +- `2026-07-30-issue-1054-waiting-pending-order-impact-map.md` — Cross-surface impact map for Issue #1054. - `2026-07-30-issue-1049-workflow-failure-timeout-plan.md` — Issue #1049 planned contention-tolerant workflow failure-event regression wait. - `2026-07-30-issue-1049-workflow-failure-timeout-impact-map.md` — Cross-surface impact map for Issue #1049. - `2026-07-30-issue-1044-ask-status-race-plan.md` — Issue #1044 planned synchronization of the AskUserQuestion status regression fixture. diff --git a/docs/plans/active-plan.md b/docs/plans/active-plan.md index 5a814acc..a5bf68b1 100644 --- a/docs/plans/active-plan.md +++ b/docs/plans/active-plan.md @@ -27,6 +27,11 @@ unrelated three-second HTTP readiness dependency. Focused repeated normal/race, complete package normal/race, and repository normal/race/coverage gates pass; promotion is pending before the cron/callback repair chain can merge. +Current status: Issue #1054 now makes pending AskUserQuestion input readable +before `waiting_for_user` status/events become visible. Focused repeated +normal/race, complete harness normal/race, and repository normal/race/coverage +gates pass; promotion is pending before the cron/callback repair chain. + Current status: Issue #1023 anytime contextual `/feedback` intake is implemented test-first and verified in its isolated worktree; targeted, full normal/race, coverage-gate, and real TUI bundle checks pass, with merge pending. @@ -41,6 +46,7 @@ Current active plans: - `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` +- `2026-07-30-issue-1054-waiting-pending-order-plan.md` - `2026-07-30-issue-1023-feedback-intake-plan.md` - `2026-06-26-adapter-first-eval-harness-plan.md` - `2026-04-05-orchestration-program-plan.md` diff --git a/internal/checkpoints/memory.go b/internal/checkpoints/memory.go index 00128df2..70e25155 100644 --- a/internal/checkpoints/memory.go +++ b/internal/checkpoints/memory.go @@ -4,6 +4,7 @@ import ( "context" "sort" "sync" + "time" ) type MemoryStore struct { @@ -29,6 +30,31 @@ func (m *MemoryStore) Update(_ context.Context, record *Record) error { return nil } +func (m *MemoryStore) ResolvePending( + ctx context.Context, + id string, + status Status, + resumePayload string, + updatedAt time.Time, +) (*Record, bool, error) { + if err := ctx.Err(); err != nil { + return nil, false, err + } + m.mu.Lock() + defer m.mu.Unlock() + record, ok := m.records[id] + if !ok { + return nil, false, &NotFoundError{ID: id} + } + if record.Status != StatusPending { + return cloneRecord(record), false, nil + } + record.Status = status + record.ResumePayload = resumePayload + record.UpdatedAt = updatedAt + return cloneRecord(record), true, nil +} + func (m *MemoryStore) Get(_ context.Context, id string) (*Record, error) { m.mu.RLock() defer m.mu.RUnlock() diff --git a/internal/checkpoints/service.go b/internal/checkpoints/service.go index 7c6bc067..e6414b51 100644 --- a/internal/checkpoints/service.go +++ b/internal/checkpoints/service.go @@ -3,6 +3,7 @@ package checkpoints import ( "context" "encoding/json" + "errors" "fmt" "sync" "time" @@ -10,11 +11,20 @@ import ( "github.com/google/uuid" ) +var ErrAlreadyResolved = errors.New("checkpoint already resolved") + type Service struct { - store Store - now func() time.Time - mu sync.Mutex - waiters map[string][]chan waitResult + store Store + now func() time.Time + resolutionLocksMu sync.Mutex + resolutionLocks map[string]*resolutionLock + mu sync.Mutex + waiters map[string][]chan waitResult +} + +type resolutionLock struct { + token chan struct{} + refs int } type waitResult struct { @@ -30,9 +40,10 @@ func NewService(store Store, now func() time.Time) *Service { now = time.Now } return &Service{ - store: store, - now: now, - waiters: make(map[string][]chan waitResult), + store: store, + now: now, + resolutionLocks: make(map[string]*resolutionLock), + waiters: make(map[string][]chan waitResult), } } @@ -118,12 +129,33 @@ func (s *Service) Wait(ctx context.Context, id string) (WaitResult, error) { return waitResultFromRecord(record) } - select { - case outcome := <-ch: - return outcome.result, outcome.err - case <-ctx.Done(): - s.unregister(id, ch) - return WaitResult{}, ctx.Err() + ticker := time.NewTicker(25 * time.Millisecond) + defer ticker.Stop() + for { + select { + case outcome := <-ch: + return outcome.result, outcome.err + case <-ticker.C: + record, err := s.store.Get(ctx, id) + if err != nil { + // Cross-Service polling is opportunistic. Once the pending record + // and local waiter are established, a transient read outage must not + // discard that waiter or mask a later local/remote resolution. The + // caller's context remains the termination boundary. + continue + } + // A resolution owned by this Service notifies only after its store + // call has fully returned. Do not let polling observe the durable + // row early and overtake that local completion boundary. Polling is + // the fallback only for resolutions performed by another Service. + if record.Status != StatusPending && !s.resolutionActive(id) { + s.unregister(id, ch) + return waitResultFromRecord(record) + } + case <-ctx.Done(): + s.unregister(id, ch) + return WaitResult{}, ctx.Err() + } } } @@ -150,28 +182,102 @@ func (s *Service) Expire(ctx context.Context, id string) error { return s.resolve(ctx, id, StatusExpired, nil) } -func (s *Service) resolve(ctx context.Context, id string, status Status, payload map[string]any) error { - record, err := s.store.Get(ctx, id) +// ExpirePending atomically expires an unresolved checkpoint. It returns false +// when another resolution already won, without overwriting that result. +func (s *Service) ExpirePending(ctx context.Context, id string) (bool, error) { + unlock, err := s.acquireResolution(ctx, id) if err != nil { - return err + return false, err + } + defer unlock() + record, won, err := s.store.ResolvePending( + ctx, + id, + StatusExpired, + "", + s.now().UTC(), + ) + if err != nil { + return false, err + } + if !won { + return false, nil } - record.Status = status - record.UpdatedAt = s.now().UTC() + result, resultErr := waitResultFromRecord(record) + s.notify(id, waitResult{result: result, err: resultErr}) + return true, resultErr +} + +func (s *Service) resolve(ctx context.Context, id string, status Status, payload map[string]any) error { + resumePayload := "" if payload != nil { raw, err := json.Marshal(payload) if err != nil { return fmt.Errorf("marshal checkpoint payload: %w", err) } - record.ResumePayload = string(raw) + resumePayload = string(raw) } - if err := s.store.Update(ctx, record); err != nil { + unlock, err := s.acquireResolution(ctx, id) + if err != nil { return err } + defer unlock() + record, won, err := s.store.ResolvePending( + ctx, + id, + status, + resumePayload, + s.now().UTC(), + ) + if err != nil { + return err + } + if !won { + return fmt.Errorf("%w: id=%s status=%s", ErrAlreadyResolved, id, record.Status) + } result, err := waitResultFromRecord(record) s.notify(id, waitResult{result: result, err: err}) return err } +func (s *Service) acquireResolution(ctx context.Context, id string) (func(), error) { + s.resolutionLocksMu.Lock() + lock := s.resolutionLocks[id] + if lock == nil { + lock = &resolutionLock{token: make(chan struct{}, 1)} + lock.token <- struct{}{} + s.resolutionLocks[id] = lock + } + lock.refs++ + s.resolutionLocksMu.Unlock() + + select { + case <-lock.token: + return func() { + lock.token <- struct{}{} + s.releaseResolutionRef(id, lock) + }, nil + case <-ctx.Done(): + s.releaseResolutionRef(id, lock) + return nil, ctx.Err() + } +} + +func (s *Service) releaseResolutionRef(id string, lock *resolutionLock) { + s.resolutionLocksMu.Lock() + lock.refs-- + if lock.refs == 0 && s.resolutionLocks[id] == lock { + delete(s.resolutionLocks, id) + } + s.resolutionLocksMu.Unlock() +} + +func (s *Service) resolutionActive(id string) bool { + s.resolutionLocksMu.Lock() + defer s.resolutionLocksMu.Unlock() + return s.resolutionLocks[id] != nil +} + func waitResultFromRecord(record *Record) (WaitResult, error) { result := WaitResult{Status: record.Status} if record.ResumePayload == "" { diff --git a/internal/checkpoints/service_test.go b/internal/checkpoints/service_test.go index ea2e649e..4fd41dbf 100644 --- a/internal/checkpoints/service_test.go +++ b/internal/checkpoints/service_test.go @@ -5,10 +5,131 @@ import ( "errors" "path/filepath" "strings" + "sync" "testing" "time" ) +type recordBlockingCheckpointStore struct { + Store + blockID string + started chan struct{} + release chan struct{} + once sync.Once +} + +func (s *recordBlockingCheckpointStore) Update(ctx context.Context, record *Record) error { + if record.ID == s.blockID { + s.once.Do(func() { close(s.started) }) + select { + case <-s.release: + case <-ctx.Done(): + return ctx.Err() + } + } + return s.Store.Update(ctx, record) +} + +func (s *recordBlockingCheckpointStore) ResolvePending( + ctx context.Context, + id string, + status Status, + resumePayload string, + updatedAt time.Time, +) (*Record, bool, error) { + if id == s.blockID { + s.once.Do(func() { close(s.started) }) + select { + case <-s.release: + case <-ctx.Done(): + return nil, false, ctx.Err() + } + } + return s.Store.ResolvePending(ctx, id, status, resumePayload, updatedAt) +} + +type racingConditionalCheckpointStore struct { + *MemoryStore + entered chan struct{} + release chan struct{} +} + +type transientPollGetStore struct { + Store + mu sync.Mutex + getCalls int + pollFailed chan struct{} +} + +func (s *transientPollGetStore) Get(ctx context.Context, id string) (*Record, error) { + s.mu.Lock() + s.getCalls++ + call := s.getCalls + s.mu.Unlock() + if call == 3 { + close(s.pollFailed) + return nil, errors.New("transient checkpoint read outage") + } + return s.Store.Get(ctx, id) +} + +func newRacingConditionalCheckpointStore() *racingConditionalCheckpointStore { + return &racingConditionalCheckpointStore{ + MemoryStore: NewMemoryStore(), + entered: make(chan struct{}, 2), + release: make(chan struct{}), + } +} + +func (s *racingConditionalCheckpointStore) waitForRace(ctx context.Context) error { + select { + case s.entered <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + } + select { + case <-s.release: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (s *racingConditionalCheckpointStore) Update(ctx context.Context, record *Record) error { + if err := s.waitForRace(ctx); err != nil { + return err + } + return s.MemoryStore.Update(ctx, record) +} + +// ResolvePending is the atomic store contract exercised by the repair. It is +// intentionally an extra method until the production Store interface adopts +// it; the pre-fix Service continues through non-conditional Update above. +func (s *racingConditionalCheckpointStore) ResolvePending( + ctx context.Context, + id string, + status Status, + resumePayload string, + updatedAt time.Time, +) (*Record, bool, error) { + if err := s.waitForRace(ctx); err != nil { + return nil, false, err + } + s.mu.Lock() + defer s.mu.Unlock() + current, ok := s.records[id] + if !ok { + return nil, false, &NotFoundError{ID: id} + } + if current.Status != StatusPending { + return cloneRecord(current), false, nil + } + current.Status = status + current.ResumePayload = resumePayload + current.UpdatedAt = updatedAt + return cloneRecord(current), true, nil +} + func TestSQLiteStorePersistsCheckpointAcrossReopen(t *testing.T) { t.Parallel() @@ -74,6 +195,128 @@ func TestSQLiteStorePersistsCheckpointAcrossReopen(t *testing.T) { } } +func TestMemoryStoreUpdateCopiesReplacementRecord(t *testing.T) { + t.Parallel() + + store := NewMemoryStore() + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + record := &Record{ + ID: "checkpoint-memory-update", + Kind: KindExternalResume, + Status: StatusPending, + WorkflowRunID: "workflow-memory-update", + CreatedAt: now, + UpdatedAt: now, + } + if err := store.Create(context.Background(), record); err != nil { + t.Fatalf("Create: %v", err) + } + + replacement := cloneRecord(record) + replacement.Status = StatusResumed + replacement.ResumePayload = `{"answer":"persisted"}` + replacement.UpdatedAt = now.Add(time.Minute) + if err := store.Update(context.Background(), replacement); err != nil { + t.Fatalf("Update: %v", err) + } + replacement.Status = StatusDenied + replacement.ResumePayload = `{"answer":"mutated-after-update"}` + + loaded, err := store.Get(context.Background(), record.ID) + if err != nil { + t.Fatalf("Get: %v", err) + } + if loaded.Status != StatusResumed { + t.Fatalf("status = %q, want %q", loaded.Status, StatusResumed) + } + if loaded.ResumePayload != `{"answer":"persisted"}` { + t.Fatalf("resume payload = %q, want copied replacement payload", loaded.ResumePayload) + } + if !loaded.UpdatedAt.Equal(now.Add(time.Minute)) { + t.Fatalf("updated_at = %s, want %s", loaded.UpdatedAt, now.Add(time.Minute)) + } +} + +func TestCheckpointStoresResolvePendingAtomically(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + open func(t *testing.T) Store + }{ + { + name: "memory", + open: func(t *testing.T) Store { + t.Helper() + return NewMemoryStore() + }, + }, + { + name: "sqlite", + open: func(t *testing.T) Store { + t.Helper() + store, err := NewSQLiteStore(filepath.Join(t.TempDir(), "checkpoints.db")) + if err != nil { + t.Fatalf("NewSQLiteStore: %v", err) + } + if err := store.Migrate(context.Background()); err != nil { + t.Fatalf("Migrate: %v", err) + } + return store + }, + }, + } + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + store := test.open(t) + t.Cleanup(func() { _ = store.Close() }) + now := time.Now().UTC() + record := &Record{ + ID: "checkpoint-atomic-" + test.name, + Kind: KindUserInput, + Status: StatusPending, + RunID: "run-atomic-" + test.name, + CreatedAt: now, + UpdatedAt: now, + } + if err := store.Create(context.Background(), record); err != nil { + t.Fatalf("Create: %v", err) + } + resumed, won, err := store.ResolvePending( + context.Background(), + record.ID, + StatusResumed, + `{"answer":"yes"}`, + now.Add(time.Second), + ) + if err != nil { + t.Fatalf("ResolvePending first: %v", err) + } + if !won || resumed.Status != StatusResumed { + t.Fatalf("first resolution = (%+v, %t), want resumed winner", resumed, won) + } + current, won, err := store.ResolvePending( + context.Background(), + record.ID, + StatusExpired, + "", + now.Add(2*time.Second), + ) + if err != nil { + t.Fatalf("ResolvePending second: %v", err) + } + if won { + t.Fatal("second terminal transition unexpectedly won") + } + if current.Status != StatusResumed || current.ResumePayload != `{"answer":"yes"}` { + t.Fatalf("current record = %+v, want original resumed payload", current) + } + }) + } +} + func TestServiceResumeWakesWaiterAndPersistsPayload(t *testing.T) { t.Parallel() @@ -132,6 +375,284 @@ func TestServiceResumeWakesWaiterAndPersistsPayload(t *testing.T) { } } +func TestServiceReportsWhenResolutionAlreadyLostToExpiry(t *testing.T) { + t.Parallel() + + svc := NewService(NewMemoryStore(), time.Now) + record, err := svc.Create(context.Background(), CreateRequest{ + Kind: KindUserInput, + RunID: "run-resolution-race", + DeadlineAt: time.Now().Add(time.Minute), + }) + if err != nil { + t.Fatalf("Create: %v", err) + } + expired, err := svc.ExpirePending(context.Background(), record.ID) + if err != nil { + t.Fatalf("ExpirePending: %v", err) + } + if !expired { + t.Fatal("ExpirePending did not resolve pending checkpoint") + } + if err := svc.Resume(context.Background(), record.ID, map[string]any{"answer": "late"}); !errors.Is(err, ErrAlreadyResolved) { + t.Fatalf("Resume error = %v, want ErrAlreadyResolved", err) + } +} + +func TestServiceResolutionDoesNotSerializeUnrelatedRecords(t *testing.T) { + t.Parallel() + + base := NewMemoryStore() + store := &recordBlockingCheckpointStore{ + Store: base, + started: make(chan struct{}), + release: make(chan struct{}), + } + released := false + t.Cleanup(func() { + if !released { + close(store.release) + } + }) + svc := NewService(store, time.Now) + first, err := svc.Create(context.Background(), CreateRequest{Kind: KindUserInput, RunID: "run-a"}) + if err != nil { + t.Fatalf("Create first: %v", err) + } + second, err := svc.Create(context.Background(), CreateRequest{Kind: KindUserInput, RunID: "run-b"}) + if err != nil { + t.Fatalf("Create second: %v", err) + } + store.blockID = first.ID + firstDone := make(chan error, 1) + go func() { + firstDone <- svc.Resume(context.Background(), first.ID, map[string]any{"answer": "a"}) + }() + select { + case <-store.started: + case <-time.After(time.Second): + t.Fatal("timed out waiting for first record resolution") + } + + secondDone := make(chan error, 1) + go func() { + secondDone <- svc.Resume(context.Background(), second.ID, map[string]any{"answer": "b"}) + }() + select { + case err := <-secondDone: + if err != nil { + t.Fatalf("unrelated Resume: %v", err) + } + case <-time.After(250 * time.Millisecond): + t.Fatal("unrelated checkpoint resolution was blocked by service-wide serialization") + } + close(store.release) + released = true + if err := <-firstDone; err != nil { + t.Fatalf("first Resume: %v", err) + } +} + +func TestServiceResolutionLockHonorsWaitingContext(t *testing.T) { + t.Parallel() + + base := NewMemoryStore() + store := &recordBlockingCheckpointStore{ + Store: base, + started: make(chan struct{}), + release: make(chan struct{}), + } + released := false + t.Cleanup(func() { + if !released { + close(store.release) + } + }) + svc := NewService(store, time.Now) + record, err := svc.Create(context.Background(), CreateRequest{Kind: KindUserInput, RunID: "run-context"}) + if err != nil { + t.Fatalf("Create: %v", err) + } + store.blockID = record.ID + firstDone := make(chan error, 1) + go func() { + firstDone <- svc.Resume(context.Background(), record.ID, map[string]any{"answer": "first"}) + }() + select { + case <-store.started: + case <-time.After(time.Second): + t.Fatal("timed out waiting for first resolution") + } + + waitCtx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + secondDone := make(chan error, 1) + go func() { + secondDone <- svc.Resume(waitCtx, record.ID, map[string]any{"answer": "second"}) + }() + select { + case err := <-secondDone: + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("waiting Resume error = %v, want context deadline", err) + } + case <-time.After(250 * time.Millisecond): + t.Fatal("checkpoint resolution lock ignored the waiting context") + } + close(store.release) + released = true + if err := <-firstDone; err != nil { + t.Fatalf("first Resume: %v", err) + } +} + +func TestServiceConditionalResolutionHasOneWinnerAcrossServices(t *testing.T) { + t.Parallel() + + store := newRacingConditionalCheckpointStore() + svcA := NewService(store, time.Now) + svcB := NewService(store, time.Now) + record, err := svcA.Create(context.Background(), CreateRequest{Kind: KindUserInput, RunID: "run-shared"}) + if err != nil { + t.Fatalf("Create: %v", err) + } + + resumeDone := make(chan error, 1) + expireDone := make(chan struct { + expired bool + err error + }, 1) + go func() { + resumeDone <- svcA.Resume(context.Background(), record.ID, map[string]any{"answer": "accepted"}) + }() + go func() { + expired, err := svcB.ExpirePending(context.Background(), record.ID) + expireDone <- struct { + expired bool + err error + }{expired: expired, err: err} + }() + for range 2 { + select { + case <-store.entered: + case <-time.After(time.Second): + t.Fatal("timed out staging cross-service resolution race") + } + } + close(store.release) + resumeErr := <-resumeDone + expireResult := <-expireDone + if expireResult.err != nil { + t.Fatalf("ExpirePending: %v", expireResult.err) + } + resumeWon := resumeErr == nil + if resumeErr != nil && !errors.Is(resumeErr, ErrAlreadyResolved) { + t.Fatalf("Resume error = %v, want nil or ErrAlreadyResolved", resumeErr) + } + if resumeWon == expireResult.expired { + t.Fatalf("resolution winners: resume=%t expire=%t, want exactly one", resumeWon, expireResult.expired) + } + loaded, err := svcA.Get(context.Background(), record.ID) + if err != nil { + t.Fatalf("Get: %v", err) + } + if resumeWon && loaded.Status != StatusResumed { + t.Fatalf("stored status = %q, want resumed winner", loaded.Status) + } + if expireResult.expired && loaded.Status != StatusExpired { + t.Fatalf("stored status = %q, want expired winner", loaded.Status) + } +} + +func TestServiceWaitObservesResolutionFromAnotherService(t *testing.T) { + t.Parallel() + + store := NewMemoryStore() + waitingService := NewService(store, time.Now) + resolvingService := NewService(store, time.Now) + record, err := waitingService.Create(context.Background(), CreateRequest{Kind: KindUserInput, RunID: "run-remote"}) + if err != nil { + t.Fatalf("Create: %v", err) + } + waitCtx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + waitDone := make(chan waitResult, 1) + go func() { + result, err := waitingService.Wait(waitCtx, record.ID) + waitDone <- waitResult{result: result, err: err} + }() + + deadline := time.Now().Add(time.Second) + for { + waitingService.mu.Lock() + registered := len(waitingService.waiters[record.ID]) == 1 + waitingService.mu.Unlock() + if registered { + break + } + if time.Now().After(deadline) { + t.Fatal("timed out waiting for waiter registration") + } + time.Sleep(time.Millisecond) + } + if err := resolvingService.Resume(context.Background(), record.ID, map[string]any{"answer": "remote"}); err != nil { + t.Fatalf("remote Resume: %v", err) + } + select { + case outcome := <-waitDone: + if outcome.err != nil { + t.Fatalf("Wait: %v", outcome.err) + } + if outcome.result.Status != StatusResumed { + t.Fatalf("Wait status = %q, want resumed", outcome.result.Status) + } + case <-time.After(250 * time.Millisecond): + t.Fatal("Wait did not observe resolution persisted by another Service") + } +} + +func TestServiceWaitSurvivesTransientPollingReadFailure(t *testing.T) { + t.Parallel() + + baseStore := NewMemoryStore() + store := &transientPollGetStore{ + Store: baseStore, + pollFailed: make(chan struct{}), + } + waitingService := NewService(store, time.Now) + resolvingService := NewService(baseStore, time.Now) + record, err := waitingService.Create(context.Background(), CreateRequest{Kind: KindUserInput, RunID: "run-transient-poll"}) + if err != nil { + t.Fatalf("Create: %v", err) + } + waitCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + waitDone := make(chan waitResult, 1) + go func() { + result, err := waitingService.Wait(waitCtx, record.ID) + waitDone <- waitResult{result: result, err: err} + }() + + select { + case <-store.pollFailed: + case <-time.After(500 * time.Millisecond): + t.Fatal("timed out waiting for transient polling failure") + } + if err := resolvingService.Resume(context.Background(), record.ID, map[string]any{"answer": "after outage"}); err != nil { + t.Fatalf("remote Resume: %v", err) + } + select { + case outcome := <-waitDone: + if outcome.err != nil { + t.Fatalf("Wait returned transient polling error: %v", outcome.err) + } + if outcome.result.Status != StatusResumed || outcome.result.Payload["answer"] != "after outage" { + t.Fatalf("Wait result = %+v, want resumed remote payload", outcome.result) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("Wait did not remain registered after transient polling failure") + } +} + func TestServiceStoreDenyExpireAndWaitCancellation(t *testing.T) { t.Parallel() @@ -213,8 +734,13 @@ func TestServiceStoreDenyExpireAndWaitCancellation(t *testing.T) { } cancel() - if err := <-errCh; !errors.Is(err, context.Canceled) { - t.Fatalf("Wait cancellation error = %v, want context.Canceled", err) + select { + case err := <-errCh: + if !errors.Is(err, context.Canceled) { + t.Fatalf("Wait cancellation error = %v, want context.Canceled", err) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for cancelled waiter") } svc.mu.Lock() _, stillRegistered := svc.waiters[cancelled.ID] diff --git a/internal/checkpoints/sqlite.go b/internal/checkpoints/sqlite.go index 16a7cd5e..fd3d11ea 100644 --- a/internal/checkpoints/sqlite.go +++ b/internal/checkpoints/sqlite.go @@ -138,6 +138,40 @@ WHERE id = ? return nil } +func (s *SQLiteStore) ResolvePending( + ctx context.Context, + id string, + status Status, + resumePayload string, + updatedAt time.Time, +) (*Record, bool, error) { + result, err := s.db.ExecContext(ctx, ` +UPDATE checkpoints +SET status = ?, + resume_payload = ?, + updated_at = ? +WHERE id = ? AND status = ? +`, + string(status), + resumePayload, + timeString(updatedAt), + id, + string(StatusPending), + ) + if err != nil { + return nil, false, fmt.Errorf("checkpoints: resolve pending: %w", err) + } + affected, err := result.RowsAffected() + if err != nil { + return nil, false, fmt.Errorf("checkpoints: resolve pending rows affected: %w", err) + } + record, err := s.Get(ctx, id) + if err != nil { + return nil, false, err + } + return record, affected == 1, nil +} + func (s *SQLiteStore) Get(ctx context.Context, id string) (*Record, error) { row := s.db.QueryRowContext(ctx, ` SELECT id, kind, status, run_id, workflow_run_id, call_id, tool, args, questions, diff --git a/internal/checkpoints/store.go b/internal/checkpoints/store.go index 7734e413..b3650b04 100644 --- a/internal/checkpoints/store.go +++ b/internal/checkpoints/store.go @@ -63,6 +63,16 @@ type WaitResult struct { type Store interface { Create(ctx context.Context, record *Record) error Update(ctx context.Context, record *Record) error + // ResolvePending atomically transitions one pending record. It returns the + // current record and won=false when another terminal transition already + // committed. + ResolvePending( + ctx context.Context, + id string, + status Status, + resumePayload string, + updatedAt time.Time, + ) (record *Record, won bool, err error) Get(ctx context.Context, id string) (*Record, error) PendingByRun(ctx context.Context, runID string) (*Record, error) PendingByWorkflowRun(ctx context.Context, workflowRunID string) (*Record, error) diff --git a/internal/harness/ask_user_broker.go b/internal/harness/ask_user_broker.go index 219c1fb8..2cd98f19 100644 --- a/internal/harness/ask_user_broker.go +++ b/internal/harness/ask_user_broker.go @@ -72,22 +72,73 @@ func (b *InMemoryAskUserQuestionBroker) Ask(ctx context.Context, req htools.AskU b.pending[req.RunID] = entry b.mu.Unlock() - timer := time.NewTimer(req.Timeout) - defer timer.Stop() + waitCtx, cancel := context.WithTimeout(ctx, req.Timeout) + defer cancel() + + if req.OnPending != nil { + notified := make(chan struct{}) + go func() { + defer close(notified) + req.OnPending(waitCtx, entry.pending) + }() + select { + case <-notified: + case <-waitCtx.Done(): + answers, answeredAt, err := b.finishAskWait(ctx, req, entry) + if err != nil { + return nil, time.Time{}, err + } + if err := waitForPendingPublication(ctx, notified); err != nil { + return nil, time.Time{}, err + } + return answers, answeredAt, nil + } + } select { case submission := <-entry.answerC: return submission.answers, submission.answeredAt, nil - case <-timer.C: - b.clearPendingIfMatch(req.RunID, entry) + case <-waitCtx.Done(): + return b.finishAskWait(ctx, req, entry) + } +} + +func (b *InMemoryAskUserQuestionBroker) finishAskWait( + ctx context.Context, + req htools.AskUserQuestionRequest, + entry *pendingUserQuestion, +) (map[string]string, time.Time, error) { + b.mu.Lock() + current, stillPending := b.pending[req.RunID] + if stillPending && current == entry { + delete(b.pending, req.RunID) + b.mu.Unlock() + if err := ctx.Err(); err != nil { + return nil, time.Time{}, err + } return nil, time.Time{}, &htools.AskUserQuestionTimeoutError{ RunID: req.RunID, CallID: req.CallID, DeadlineAt: entry.pending.DeadlineAt, } + } + b.mu.Unlock() + + submission := <-entry.answerC + return submission.answers, submission.answeredAt, nil +} + +// waitForPendingPublication preserves the externally visible lifecycle order +// when an answer wins the deadline race. Once Submit has accepted an answer, +// Ask must not let the caller emit run.resumed until the pending notifier has +// completed run.waiting_for_user publication. The parent context remains the +// cancellation escape hatch for a notifier that cannot complete. +func waitForPendingPublication(ctx context.Context, notified <-chan struct{}) error { + select { + case <-notified: + return nil case <-ctx.Done(): - b.clearPendingIfMatch(req.RunID, entry) - return nil, time.Time{}, ctx.Err() + return ctx.Err() } } @@ -115,22 +166,10 @@ func (b *InMemoryAskUserQuestionBroker) Submit(runID string, answers map[string] b.mu.Unlock() return fmt.Errorf("%w: %v", ErrInvalidUserQuestionInput, err) } - delete(b.pending, runID) answeredAt := b.now().UTC() + entry.answerC <- askUserSubmission{answers: normalized, answeredAt: answeredAt} + delete(b.pending, runID) b.mu.Unlock() - entry.answerC <- askUserSubmission{answers: normalized, answeredAt: answeredAt} return nil } - -func (b *InMemoryAskUserQuestionBroker) clearPendingIfMatch(runID string, entry *pendingUserQuestion) { - b.mu.Lock() - defer b.mu.Unlock() - current, ok := b.pending[runID] - if !ok { - return - } - if current == entry { - delete(b.pending, runID) - } -} diff --git a/internal/harness/ask_user_broker_test.go b/internal/harness/ask_user_broker_test.go index 5900c6b9..9cc06002 100644 --- a/internal/harness/ask_user_broker_test.go +++ b/internal/harness/ask_user_broker_test.go @@ -27,6 +27,7 @@ func TestInMemoryAskUserQuestionBrokerLifecycle(t *testing.T) { broker := NewInMemoryAskUserQuestionBroker(time.Now) errCh := make(chan error, 1) answersCh := make(chan map[string]string, 1) + pendingReady := make(chan htools.AskUserQuestionPending, 1) go func() { answers, _, err := broker.Ask(context.Background(), htools.AskUserQuestionRequest{ @@ -34,6 +35,13 @@ func TestInMemoryAskUserQuestionBrokerLifecycle(t *testing.T) { CallID: "call_1", Questions: askQuestionsFixture(), Timeout: 2 * time.Second, + OnPending: func(_ context.Context, pending htools.AskUserQuestionPending) { + if current, ok := broker.Pending("run_1"); !ok || current.CallID != pending.CallID { + errCh <- errors.New("pending input was not readable inside OnPending") + return + } + pendingReady <- pending + }, }) if err != nil { errCh <- err @@ -42,18 +50,15 @@ func TestInMemoryAskUserQuestionBrokerLifecycle(t *testing.T) { answersCh <- answers }() - deadline := time.Now().Add(1 * time.Second) - for { - if pending, ok := broker.Pending("run_1"); ok { - if pending.CallID != "call_1" { - t.Fatalf("unexpected call id: %q", pending.CallID) - } - break - } - if time.Now().After(deadline) { - t.Fatalf("pending question did not appear") + select { + case err := <-errCh: + t.Fatalf("unexpected readiness error: %v", err) + case pending := <-pendingReady: + if pending.CallID != "call_1" { + t.Fatalf("unexpected call id: %q", pending.CallID) } - time.Sleep(5 * time.Millisecond) + case <-time.After(time.Second): + t.Fatal("timed out waiting for pending notification") } if err := broker.Submit("run_1", map[string]string{"Where next?": "Docs"}); err != nil { @@ -76,6 +81,109 @@ func TestInMemoryAskUserQuestionBrokerLifecycle(t *testing.T) { } } +func TestInMemoryAskUserQuestionBrokerTimeoutIncludesPendingNotification(t *testing.T) { + t.Parallel() + + const timeout = 200 * time.Millisecond + broker := NewInMemoryAskUserQuestionBroker(time.Now) + notificationStarted := make(chan struct{}) + releaseNotification := make(chan struct{}) + result := make(chan error, 1) + released := false + defer func() { + if !released { + close(releaseNotification) + } + }() + + go func() { + _, _, err := broker.Ask(context.Background(), htools.AskUserQuestionRequest{ + RunID: "run_slow_notification", + CallID: "call_slow_notification", + Questions: askQuestionsFixture(), + Timeout: timeout, + OnPending: func(_ context.Context, _ htools.AskUserQuestionPending) { + close(notificationStarted) + <-releaseNotification + }, + }) + result <- err + }() + + select { + case <-notificationStarted: + case <-time.After(time.Second): + t.Fatal("timed out waiting for pending notification") + } + time.Sleep(timeout + 50*time.Millisecond) + + select { + case err := <-result: + if !htools.IsAskUserQuestionTimeout(err) { + t.Fatalf("Ask error = %v, want timeout", err) + } + case <-time.After(timeout / 2): + t.Fatal("Ask did not honor its timeout while OnPending remained blocked") + } + close(releaseNotification) + released = true +} + +func TestInMemoryAskUserQuestionBrokerKeepsAnswerSubmittedBeforeNotifierDeadline(t *testing.T) { + t.Parallel() + + const timeout = 150 * time.Millisecond + broker := NewInMemoryAskUserQuestionBroker(time.Now) + started := make(chan struct{}) + release := make(chan struct{}) + type askResult struct { + answers map[string]string + err error + } + result := make(chan askResult, 1) + go func() { + answers, _, err := broker.Ask(context.Background(), htools.AskUserQuestionRequest{ + RunID: "run_answered_before_deadline", + CallID: "call_answered_before_deadline", + Questions: askQuestionsFixture(), + Timeout: timeout, + OnPending: func(_ context.Context, _ htools.AskUserQuestionPending) { + close(started) + <-release + }, + }) + result <- askResult{answers: answers, err: err} + }() + + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for pending notifier") + } + if err := broker.Submit("run_answered_before_deadline", map[string]string{"Where next?": "Docs"}); err != nil { + t.Fatalf("Submit: %v", err) + } + time.Sleep(timeout + 50*time.Millisecond) + select { + case out := <-result: + t.Fatalf("Ask returned before pending publication completed: %+v", out) + default: + } + close(release) + var out askResult + select { + case out = <-result: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for Ask result after pending publication") + } + if out.err != nil { + t.Fatalf("Ask returned error after timely answer: %v", out.err) + } + if got := out.answers["Where next?"]; got != "Docs" { + t.Fatalf("answer = %q, want Docs", got) + } +} + func TestInMemoryAskUserQuestionBrokerTimeoutAndValidation(t *testing.T) { t.Parallel() diff --git a/internal/harness/checkpoint_broker_test.go b/internal/harness/checkpoint_broker_test.go index d870553f..c3408c5e 100644 --- a/internal/harness/checkpoint_broker_test.go +++ b/internal/harness/checkpoint_broker_test.go @@ -3,6 +3,7 @@ package harness import ( "context" "encoding/json" + "errors" "testing" "time" @@ -134,6 +135,7 @@ func TestCheckpointAskUserBrokerPersistsQuestionsAndAnswers(t *testing.T) { broker := NewCheckpointAskUserQuestionBroker(checkpointSvc, func() time.Time { return now }) done := make(chan error, 1) + pendingReady := make(chan htools.AskUserQuestionPending, 1) go func() { answers, answeredAt, err := broker.Ask(context.Background(), htools.AskUserQuestionRequest{ RunID: "run-ask", @@ -147,6 +149,13 @@ func TestCheckpointAskUserBrokerPersistsQuestionsAndAnswers(t *testing.T) { }, }}, Timeout: time.Minute, + OnPending: func(_ context.Context, pending htools.AskUserQuestionPending) { + if current, ok := broker.Pending("run-ask"); !ok || current.CallID != pending.CallID { + done <- errors.New("persisted pending input was not readable inside OnPending") + return + } + pendingReady <- pending + }, }) if err != nil { done <- err @@ -164,17 +173,12 @@ func TestCheckpointAskUserBrokerPersistsQuestionsAndAnswers(t *testing.T) { }() var pending htools.AskUserQuestionPending - deadline := time.Now().Add(2 * time.Second) - for { - current, ok := broker.Pending("run-ask") - if ok { - pending = current - break - } - if time.Now().After(deadline) { - t.Fatal("timed out waiting for pending question") - } - time.Sleep(10 * time.Millisecond) + select { + case err := <-done: + t.Fatalf("unexpected readiness error: %v", err) + case pending = <-pendingReady: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for pending notification") } if pending.CallID != "call-ask" { @@ -206,6 +210,111 @@ func TestCheckpointAskUserBrokerPersistsQuestionsAndAnswers(t *testing.T) { } } +func TestCheckpointAskUserBrokerTimeoutIncludesPendingNotification(t *testing.T) { + t.Parallel() + + const timeout = 200 * time.Millisecond + checkpointSvc := checkpoints.NewService(checkpoints.NewMemoryStore(), time.Now) + broker := NewCheckpointAskUserQuestionBroker(checkpointSvc, time.Now) + notificationStarted := make(chan struct{}) + releaseNotification := make(chan struct{}) + result := make(chan error, 1) + released := false + defer func() { + if !released { + close(releaseNotification) + } + }() + + go func() { + _, _, err := broker.Ask(context.Background(), htools.AskUserQuestionRequest{ + RunID: "run-slow-notification", + CallID: "call-slow-notification", + Questions: askQuestionsFixture(), + Timeout: timeout, + OnPending: func(_ context.Context, _ htools.AskUserQuestionPending) { + close(notificationStarted) + <-releaseNotification + }, + }) + result <- err + }() + + select { + case <-notificationStarted: + case <-time.After(time.Second): + t.Fatal("timed out waiting for pending notification") + } + time.Sleep(timeout + 50*time.Millisecond) + + select { + case err := <-result: + if !htools.IsAskUserQuestionTimeout(err) { + t.Fatalf("Ask error = %v, want timeout", err) + } + case <-time.After(timeout / 2): + t.Fatal("Ask did not honor its timeout while OnPending remained blocked") + } + close(releaseNotification) + released = true +} + +func TestCheckpointAskUserBrokerKeepsAnswerSubmittedBeforeNotifierDeadline(t *testing.T) { + t.Parallel() + + const timeout = 150 * time.Millisecond + checkpointSvc := checkpoints.NewService(checkpoints.NewMemoryStore(), time.Now) + broker := NewCheckpointAskUserQuestionBroker(checkpointSvc, time.Now) + started := make(chan struct{}) + release := make(chan struct{}) + type askResult struct { + answers map[string]string + err error + } + result := make(chan askResult, 1) + go func() { + answers, _, err := broker.Ask(context.Background(), htools.AskUserQuestionRequest{ + RunID: "run-answered-before-deadline", + CallID: "call-answered-before-deadline", + Questions: askQuestionsFixture(), + Timeout: timeout, + OnPending: func(_ context.Context, _ htools.AskUserQuestionPending) { + close(started) + <-release + }, + }) + result <- askResult{answers: answers, err: err} + }() + + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for pending publication") + } + if err := broker.Submit("run-answered-before-deadline", map[string]string{"Where next?": "Docs"}); err != nil { + t.Fatalf("Submit: %v", err) + } + time.Sleep(timeout + 50*time.Millisecond) + select { + case out := <-result: + t.Fatalf("Ask returned before pending publication completed: %+v", out) + default: + } + close(release) + var out askResult + select { + case out = <-result: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for Ask result after pending publication") + } + if out.err != nil { + t.Fatalf("Ask returned error after timely answer: %v", out.err) + } + if got := out.answers["Where next?"]; got != "Docs" { + t.Fatalf("answer = %q, want Docs", got) + } +} + type ApprovalPendingView PendingApproval // TestCheckpointApprovalBrokerOptionsRoundTrip proves plan approach options diff --git a/internal/harness/checkpoint_brokers.go b/internal/harness/checkpoint_brokers.go index 3366437d..b4f78cbb 100644 --- a/internal/harness/checkpoint_brokers.go +++ b/internal/harness/checkpoint_brokers.go @@ -102,9 +102,15 @@ func (b *checkpointApprovalBroker) ApproveWithOption(runID, option string) error return ErrNoPendingApproval } if option == "" { - return b.service.Approve(context.Background(), record.ID) + return mapCheckpointResolutionError( + b.service.Approve(context.Background(), record.ID), + ErrNoPendingApproval, + ) } - return b.service.ApproveWithPayload(context.Background(), record.ID, map[string]any{"option": option}) + return mapCheckpointResolutionError( + b.service.ApproveWithPayload(context.Background(), record.ID, map[string]any{"option": option}), + ErrNoPendingApproval, + ) } func (b *checkpointApprovalBroker) Deny(runID string) error { @@ -115,7 +121,17 @@ func (b *checkpointApprovalBroker) Deny(runID string) error { if !ok || record.Kind != checkpoints.KindApproval { return ErrNoPendingApproval } - return b.service.Deny(context.Background(), record.ID) + return mapCheckpointResolutionError( + b.service.Deny(context.Background(), record.ID), + ErrNoPendingApproval, + ) +} + +func mapCheckpointResolutionError(err, noPending error) error { + if errors.Is(err, checkpoints.ErrAlreadyResolved) { + return noPending + } + return err } type checkpointAskUserQuestionBroker struct { @@ -151,29 +167,84 @@ func (b *checkpointAskUserQuestionBroker) Ask(ctx context.Context, req htools.As if err != nil { return nil, time.Time{}, err } - waitCtx, cancel := context.WithTimeout(ctx, req.Timeout) defer cancel() + if req.OnPending != nil { + notified := make(chan struct{}) + go func() { + defer close(notified) + req.OnPending( + waitCtx, + htools.AskUserQuestionPending{ + RunID: record.RunID, + CallID: record.CallID, + Tool: htools.AskUserQuestionToolName, + Questions: req.Questions, + DeadlineAt: record.DeadlineAt, + }, + ) + }() + select { + case <-notified: + case <-waitCtx.Done(): + answers, answeredAt, err := b.finishAskWait(ctx, req, record) + if err != nil { + return nil, time.Time{}, err + } + if err := waitForPendingPublication(ctx, notified); err != nil { + return nil, time.Time{}, err + } + return answers, answeredAt, nil + } + } + result, err := b.service.Wait(waitCtx, record.ID) if err != nil { if errors.Is(err, context.DeadlineExceeded) { - _ = b.service.Expire(context.Background(), record.ID) - return nil, time.Time{}, &htools.AskUserQuestionTimeoutError{ - RunID: req.RunID, - CallID: req.CallID, - DeadlineAt: record.DeadlineAt, - } + return b.finishAskWait(ctx, req, record) } return nil, time.Time{}, err } + return askUserAnswers(result), b.now().UTC(), nil +} + +func (b *checkpointAskUserQuestionBroker) finishAskWait( + ctx context.Context, + req htools.AskUserQuestionRequest, + record checkpoints.Record, +) (map[string]string, time.Time, error) { + if err := ctx.Err(); err != nil { + return nil, time.Time{}, err + } + expired, err := b.service.ExpirePending(context.Background(), record.ID) + if err != nil { + return nil, time.Time{}, err + } + if !expired { + result, waitErr := b.service.Wait(context.Background(), record.ID) + if waitErr != nil { + return nil, time.Time{}, waitErr + } + if result.Status == checkpoints.StatusResumed { + return askUserAnswers(result), b.now().UTC(), nil + } + } + return nil, time.Time{}, &htools.AskUserQuestionTimeoutError{ + RunID: req.RunID, + CallID: req.CallID, + DeadlineAt: record.DeadlineAt, + } +} + +func askUserAnswers(result checkpoints.WaitResult) map[string]string { answers := make(map[string]string, len(result.Payload)) for key, value := range result.Payload { if str, ok := value.(string); ok { answers[key] = str } } - return answers, b.now().UTC(), nil + return answers } func (b *checkpointAskUserQuestionBroker) Pending(runID string) (htools.AskUserQuestionPending, bool) { @@ -214,7 +285,10 @@ func (b *checkpointAskUserQuestionBroker) Submit(runID string, answers map[strin for key, value := range normalized { payload[key] = value } - return b.service.Resume(context.Background(), record.ID, payload) + return mapCheckpointResolutionError( + b.service.Resume(context.Background(), record.ID, payload), + ErrNoPendingUserQuestion, + ) } func decodeQuestions(raw string) ([]htools.AskUserQuestion, error) { diff --git a/internal/harness/checkpoint_brokers_test.go b/internal/harness/checkpoint_brokers_test.go index 473630b2..3b215f3e 100644 --- a/internal/harness/checkpoint_brokers_test.go +++ b/internal/harness/checkpoint_brokers_test.go @@ -9,6 +9,8 @@ package harness import ( "context" + "errors" + "sync" "testing" "time" @@ -16,6 +18,18 @@ import ( htools "go-agent-harness/internal/harness/tools" ) +func receiveWithin[T any](t *testing.T, ch <-chan T) T { + t.Helper() + select { + case value := <-ch: + return value + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for channel result") + var zero T + return zero + } +} + func newTestCheckpointService() *checkpoints.Service { return checkpoints.NewService(nil, time.Now) } @@ -89,7 +103,7 @@ func TestCheckpointApprovalBroker_ApproveWithOption(t *testing.T) { t.Fatalf("approve: %v", err) } - got := <-done + got := receiveWithin(t, done) if got.err != nil { t.Fatalf("Ask returned an error: %v", got.err) } @@ -116,7 +130,7 @@ func TestCheckpointApprovalBroker_Deny(t *testing.T) { if err := broker.Deny("run-2"); err != nil { t.Fatalf("deny: %v", err) } - if <-done { + if receiveWithin(t, done) { t.Error("a denied request must not report approval") } } @@ -160,6 +174,68 @@ func TestCheckpointApprovalBroker_NoPendingRecord(t *testing.T) { } } +type gatedPendingLookupStore struct { + checkpoints.Store + captured chan struct{} + release chan struct{} + once sync.Once +} + +func (s *gatedPendingLookupStore) PendingByRun(ctx context.Context, runID string) (*checkpoints.Record, error) { + record, err := s.Store.PendingByRun(ctx, runID) + s.once.Do(func() { close(s.captured) }) + <-s.release + return record, err +} + +func TestCheckpointApprovalBroker_ResolutionLostToExpiryIsNoPending(t *testing.T) { + tests := []struct { + name string + resolve func(*checkpointApprovalBroker, string) error + }{ + {name: "approve", resolve: func(b *checkpointApprovalBroker, runID string) error { + return b.Approve(runID) + }}, + {name: "approve with option", resolve: func(b *checkpointApprovalBroker, runID string) error { + return b.ApproveWithOption(runID, "a") + }}, + {name: "deny", resolve: func(b *checkpointApprovalBroker, runID string) error { + return b.Deny(runID) + }}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := &gatedPendingLookupStore{ + Store: checkpoints.NewMemoryStore(), + captured: make(chan struct{}), + release: make(chan struct{}), + } + service := checkpoints.NewService(store, time.Now) + record, err := service.Create(context.Background(), checkpoints.CreateRequest{ + Kind: checkpoints.KindApproval, + RunID: "run-expiry-race", + DeadlineAt: time.Now().Add(time.Minute), + }) + if err != nil { + t.Fatalf("Create: %v", err) + } + broker := &checkpointApprovalBroker{service: service} + result := make(chan error, 1) + go func() { result <- tt.resolve(broker, record.RunID) }() + + receiveWithin(t, store.captured) + if expired, err := service.ExpirePending(context.Background(), record.ID); err != nil || !expired { + t.Fatalf("ExpirePending = (%v, %v), want (true, nil)", expired, err) + } + close(store.release) + if err := receiveWithin(t, result); !errors.Is(err, ErrNoPendingApproval) { + t.Fatalf("resolution error = %v, want ErrNoPendingApproval", err) + } + }) + } +} + // --- ask-user broker -------------------------------------------------- func TestCheckpointAskUserQuestionBroker_SubmitAnswers(t *testing.T) { @@ -196,7 +272,7 @@ func TestCheckpointAskUserQuestionBroker_SubmitAnswers(t *testing.T) { t.Fatalf("submit: %v", err) } - got := <-done + got := receiveWithin(t, done) if got.err != nil { t.Fatalf("Ask: %v", got.err) } @@ -269,6 +345,90 @@ func TestCheckpointAskUserQuestionBroker_Timeout(t *testing.T) { } } +type blockResolvedUpdateStore struct { + checkpoints.Store + applied chan struct{} + release chan struct{} + once sync.Once +} + +func (s *blockResolvedUpdateStore) Update(ctx context.Context, record *checkpoints.Record) error { + if err := s.Store.Update(ctx, record); err != nil { + return err + } + if record.Status == checkpoints.StatusResumed { + s.once.Do(func() { close(s.applied) }) + <-s.release + } + return nil +} + +func (s *blockResolvedUpdateStore) ResolvePending( + ctx context.Context, + id string, + status checkpoints.Status, + resumePayload string, + updatedAt time.Time, +) (*checkpoints.Record, bool, error) { + record, won, err := s.Store.ResolvePending(ctx, id, status, resumePayload, updatedAt) + if err != nil { + return nil, false, err + } + if won && status == checkpoints.StatusResumed { + s.once.Do(func() { close(s.applied) }) + <-s.release + } + return record, won, nil +} + +func TestCheckpointAskUserQuestionBroker_WaitDeadlineReturnsAcceptedAnswer(t *testing.T) { + store := &blockResolvedUpdateStore{ + Store: checkpoints.NewMemoryStore(), + applied: make(chan struct{}), + release: make(chan struct{}), + } + service := checkpoints.NewService(store, time.Now) + broker := NewCheckpointAskUserQuestionBroker(service, time.Now) + type result struct { + answers map[string]string + err error + } + asked := make(chan result, 1) + go func() { + answers, _, err := broker.Ask(context.Background(), htools.AskUserQuestionRequest{ + RunID: "run-wait-deadline", CallID: "call-wait-deadline", + Questions: sampleQuestions(), Timeout: 100 * time.Millisecond, + }) + asked <- result{answers: answers, err: err} + }() + waitForPending(t, func() bool { + _, ok := broker.Pending("run-wait-deadline") + return ok + }) + + submitted := make(chan error, 1) + go func() { + submitted <- broker.Submit("run-wait-deadline", map[string]string{"Pick one": "a"}) + }() + receiveWithin(t, store.applied) + select { + case outcome := <-asked: + t.Fatalf("Ask returned before the accepted resume completed: %+v", outcome) + case <-time.After(150 * time.Millisecond): + } + close(store.release) + if err := receiveWithin(t, submitted); err != nil { + t.Fatalf("Submit: %v", err) + } + outcome := receiveWithin(t, asked) + if outcome.err != nil { + t.Fatalf("Ask returned error after accepted resume: %v", outcome.err) + } + if got := outcome.answers["Pick one"]; got != "a" { + t.Fatalf("answer = %q, want a", got) + } +} + func TestDecodeQuestions(t *testing.T) { if _, err := decodeQuestions("not json"); err == nil { t.Error("malformed stored questions must produce an error") diff --git a/internal/harness/runner.go b/internal/harness/runner.go index f82640e0..734f2b49 100644 --- a/internal/harness/runner.go +++ b/internal/harness/runner.go @@ -18,6 +18,7 @@ import ( "github.com/google/uuid" "unicode/utf8" + "go-agent-harness/internal/checkpoints" "go-agent-harness/internal/forensics/audittrail" "go-agent-harness/internal/forensics/contextwindow" "go-agent-harness/internal/forensics/errorchain" @@ -35,6 +36,7 @@ import ( type runState struct { run Run + statusPersist contextMutex planMode PlanModeState planFile string staticSystemPrompt string @@ -153,6 +155,31 @@ type runState struct { forkDepth int } +// contextMutex is a zero-value, context-aware mutex. Status persistence uses +// one per run so an older write can never land after a newer terminal write, +// while a deadline-bound pending notifier can still stop waiting promptly. +type contextMutex struct { + once sync.Once + token chan struct{} +} + +func (m *contextMutex) lock(ctx context.Context) error { + m.once.Do(func() { + m.token = make(chan struct{}, 1) + m.token <- struct{}{} + }) + select { + case <-m.token: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (m *contextMutex) unlock() { + m.token <- struct{}{} +} + // Runner concurrency/lifecycle invariants // // Event ledger: @@ -1877,7 +1904,7 @@ func (r *Runner) SubmitInput(runID string, answers map[string]string) error { return ErrNoPendingInput } if err := rc.AskUserBroker.Submit(runID, answers); err != nil { - if errors.Is(err, ErrNoPendingUserQuestion) { + if errors.Is(err, ErrNoPendingUserQuestion) || errors.Is(err, checkpoints.ErrAlreadyResolved) { return ErrNoPendingInput } if errors.Is(err, ErrInvalidUserQuestionInput) { @@ -1924,6 +1951,13 @@ func (r *Runner) SteerRun(runID, message string) error { } func (r *Runner) Subscribe(runID string) ([]Event, <-chan Event, func(), error) { + // Match emit's lock order so replay cannot snapshot a strict event while + // its durable append is still in flight. Once this lock is acquired, that + // event has either committed and will be in history, or has been discarded + // with its sequence rolled back. + r.conversationEventMu.Lock() + defer r.conversationEventMu.Unlock() + r.mu.Lock() defer r.mu.Unlock() @@ -4174,12 +4208,66 @@ func (a usageTotalsAccumulator) completionUsage() CompletionUsage { } func (r *Runner) setStatus(runID string, status RunStatus, output, runErr string) { - r.mu.Lock() + r.setStatusContext(context.Background(), runID, status, output, runErr) +} + +func (r *Runner) setStatusContext( + ctx context.Context, + runID string, + status RunStatus, + output, runErr string, +) bool { + return r.updateStatusContext(ctx, runID, status, output, runErr, nil) +} + +func (r *Runner) setStatusAndEmitContext( + ctx context.Context, + runID string, + status RunStatus, + output, runErr string, + eventType EventType, + payload map[string]any, +) bool { + return r.updateStatusContext(ctx, runID, status, output, runErr, func() bool { + return r.emitWithPersistence(ctx, runID, eventType, payload, true) + }) +} +func (r *Runner) updateStatusContext( + ctx context.Context, + runID string, + status RunStatus, + output, runErr string, + afterPersist func() bool, +) bool { + if ctx == nil { + ctx = context.Background() + } + if ctx.Err() != nil { + return false + } + r.mu.RLock() state, ok := r.runs[runID] + r.mu.RUnlock() if !ok { + return false + } + if err := state.statusPersist.lock(ctx); err != nil { + return false + } + defer state.statusPersist.unlock() + + r.mu.Lock() + current, ok := r.runs[runID] + if !ok || current != state || ctx.Err() != nil { r.mu.Unlock() - return + 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() + return false } state.run.Status = status state.run.Output = output @@ -4190,10 +4278,34 @@ func (r *Runner) setStatus(runID string, status RunStatus, output, runErr string } else { state.run.Recap = nil } + run := state.run r.mu.Unlock() - // Persist the updated run state to the store (non-fatal, called after unlock). - r.storeUpdateRun(runID) + rc := r.configForRun(runID) + if rc.Store == nil { + if ctx.Err() != nil { + return false + } + if afterPersist != nil { + return afterPersist() + } + return ctx.Err() == nil + } + 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) + } + return false + } + if ctx.Err() != nil { + return false + } + if afterPersist != nil { + // Keep statusPersist held through its corresponding lifecycle event so + // terminal mutation/publication cannot overtake a waiting transition. + return afterPersist() + } + return ctx.Err() == nil } func (r *Runner) setMessages(runID string, messages []Message) { @@ -5330,6 +5442,22 @@ 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) emitWithPersistence( + ctx context.Context, + runID string, + eventType EventType, + payload map[string]any, + requirePersistence bool, +) bool { + if ctx == nil { + ctx = context.Background() + } + if ctx.Err() != nil { + return false + } r.conversationEventMu.Lock() conversationLocked := true defer func() { @@ -5342,7 +5470,7 @@ func (r *Runner) emit(runID string, eventType EventType, payload map[string]any) state, ok := r.runs[runID] if !ok { r.mu.Unlock() - return + return false } // Drop post-terminal events to preserve forensic ordering. Provider @@ -5351,32 +5479,32 @@ func (r *Runner) emit(runID string, eventType EventType, payload map[string]any) // appended to the forensic record after it is sealed. if state.terminated { r.mu.Unlock() - return + return false } journal := newEventJournal(r) delivery, deliver := journal.prepareLocked(state, runID, eventType, payload) publishTerminal := deliver && !delivery.dropped && IsTerminalEvent(eventType) r.mu.Unlock() if !deliver { - return + return false } if publishTerminal { - journal.publishTerminal(delivery) + persisted := journal.publishTerminalContext(ctx, delivery) r.conversationEventMu.Unlock() conversationLocked = false r.pruneCompletedRuns() - journal.dispatch(delivery) - return + journal.dispatchContext(ctx, delivery, requirePersistence) + return persisted } if delivery.dropped { r.conversationEventMu.Unlock() conversationLocked = false - journal.dispatch(delivery) - return + return journal.dispatchContext(ctx, delivery, requirePersistence) } - journal.dispatch(delivery) + dispatched := journal.dispatchContext(ctx, delivery, requirePersistence) r.conversationEventMu.Unlock() conversationLocked = false + return dispatched } // EmitEvent publishes an additive adapter-originated event through the run's @@ -5569,38 +5697,11 @@ func (r *Runner) storeCreateRun(run Run) { } } -// storeUpdateRun persists the current run state (status, output, error) to the store. -// Called from setStatus after each status transition. -func (r *Runner) storeUpdateRun(runID string) { - rc := r.configForRun(runID) - if rc.Store == nil { - return - } - r.mu.RLock() - state, ok := r.runs[runID] - if !ok { - r.mu.RUnlock() - return - } - run := state.run - r.mu.RUnlock() - - sr := runToStoreRun(run) - if err := rc.Store.UpdateRun(context.Background(), sr); err != nil { - if rc.Logger != nil { - rc.Logger.Error("store: UpdateRun failed", "run_id", runID, "error", err) - } - } -} - func shouldPersistWorkflowRecap(status RunStatus) bool { return status == RunStatusCompleted || status == RunStatusFailed || status == RunStatusCancelled } -// storeAppendEvent persists a single event to the store. -// Called from emit() after the event is appended to state.events. -// Executed outside the lock to avoid increasing lock hold time. -func (r *Runner) storeAppendEvent(ev Event, seq uint64) bool { +func (r *Runner) storeAppendEventContext(parent context.Context, ev Event, seq uint64) bool { rc := r.configForRun(ev.RunID) if rc.Store == nil { return true @@ -5617,7 +5718,7 @@ func (r *Runner) storeAppendEvent(ev Event, seq uint64) bool { Payload: string(payloadJSON), Timestamp: ev.Timestamp, } - ctx, cancel := context.WithTimeout(context.Background(), terminalEventStoreTimeout) + ctx, cancel := context.WithTimeout(parent, terminalEventStoreTimeout) defer cancel() if err := rc.Store.AppendEvent(ctx, se); err != nil { if rc.Logger != nil { diff --git a/internal/harness/runner_event_journal.go b/internal/harness/runner_event_journal.go index 2792dfa7..c4049506 100644 --- a/internal/harness/runner_event_journal.go +++ b/internal/harness/runner_event_journal.go @@ -1,6 +1,7 @@ package harness import ( + "context" "fmt" "time" @@ -72,6 +73,8 @@ func (j *eventJournal) prepareLocked(state *runState, runID string, eventType Ev state.recorderCh = nil state.recorderDone = nil state.closeRecorderOnce = nil + } else { + delivery.recorderCh = state.recorderCh } // Apply PII/secret redaction pipeline if configured. @@ -142,45 +145,16 @@ func (j *eventJournal) prepareLocked(state *runState, runID string, eventType Ev }) } - // Queue non-terminal recorder events while still holding the runner lock. - // Otherwise a terminal emit can close the recorder channel after this event - // has been appended to state.events but before dispatch() queues it, leaving - // the JSONL ledger shorter than the canonical in-memory history. - if !isTerminal && state.recorderCh != nil { - rev := rollout.RecordableEvent{ - ID: event.ID, - RunID: event.RunID, - Type: string(event.Type), - Timestamp: event.Timestamp, - Payload: event.Payload, - Seq: eventSeq, - } - if !safeRecorderSend(state.recorderCh, rev) { - if rc.Logger != nil { - rc.Logger.Error("rollout recorder: channel full, event dropped", - "run_id", runID, "event_type", string(eventType), "seq", eventSeq) - } - dropMarker := rollout.RecordableEvent{ - ID: fmt.Sprintf("%s:drop:%d", runID, eventSeq), - RunID: runID, - Type: string(EventRecorderDropDetected), - Timestamp: time.Now().UTC(), - Seq: eventSeq, - Payload: map[string]any{ - "dropped_event_id": event.ID, - "dropped_event_type": string(eventType), - "dropped_seq": eventSeq, - }, - } - safeRecorderSend(state.recorderCh, dropMarker) - } - } - return delivery, true } func (j *eventJournal) publishTerminal(delivery eventDispatch) { - if j.runner.storeAppendEvent(delivery.event, delivery.eventSeq) { + j.publishTerminalContext(context.Background(), delivery) +} + +func (j *eventJournal) publishTerminalContext(ctx context.Context, delivery eventDispatch) bool { + persisted := j.runner.storeAppendEventContext(ctx, delivery.event, delivery.eventSeq) + if persisted { j.runner.markTerminalEventPersisted(delivery.runID) } j.runner.recordConversationEvent(delivery.conversationID, delivery.event) @@ -188,31 +162,29 @@ func (j *eventJournal) publishTerminal(delivery eventDispatch) { for _, sub := range delivery.subscribers { j.runner.sendTerminalSubscriberEvent(sub.ch, sub.event) } + return persisted } func (j *eventJournal) dispatch(delivery eventDispatch) { + j.dispatchContext(context.Background(), delivery, false) +} + +func (j *eventJournal) dispatchContext(ctx context.Context, delivery eventDispatch, requirePersistence bool) bool { if delivery.dropped { if delivery.closeRecorder != nil { delivery.closeRecorder() } - return + // StorageModeNone is an intentional policy outcome, not a transient + // publication failure. Treat it as consumed so strict publishers do not + // retry a deliberately suppressed lifecycle event until its deadline. + return true } // Logger comes from the run's config snapshot when available so logging // stays consistent with the config the run started with. rc := j.runner.configForRun(delivery.runID) - if !IsTerminalEvent(delivery.eventType) { - j.runner.storeAppendEvent(delivery.event, delivery.eventSeq) - j.runner.recordConversationEvent(delivery.conversationID, delivery.event) - for _, sub := range delivery.subscribers { - j.runner.sendTerminalSubscriberEvent(sub.ch, sub.event) - } - } - - // Record to the JSONL rollout file via the per-run recorder goroutine. - // The goroutine owns all writes to the file and is the only entity that - // calls rec.Record / rec.Close, so no additional serialisation is needed. + // Recordable form shared by non-terminal queuing and terminal draining. rev := rollout.RecordableEvent{ ID: delivery.event.ID, RunID: delivery.event.RunID, @@ -221,6 +193,40 @@ func (j *eventJournal) dispatch(delivery eventDispatch) { Payload: delivery.event.Payload, Seq: delivery.eventSeq, } + + if !IsTerminalEvent(delivery.eventType) { + persisted := j.runner.storeAppendEventContext(ctx, delivery.event, delivery.eventSeq) + if (!persisted && requirePersistence) || ctx.Err() != nil { + j.discardPreparedEvent(delivery) + return false + } + j.runner.recordConversationEvent(delivery.conversationID, delivery.event) + for _, sub := range delivery.subscribers { + j.runner.sendTerminalSubscriberEvent(sub.ch, sub.event) + } + if delivery.recorderCh != nil { + if !safeRecorderSend(delivery.recorderCh, rev) { + if rc.Logger != nil { + rc.Logger.Error("rollout recorder: channel full, event dropped", + "run_id", delivery.runID, "event_type", string(delivery.eventType), "seq", delivery.eventSeq) + } + dropMarker := rollout.RecordableEvent{ + ID: fmt.Sprintf("%s:drop:%d", delivery.runID, delivery.eventSeq), + RunID: delivery.runID, + Type: string(EventRecorderDropDetected), + Timestamp: time.Now().UTC(), + Seq: delivery.eventSeq, + Payload: map[string]any{ + "dropped_event_id": delivery.event.ID, + "dropped_event_type": string(delivery.eventType), + "dropped_seq": delivery.eventSeq, + }, + } + safeRecorderSend(delivery.recorderCh, dropMarker) + } + } + } + if IsTerminalEvent(delivery.eventType) { if delivery.recorderCh != nil { sendTimer := time.NewTimer(recorderDrainTimeout) @@ -245,9 +251,33 @@ func (j *eventJournal) dispatch(delivery eventDispatch) { } } } - return + return true } - // Non-terminal recorder events are queued in prepareLocked while the runner - // lock is held so terminal close cannot overtake them. + // Non-terminal recorder events are queued above while emit still owns the + // conversation event lock, after context-bound persistence succeeds. That + // keeps terminal close ordered without recording an expired wait event. + return true +} + +func (j *eventJournal) discardPreparedEvent(delivery eventDispatch) { + j.runner.mu.Lock() + defer j.runner.mu.Unlock() + state, ok := j.runner.runs[delivery.runID] + if !ok { + return + } + // emit still owns conversationEventMu, so no later event can have been + // prepared while a strict append was in flight. Roll back both the final + // ledger entry and its sequence allocation to keep event IDs contiguous; + // run SSE reconnect treats the sequence as the visible history index. + last := len(state.events) - 1 + if last < 0 || state.events[last].ID != delivery.event.ID { + return + } + if state.nextEventSeq != delivery.eventSeq+1 { + return + } + state.events = state.events[:last] + state.nextEventSeq = delivery.eventSeq } diff --git a/internal/harness/runner_event_journal_test.go b/internal/harness/runner_event_journal_test.go index c2a3922e..0c83845f 100644 --- a/internal/harness/runner_event_journal_test.go +++ b/internal/harness/runner_event_journal_test.go @@ -2,11 +2,20 @@ package harness import ( "context" + "errors" "testing" "go-agent-harness/internal/store" ) +type failingOrdinaryEventStore struct { + *store.MemoryStore +} + +func (s *failingOrdinaryEventStore) AppendEvent(context.Context, *store.Event) error { + return errors.New("ordinary event persistence unavailable") +} + type terminalOrderingStore struct { *store.MemoryStore terminalAppendStarted chan struct{} @@ -165,3 +174,23 @@ func TestEventJournalDispatch_NonTerminalStoreAppendPrecedesSubscriberNotificati t.Fatalf("subscriber event type = %q, want %q", event.Type, EventRunStarted) } } + +func TestEventJournalDispatch_OrdinaryEventRemainsVisibleWhenStoreAppendFails(t *testing.T) { + t.Parallel() + + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{ + DefaultModel: "test-model", + Store: &failingOrdinaryEventStore{MemoryStore: store.NewMemoryStore()}, + }) + const runID = "run_ordinary_append_failure" + runner.runs[runID] = &runState{ + run: Run{ID: runID, ConversationID: "conv_ordinary_append_failure", Status: RunStatusRunning}, + } + + runner.emit(runID, EventToolCallStarted, map[string]any{"tool": "read"}) + + events := runner.getEvents(runID) + if len(events) != 1 || events[0].Type != EventToolCallStarted { + t.Fatalf("ordinary event was removed after non-fatal store failure: %+v", events) + } +} diff --git a/internal/harness/runner_execute_lifecycle_test.go b/internal/harness/runner_execute_lifecycle_test.go index 2b05fbdc..4c74c42a 100644 --- a/internal/harness/runner_execute_lifecycle_test.go +++ b/internal/harness/runner_execute_lifecycle_test.go @@ -7,12 +7,16 @@ package harness import ( "context" "encoding/json" + "errors" "strings" + "sync" "testing" "time" + "go-agent-harness/internal/forensics/redaction" htools "go-agent-harness/internal/harness/tools" om "go-agent-harness/internal/observationalmemory" + runstore "go-agent-harness/internal/store" ) // ------------------------------------------------------------------------- @@ -190,6 +194,941 @@ func TestExecuteLifecycle_AutoCompactAndMemorySnippetSameTurn(t *testing.T) { // - Event ordering is pinned precisely. // // ------------------------------------------------------------------------- +type gatedAskUserQuestionBroker struct { + inner *InMemoryAskUserQuestionBroker + entered chan struct{} + release chan struct{} +} + +type notifierIgnoringAskUserQuestionBroker struct { + mu sync.Mutex + pending htools.AskUserQuestionPending + hasPending bool + registered chan struct{} + returned chan struct{} + answerC chan map[string]string +} + +func newNotifierIgnoringAskUserQuestionBroker() *notifierIgnoringAskUserQuestionBroker { + return ¬ifierIgnoringAskUserQuestionBroker{ + registered: make(chan struct{}), + returned: make(chan struct{}), + answerC: make(chan map[string]string, 1), + } +} + +func (b *notifierIgnoringAskUserQuestionBroker) Ask( + ctx context.Context, + req htools.AskUserQuestionRequest, +) (map[string]string, time.Time, error) { + b.mu.Lock() + b.pending = htools.AskUserQuestionPending{ + RunID: req.RunID, + CallID: req.CallID, + Tool: htools.AskUserQuestionToolName, + Questions: req.Questions, + DeadlineAt: time.Now().UTC().Add(req.Timeout), + } + b.hasPending = true + b.mu.Unlock() + close(b.registered) + + select { + case answers := <-b.answerC: + close(b.returned) + return answers, time.Now().UTC(), nil + case <-ctx.Done(): + return nil, time.Time{}, ctx.Err() + } +} + +func (b *notifierIgnoringAskUserQuestionBroker) Pending(runID string) (htools.AskUserQuestionPending, bool) { + b.mu.Lock() + defer b.mu.Unlock() + if !b.hasPending || b.pending.RunID != runID { + return htools.AskUserQuestionPending{}, false + } + return b.pending, true +} + +func (b *notifierIgnoringAskUserQuestionBroker) Submit(runID string, answers map[string]string) error { + b.mu.Lock() + if !b.hasPending || b.pending.RunID != runID { + b.mu.Unlock() + return ErrNoPendingInput + } + b.hasPending = false + b.mu.Unlock() + b.answerC <- answers + return nil +} + +type waitingStatusBlockingStore struct { + *runstore.MemoryStore + once sync.Once + cancelOnce sync.Once + started chan struct{} + cancelledWait chan struct{} + release chan struct{} +} + +type transientWaitingStatusStore struct { + *runstore.MemoryStore + mu sync.Mutex + failureSurface string + waitAttempts int + firstFailed chan struct{} + secondAttempt chan struct{} +} + +func newTransientWaitingStatusStore(failureSurface string) *transientWaitingStatusStore { + return &transientWaitingStatusStore{ + MemoryStore: runstore.NewMemoryStore(), + failureSurface: failureSurface, + firstFailed: make(chan struct{}), + secondAttempt: make(chan struct{}), + } +} + +func (s *transientWaitingStatusStore) UpdateRun(ctx context.Context, run *runstore.Run) error { + if s.failureSurface == "UpdateRun" && run.Status == runstore.RunStatusWaitingForUser { + s.mu.Lock() + s.waitAttempts++ + attempt := s.waitAttempts + s.mu.Unlock() + switch attempt { + case 1: + close(s.firstFailed) + return errors.New("transient waiting status write failure") + case 2: + close(s.secondAttempt) + } + } + return s.MemoryStore.UpdateRun(ctx, run) +} + +func (s *transientWaitingStatusStore) AppendEvent(ctx context.Context, event *runstore.Event) error { + if s.failureSurface == "AppendEvent" && event.EventType == string(EventRunWaitingForUser) { + s.mu.Lock() + s.waitAttempts++ + attempt := s.waitAttempts + s.mu.Unlock() + switch attempt { + case 1: + close(s.firstFailed) + return errors.New("transient waiting event write failure") + case 2: + close(s.secondAttempt) + } + } + return s.MemoryStore.AppendEvent(ctx, event) +} + +type staleWaitRepairFailingStore struct { + *runstore.MemoryStore + once sync.Once + started chan struct{} + release chan struct{} + mu sync.Mutex + failedWrites int +} + +func newStaleWaitRepairFailingStore() *staleWaitRepairFailingStore { + return &staleWaitRepairFailingStore{ + MemoryStore: runstore.NewMemoryStore(), + started: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (s *staleWaitRepairFailingStore) UpdateRun(ctx context.Context, run *runstore.Run) error { + if run.Status == runstore.RunStatusWaitingForUser { + s.once.Do(func() { close(s.started) }) + <-s.release + return s.MemoryStore.UpdateRun(ctx, run) + } + if run.Status == runstore.RunStatusFailed { + s.mu.Lock() + s.failedWrites++ + attempt := s.failedWrites + s.mu.Unlock() + if attempt == 2 { + return errors.New("transient terminal status write failure") + } + } + return s.MemoryStore.UpdateRun(ctx, run) +} + +type waitingEventDeadlineStore struct { + *runstore.MemoryStore + once sync.Once + cancelOnce sync.Once + started chan struct{} + cancelled chan struct{} +} + +func newWaitingEventDeadlineStore() *waitingEventDeadlineStore { + return &waitingEventDeadlineStore{ + MemoryStore: runstore.NewMemoryStore(), + started: make(chan struct{}), + cancelled: make(chan struct{}), + } +} + +func (s *waitingEventDeadlineStore) AppendEvent(ctx context.Context, event *runstore.Event) error { + if event.EventType == string(EventRunWaitingForUser) { + s.once.Do(func() { close(s.started) }) + <-ctx.Done() + s.cancelOnce.Do(func() { close(s.cancelled) }) + return ctx.Err() + } + return s.MemoryStore.AppendEvent(ctx, event) +} + +func newWaitingStatusBlockingStore() *waitingStatusBlockingStore { + return &waitingStatusBlockingStore{ + MemoryStore: runstore.NewMemoryStore(), + started: make(chan struct{}), + cancelledWait: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (s *waitingStatusBlockingStore) UpdateRun(ctx context.Context, run *runstore.Run) error { + if run.Status == runstore.RunStatusWaitingForUser { + s.once.Do(func() { close(s.started) }) + select { + case <-s.release: + case <-ctx.Done(): + s.cancelOnce.Do(func() { close(s.cancelledWait) }) + return ctx.Err() + } + } + return s.MemoryStore.UpdateRun(ctx, run) +} + +func (b *gatedAskUserQuestionBroker) Ask(ctx context.Context, req htools.AskUserQuestionRequest) (map[string]string, time.Time, error) { + close(b.entered) + select { + case <-b.release: + case <-ctx.Done(): + return nil, time.Time{}, ctx.Err() + } + return b.inner.Ask(ctx, req) +} + +func (b *gatedAskUserQuestionBroker) Pending(runID string) (htools.AskUserQuestionPending, bool) { + return b.inner.Pending(runID) +} + +func (b *gatedAskUserQuestionBroker) Submit(runID string, answers map[string]string) error { + return b.inner.Submit(runID, answers) +} + +func TestExecuteLifecycle_WaitingForUserRequiresPendingInput(t *testing.T) { + t.Parallel() + + provider := &stubProvider{turns: []CompletionResult{ + { + ToolCalls: []ToolCall{{ + ID: "call_ask_pending", + Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Continue?","header":"Continue","options":[{"label":"Yes","description":"Continue"},{"label":"No","description":"Stop"}],"multiSelect":false}]}`, + }}, + }, + {Content: "continued"}, + }} + broker := &gatedAskUserQuestionBroker{ + inner: NewInMemoryAskUserQuestionBroker(time.Now), + entered: make(chan struct{}), + release: make(chan struct{}), + } + released := false + defer func() { + if !released { + close(broker.release) + } + }() + + const waitForUserTimeout = 10 * time.Second + runner := NewRunner(provider, NewDefaultRegistryWithOptions(t.TempDir(), DefaultRegistryOptions{ + ApprovalMode: ToolApprovalModeFullAuto, + AskUserBroker: broker, + AskUserTimeout: waitForUserTimeout, + }), RunnerConfig{ + DefaultModel: "gpt-5-nano", + MaxSteps: 4, + AskUserBroker: broker, + AskUserTimeout: waitForUserTimeout, + }) + + run, err := runner.StartRun(RunRequest{Prompt: "wait until pending exists"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + + select { + case <-broker.entered: + case <-time.After(waitForUserTimeout): + t.Fatal("timed out waiting for AskUserQuestion broker") + } + + state, ok := runner.GetRun(run.ID) + if !ok { + t.Fatal("run not found before pending registration") + } + if state.Status != RunStatusRunning { + t.Fatalf("status before pending registration = %q, want %q", state.Status, RunStatusRunning) + } + if _, err := runner.PendingInput(run.ID); err != ErrNoPendingInput { + t.Fatalf("PendingInput before registration error = %v, want %v", err, ErrNoPendingInput) + } + + close(broker.release) + released = true + + var pending htools.AskUserQuestionPending + deadline := time.Now().Add(waitForUserTimeout) + for { + pending, err = runner.PendingInput(run.ID) + if err == nil { + break + } + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for pending input: %v", err) + } + time.Sleep(time.Millisecond) + } + if pending.CallID != "call_ask_pending" { + t.Fatalf("pending call ID = %q, want call_ask_pending", pending.CallID) + } + + deadline = time.Now().Add(waitForUserTimeout) + for { + state, ok = runner.GetRun(run.ID) + if !ok { + t.Fatal("run not found after pending registration") + } + if state.Status == RunStatusWaitingForUser { + break + } + if time.Now().After(deadline) { + t.Fatalf("status after pending registration = %q, want %q", state.Status, RunStatusWaitingForUser) + } + time.Sleep(time.Millisecond) + } + + if err := runner.SubmitInput(run.ID, map[string]string{"Continue?": "Yes"}); err != nil { + t.Fatalf("SubmitInput: %v", err) + } + events, err := collectRunEvents(t, runner, run.ID) + if err != nil { + t.Fatalf("collectRunEvents: %v", err) + } + requireEventOrder(t, events, + "tool.call.started", + "run.waiting_for_user", + "run.resumed", + "run.completed", + ) + assertSingleWaitAndResume(t, events) +} + +func TestExecuteLifecycle_ObservesPendingInputWhenBrokerIgnoresNotifier(t *testing.T) { + t.Parallel() + + provider := &stubProvider{turns: []CompletionResult{ + { + ToolCalls: []ToolCall{{ + ID: "call_ask_ignored_notifier", + Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Continue?","header":"Continue","options":[{"label":"Yes","description":"Continue"},{"label":"No","description":"Stop"}],"multiSelect":false}]}`, + }}, + }, + {Content: "continued after fallback"}, + }} + broker := newNotifierIgnoringAskUserQuestionBroker() + submitted := false + t.Cleanup(func() { + if !submitted { + _ = broker.Submit("ignored", map[string]string{"Continue?": "Yes"}) + } + }) + const timeout = 2 * time.Second + runner := NewRunner(provider, NewDefaultRegistryWithOptions(t.TempDir(), DefaultRegistryOptions{ + ApprovalMode: ToolApprovalModeFullAuto, + AskUserBroker: broker, + AskUserTimeout: timeout, + }), RunnerConfig{ + DefaultModel: "gpt-5-nano", + MaxSteps: 4, + AskUserBroker: broker, + AskUserTimeout: timeout, + }) + + run, err := runner.StartRun(RunRequest{Prompt: "use a third-party ask broker"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + select { + case <-broker.registered: + case <-time.After(timeout): + t.Fatal("timed out waiting for third-party broker registration") + } + + deadline := time.Now().Add(500 * time.Millisecond) + for { + state, ok := runner.GetRun(run.ID) + if !ok { + t.Fatal("run not found") + } + if state.Status == RunStatusWaitingForUser { + break + } + if time.Now().After(deadline) { + _ = broker.Submit(run.ID, map[string]string{"Continue?": "Yes"}) + submitted = true + t.Fatalf("status = %q, want waiting_for_user after broker exposed pending input", state.Status) + } + time.Sleep(time.Millisecond) + } + if _, err := runner.PendingInput(run.ID); err != nil { + t.Fatalf("PendingInput: %v", err) + } + if err := runner.SubmitInput(run.ID, map[string]string{"Continue?": "Yes"}); err != nil { + t.Fatalf("SubmitInput: %v", err) + } + submitted = true + events, err := collectRunEvents(t, runner, run.ID) + if err != nil { + t.Fatalf("collectRunEvents: %v", err) + } + requireEventOrder(t, events, + "run.waiting_for_user", + "run.resumed", + "run.completed", + ) + assertSingleWaitAndResume(t, events) +} + +func TestExecuteLifecycle_ObserverFinishesStartedPendingPublicationAfterToolReturns(t *testing.T) { + t.Parallel() + + provider := &stubProvider{turns: []CompletionResult{ + { + ToolCalls: []ToolCall{{ + ID: "call_ask_observer_inflight", + Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Continue?","header":"Continue","options":[{"label":"Yes","description":"Continue"},{"label":"No","description":"Stop"}],"multiSelect":false}]}`, + }}, + }, + {Content: "continued after observer publication"}, + }} + broker := newNotifierIgnoringAskUserQuestionBroker() + persistence := newWaitingStatusBlockingStore() + released := false + t.Cleanup(func() { + if !released { + close(persistence.release) + } + }) + const timeout = 2 * time.Second + runner := NewRunner(provider, NewDefaultRegistryWithOptions(t.TempDir(), DefaultRegistryOptions{ + ApprovalMode: ToolApprovalModeFullAuto, + AskUserBroker: broker, + AskUserTimeout: timeout, + }), RunnerConfig{ + DefaultModel: "gpt-5-nano", + MaxSteps: 4, + AskUserBroker: broker, + AskUserTimeout: timeout, + Store: persistence, + }) + + run, err := runner.StartRun(RunRequest{Prompt: "answer while observer publication is blocked"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + select { + case <-persistence.started: + case <-time.After(timeout): + t.Fatal("timed out waiting for observer pending publication") + } + if err := runner.SubmitInput(run.ID, map[string]string{"Continue?": "Yes"}); err != nil { + t.Fatalf("SubmitInput: %v", err) + } + select { + case <-broker.returned: + case <-time.After(timeout): + t.Fatal("broker did not return accepted answer") + } + + select { + case <-persistence.cancelledWait: + t.Fatal("observer stop cancelled a pending publication that had already started") + case <-time.After(25 * time.Millisecond): + } + close(persistence.release) + released = true + + events, err := collectRunEvents(t, runner, run.ID) + if err != nil { + t.Fatalf("collectRunEvents: %v", err) + } + requireEventOrder(t, events, + "run.waiting_for_user", + "run.resumed", + "run.completed", + ) + assertSingleWaitAndResume(t, events) +} + +func TestExecuteLifecycle_PendingPublicationRetriesAfterTransientPersistenceFailure(t *testing.T) { + for _, failureSurface := range []string{"UpdateRun", "AppendEvent"} { + failureSurface := failureSurface + t.Run(failureSurface, func(t *testing.T) { + t.Parallel() + + provider := &stubProvider{turns: []CompletionResult{ + { + ToolCalls: []ToolCall{{ + ID: "call_ask_retry_pending", + Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Continue?","header":"Continue","options":[{"label":"Yes","description":"Continue"},{"label":"No","description":"Stop"}],"multiSelect":false}]}`, + }}, + }, + {Content: "continued after retry"}, + }} + broker := NewInMemoryAskUserQuestionBroker(time.Now) + persistence := newTransientWaitingStatusStore(failureSurface) + const timeout = 2 * time.Second + runner := NewRunner(provider, NewDefaultRegistryWithOptions(t.TempDir(), DefaultRegistryOptions{ + ApprovalMode: ToolApprovalModeFullAuto, + AskUserBroker: broker, + AskUserTimeout: timeout, + }), RunnerConfig{ + DefaultModel: "gpt-5-nano", + MaxSteps: 4, + AskUserBroker: broker, + AskUserTimeout: timeout, + Store: persistence, + }) + + run, err := runner.StartRun(RunRequest{Prompt: "retry transient pending persistence"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + select { + case <-persistence.firstFailed: + case <-time.After(timeout): + t.Fatal("timed out waiting for first pending persistence failure") + } + select { + case <-persistence.secondAttempt: + case <-time.After(250 * time.Millisecond): + _ = runner.SubmitInput(run.ID, map[string]string{"Continue?": "Yes"}) + t.Fatalf("pending publication was not retried after transient %s failure", failureSurface) + } + if err := runner.SubmitInput(run.ID, map[string]string{"Continue?": "Yes"}); err != nil { + t.Fatalf("SubmitInput: %v", err) + } + + events, err := collectRunEvents(t, runner, run.ID) + if err != nil { + t.Fatalf("collectRunEvents: %v", err) + } + requireEventOrder(t, events, + "run.waiting_for_user", + "run.resumed", + "run.completed", + ) + assertSingleWaitAndResume(t, events) + }) + } +} + +func TestExecuteLifecycle_RedactionDroppedPendingEventDoesNotBlockAcceptedAnswer(t *testing.T) { + t.Parallel() + + provider := &stubProvider{turns: []CompletionResult{ + { + ToolCalls: []ToolCall{{ + ID: "call_ask_redaction_drop", + Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Continue?","header":"Continue","options":[{"label":"Yes","description":"Continue"},{"label":"No","description":"Stop"}],"multiSelect":false}]}`, + }}, + }, + {Content: "continued with waiting event redacted"}, + }} + broker := NewInMemoryAskUserQuestionBroker(time.Now) + pipeline := redaction.NewPipeline(nil, redaction.EventClassConfig{ + string(EventRunWaitingForUser): redaction.StorageModeNone, + }) + const timeout = time.Second + runner := NewRunner(provider, NewDefaultRegistryWithOptions(t.TempDir(), DefaultRegistryOptions{ + ApprovalMode: ToolApprovalModeFullAuto, + AskUserBroker: broker, + AskUserTimeout: timeout, + }), RunnerConfig{ + DefaultModel: "gpt-5-nano", + MaxSteps: 4, + AskUserBroker: broker, + AskUserTimeout: timeout, + RedactionPipeline: pipeline, + }) + + run, err := runner.StartRun(RunRequest{Prompt: "accept input with dropped waiting event"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + deadline := time.Now().Add(timeout) + for { + if _, err := runner.PendingInput(run.ID); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("timed out waiting for pending input") + } + time.Sleep(time.Millisecond) + } + // Give the fallback observer time to see the already-published pending + // record. A deliberate redaction drop must complete that observer instead + // of making it retry until the question deadline. + time.Sleep(25 * time.Millisecond) + if err := runner.SubmitInput(run.ID, map[string]string{"Continue?": "Yes"}); err != nil { + t.Fatalf("SubmitInput: %v", err) + } + completionDeadline := time.Now().Add(250 * time.Millisecond) + for { + state, ok := runner.GetRun(run.ID) + if ok && state.Status == RunStatusCompleted { + break + } + if time.Now().After(completionDeadline) { + t.Fatal("accepted answer remained blocked by retries of a deliberately dropped waiting event") + } + time.Sleep(time.Millisecond) + } + for _, event := range runner.getEvents(run.ID) { + if event.Type == EventRunWaitingForUser { + t.Fatal("redaction-dropped waiting event became visible") + } + } +} + +func TestExecuteLifecycle_WaitEventPrecedesQuickAnswerWhenStatusPersistenceBlocks(t *testing.T) { + t.Parallel() + + provider := &stubProvider{turns: []CompletionResult{ + { + ToolCalls: []ToolCall{{ + ID: "call_ask_quick_answer", + Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Continue?","header":"Continue","options":[{"label":"Yes","description":"Continue"},{"label":"No","description":"Stop"}],"multiSelect":false}]}`, + }}, + }, + {Content: "continued"}, + }} + broker := NewInMemoryAskUserQuestionBroker(time.Now) + persistence := newWaitingStatusBlockingStore() + released := false + t.Cleanup(func() { + if !released { + close(persistence.release) + } + }) + + const waitForUserTimeout = 10 * time.Second + runner := NewRunner(provider, NewDefaultRegistryWithOptions(t.TempDir(), DefaultRegistryOptions{ + ApprovalMode: ToolApprovalModeFullAuto, + AskUserBroker: broker, + }), RunnerConfig{ + DefaultModel: "gpt-5-nano", + MaxSteps: 4, + AskUserBroker: broker, + AskUserTimeout: waitForUserTimeout, + Store: persistence, + }) + + run, err := runner.StartRun(RunRequest{Prompt: "accept a quick answer"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + + select { + case <-persistence.started: + case <-time.After(waitForUserTimeout): + t.Fatal("timed out waiting for waiting_for_user persistence") + } + if _, err := runner.PendingInput(run.ID); err != nil { + t.Fatalf("PendingInput while status persistence is blocked: %v", err) + } + if err := runner.SubmitInput(run.ID, map[string]string{"Continue?": "Yes"}); err != nil { + t.Fatalf("SubmitInput: %v", err) + } + + time.Sleep(50 * time.Millisecond) + for _, event := range runner.getEvents(run.ID) { + if event.Type == EventRunResumed { + t.Fatal("run.resumed overtook the blocked run.waiting_for_user publication") + } + } + + close(persistence.release) + released = true + + events, err := collectRunEvents(t, runner, run.ID) + if err != nil { + t.Fatalf("collectRunEvents: %v", err) + } + requireEventOrder(t, events, + "run.waiting_for_user", + "run.resumed", + "run.completed", + ) +} + +func TestExecuteLifecycle_LateWaitPersistenceCannotReplaceTerminalStatus(t *testing.T) { + t.Parallel() + + provider := &stubProvider{turns: []CompletionResult{{ + ToolCalls: []ToolCall{{ + ID: "call_ask_timeout", + Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Continue?","header":"Continue","options":[{"label":"Yes","description":"Continue"},{"label":"No","description":"Stop"}],"multiSelect":false}]}`, + }}, + }}} + broker := NewInMemoryAskUserQuestionBroker(time.Now) + persistence := newWaitingStatusBlockingStore() + released := false + t.Cleanup(func() { + if !released { + close(persistence.release) + } + }) + const timeout = 150 * time.Millisecond + runner := NewRunner(provider, NewDefaultRegistryWithOptions(t.TempDir(), DefaultRegistryOptions{ + ApprovalMode: ToolApprovalModeFullAuto, + AskUserBroker: broker, + AskUserTimeout: timeout, + }), RunnerConfig{ + DefaultModel: "gpt-5-nano", + MaxSteps: 2, + AskUserBroker: broker, + AskUserTimeout: timeout, + Store: persistence, + }) + + run, err := runner.StartRun(RunRequest{Prompt: "time out while persistence is blocked"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + <-persistence.started + if _, err := collectRunEvents(t, runner, run.ID); err != nil { + t.Fatalf("collectRunEvents: %v", err) + } + select { + case <-persistence.cancelledWait: + case <-time.After(time.Second): + t.Fatal("waiting status persistence outlived the pending notifier context") + } + close(persistence.release) + released = true + + deadline := time.Now().Add(time.Second) + for { + stored, err := persistence.GetRun(context.Background(), run.ID) + if err != nil { + t.Fatalf("GetRun: %v", err) + } + if stored.Status == runstore.RunStatusFailed { + break + } + if time.Now().After(deadline) { + t.Fatalf("durable status = %q, want failed", stored.Status) + } + time.Sleep(time.Millisecond) + } +} + +func TestExecuteLifecycle_ExpiredWaitEventIsNotPublishedAfterBlockedAppend(t *testing.T) { + t.Parallel() + + provider := &stubProvider{turns: []CompletionResult{{ + ToolCalls: []ToolCall{{ + ID: "call_ask_event_deadline", + Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Continue?","header":"Continue","options":[{"label":"Yes","description":"Continue"},{"label":"No","description":"Stop"}],"multiSelect":false}]}`, + }}, + }}} + broker := NewInMemoryAskUserQuestionBroker(time.Now) + persistence := newWaitingEventDeadlineStore() + const timeout = 100 * time.Millisecond + runner := NewRunner(provider, NewDefaultRegistryWithOptions(t.TempDir(), DefaultRegistryOptions{ + ApprovalMode: ToolApprovalModeFullAuto, + AskUserBroker: broker, + AskUserTimeout: timeout, + }), RunnerConfig{ + DefaultModel: "gpt-5-nano", + MaxSteps: 2, + AskUserBroker: broker, + AskUserTimeout: timeout, + Store: persistence, + }) + + run, err := runner.StartRun(RunRequest{Prompt: "expire while waiting event append is blocked"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + select { + case <-persistence.started: + case <-time.After(time.Second): + t.Fatal("timed out waiting for waiting event append") + } + select { + case <-persistence.cancelled: + case <-time.After(time.Second): + t.Fatal("waiting event append did not honor notifier deadline") + } + events, err := collectRunEvents(t, runner, run.ID) + if err != nil { + t.Fatalf("collectRunEvents: %v", err) + } + for _, event := range events { + if event.Type == EventRunWaitingForUser { + t.Fatal("expired run.waiting_for_user was published after event append deadline") + } + } +} + +func TestExecuteLifecycle_StaleWaitCannotSurviveFailedCorrectiveWrite(t *testing.T) { + t.Parallel() + + provider := &stubProvider{turns: []CompletionResult{{ + ToolCalls: []ToolCall{{ + ID: "call_ask_stale_repair", + Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Continue?","header":"Continue","options":[{"label":"Yes","description":"Continue"},{"label":"No","description":"Stop"}],"multiSelect":false}]}`, + }}, + }}} + broker := NewInMemoryAskUserQuestionBroker(time.Now) + persistence := newStaleWaitRepairFailingStore() + released := false + t.Cleanup(func() { + if !released { + close(persistence.release) + } + }) + const timeout = 50 * time.Millisecond + runner := NewRunner(provider, NewDefaultRegistryWithOptions(t.TempDir(), DefaultRegistryOptions{ + ApprovalMode: ToolApprovalModeFullAuto, + AskUserBroker: broker, + AskUserTimeout: timeout, + }), RunnerConfig{ + DefaultModel: "gpt-5-nano", + MaxSteps: 2, + AskUserBroker: broker, + AskUserTimeout: timeout, + Store: persistence, + }) + + run, err := runner.StartRun(RunRequest{Prompt: "preserve terminal state after a stale wait write"}) + if err != nil { + t.Fatalf("StartRun: %v", err) + } + select { + case <-persistence.started: + case <-time.After(time.Second): + t.Fatal("timed out waiting for waiting status persistence") + } + time.Sleep(timeout + 50*time.Millisecond) + close(persistence.release) + released = true + if _, err := collectRunEvents(t, runner, run.ID); err != nil { + t.Fatalf("collectRunEvents: %v", err) + } + + deadline := time.Now().Add(time.Second) + for { + stored, err := persistence.GetRun(context.Background(), run.ID) + if err != nil { + t.Fatalf("GetRun: %v", err) + } + if stored.Status == runstore.RunStatusFailed { + break + } + if time.Now().After(deadline) { + t.Fatalf("durable status = %q, want failed after stale wait write", stored.Status) + } + time.Sleep(time.Millisecond) + } +} + +func TestSetStatusContext_DelayedWaitCannotDowngradeTerminalRun(t *testing.T) { + t.Parallel() + + persistence := runstore.NewMemoryStore() + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{Store: persistence}) + const runID = "run-delayed-wait-after-terminal" + now := time.Now().UTC() + runner.mu.Lock() + runner.runs[runID] = &runState{run: Run{ + ID: runID, + Status: RunStatusRunning, + CreatedAt: now, + UpdatedAt: now, + }} + runner.mu.Unlock() + if err := persistence.CreateRun(context.Background(), runToStoreRun(Run{ + ID: runID, + Status: RunStatusRunning, + CreatedAt: now, + UpdatedAt: now, + })); err != nil { + t.Fatalf("CreateRun: %v", err) + } + + releaseWait := make(chan struct{}) + waitResult := make(chan bool, 1) + go func() { + <-releaseWait + waitResult <- runner.setStatusAndEmitContext( + context.Background(), + runID, + RunStatusWaitingForUser, + "", + "", + EventRunWaitingForUser, + map[string]any{"call_id": "late"}, + ) + }() + runner.setStatus(runID, RunStatusFailed, "", "terminal") + close(releaseWait) + if published := <-waitResult; published { + t.Fatal("delayed waiting status was accepted after terminal state") + } + + inMemory, ok := runner.GetRun(runID) + if !ok { + t.Fatal("run not found") + } + if inMemory.Status != RunStatusFailed { + t.Fatalf("in-memory status = %q, want failed", inMemory.Status) + } + for _, event := range runner.getEvents(runID) { + if event.Type == EventRunWaitingForUser { + t.Fatal("delayed notifier emitted waiting event after terminal state") + } + } + stored, err := persistence.GetRun(context.Background(), runID) + if err != nil { + t.Fatalf("GetRun: %v", err) + } + if stored.Status != runstore.RunStatusFailed { + t.Fatalf("durable status = %q, want failed", stored.Status) + } +} + func TestExecuteLifecycle_WaitForUserFlowEventOrderAndStateRestoration(t *testing.T) { t.Parallel() @@ -299,6 +1238,23 @@ func TestExecuteLifecycle_WaitForUserFlowEventOrderAndStateRestoration(t *testin "assistant.message", "run.completed", ) + assertSingleWaitAndResume(t, events) +} + +func assertSingleWaitAndResume(t *testing.T, events []Event) { + t.Helper() + var waits, resumes int + for _, event := range events { + switch event.Type { + case EventRunWaitingForUser: + waits++ + case EventRunResumed: + resumes++ + } + } + if waits != 1 || resumes != 1 { + t.Fatalf("wait/resume event counts = %d/%d, want exactly 1/1", waits, resumes) + } } // ------------------------------------------------------------------------- diff --git a/internal/harness/runner_orchestration_test.go b/internal/harness/runner_orchestration_test.go index 1c990763..8c768b58 100644 --- a/internal/harness/runner_orchestration_test.go +++ b/internal/harness/runner_orchestration_test.go @@ -6,9 +6,22 @@ import ( "testing" "time" + "go-agent-harness/internal/checkpoints" htools "go-agent-harness/internal/harness/tools" ) +type resolvedCheckpointAskBroker struct{} + +func (resolvedCheckpointAskBroker) Ask(context.Context, htools.AskUserQuestionRequest) (map[string]string, time.Time, error) { + return nil, time.Time{}, nil +} +func (resolvedCheckpointAskBroker) Pending(string) (htools.AskUserQuestionPending, bool) { + return htools.AskUserQuestionPending{}, false +} +func (resolvedCheckpointAskBroker) Submit(string, map[string]string) error { + return checkpoints.ErrAlreadyResolved +} + func TestSubmitInput_MapsBrokerValidationFailure(t *testing.T) { t.Parallel() @@ -67,6 +80,25 @@ func TestSubmitInput_MapsMissingPendingQuestion(t *testing.T) { } } +func TestSubmitInput_MapsAlreadyResolvedCheckpointToNoPendingInput(t *testing.T) { + t.Parallel() + + runner := NewRunner(&stubProvider{}, NewRegistry(), RunnerConfig{ + DefaultModel: "gpt-4.1-mini", + MaxSteps: 1, + AskUserBroker: resolvedCheckpointAskBroker{}, + }) + const runID = "run-submit-resolved" + runner.mu.Lock() + runner.runs[runID] = &runState{run: Run{ID: runID}} + runner.mu.Unlock() + + err := runner.SubmitInput(runID, map[string]string{"Where next?": "Docs"}) + if !errors.Is(err, ErrNoPendingInput) { + t.Fatalf("SubmitInput error = %v, want %v", err, ErrNoPendingInput) + } +} + func TestWaitForTerminalResult_UsesTerminalHistory(t *testing.T) { t.Parallel() diff --git a/internal/harness/runner_step_engine.go b/internal/harness/runner_step_engine.go index 4a8f6efa..c4c02d9d 100644 --- a/internal/harness/runner_step_engine.go +++ b/internal/harness/runner_step_engine.go @@ -767,6 +767,7 @@ func (se *stepEngine) run() { callArgs json.RawMessage toolCtx context.Context waitingForUser bool + publishPending askUserPendingPublisher rewindPointID string } @@ -1096,18 +1097,34 @@ func (se *stepEngine) run() { // status-restoring code below, so the run kept executing while // clients saw it as blocked on input that would never be asked for. waitingForUser := false + var pendingNotifier htools.AskUserQuestionPendingNotifier + var publishPending askUserPendingPublisher if call.Name == htools.AskUserQuestionToolName { - questions, err := htools.ParseAskUserQuestionArgs(callArgs) + _, err := htools.ParseAskUserQuestionArgs(callArgs) if err == nil { waitingForUser = true - deadlineAt := time.Now().UTC().Add(rc.AskUserTimeout) - r.setStatus(runID, RunStatusWaitingForUser, "", "") - r.emit(runID, EventRunWaitingForUser, map[string]any{ - "call_id": call.ID, - "tool": call.Name, - "questions": questions, - "deadline_at": deadlineAt, - }) + publication := &askUserPendingPublication{} + publishPending = func(notifyCtx context.Context, pending htools.AskUserQuestionPending) bool { + return publication.publish(notifyCtx, func() bool { + return r.setStatusAndEmitContext( + notifyCtx, + runID, + RunStatusWaitingForUser, + "", + "", + EventRunWaitingForUser, + map[string]any{ + "call_id": pending.CallID, + "tool": pending.Tool, + "questions": pending.Questions, + "deadline_at": pending.DeadlineAt, + }, + ) + }) + } + pendingNotifier = func(notifyCtx context.Context, pending htools.AskUserQuestionPending) { + publishPending(notifyCtx, pending) + } } } @@ -1116,6 +1133,9 @@ func (se *stepEngine) run() { toolCtx = context.WithValue(toolCtx, htools.ContextKeyPlanModeGate, runPlanModeGate{runner: r, runID: runID}) toolCtx = context.WithValue(toolCtx, htools.ContextKeyToolCallID, call.ID) toolCtx = context.WithValue(toolCtx, htools.ContextKeyRunMetadata, meta) + if pendingNotifier != nil { + toolCtx = htools.WithAskUserQuestionPendingNotifier(toolCtx, pendingNotifier) + } toolCtx = htools.WithSandboxScope(toolCtx, effectiveSandboxScope) // Extra directory roots granted on the run request (TUI /add-dir) // ride the same per-call context so file-tool confinement permits @@ -1173,6 +1193,7 @@ func (se *stepEngine) run() { callArgs: callArgs, toolCtx: toolCtx, waitingForUser: waitingForUser, + publishPending: publishPending, }) } @@ -1198,7 +1219,17 @@ func (se *stepEngine) run() { } } start := time.Now() + stopPendingObserver := func() {} + if pe.waitingForUser { + stopPendingObserver = r.observeAskUserPending( + pe.toolCtx, + runID, + rc.AskUserBroker, + pe.publishPending, + ) + } out, err := runTools.Execute(pe.toolCtx, pe.call.Name, pe.callArgs) + stopPendingObserver() if err == nil && pe.rewindPointID != "" { if rewind, ok := rc.ConversationStore.(RewindStore); ok { _ = FinalizeRewindPoint(pe.toolCtx, rewind, pe.rewindPointID, rc.WorkspaceBaseOptions.RepoPath) @@ -1460,3 +1491,95 @@ func (se *stepEngine) run() { r.failRunMaxSteps(runID, effectiveMaxSteps) } } + +type askUserPendingPublisher func(context.Context, htools.AskUserQuestionPending) bool + +// askUserPendingPublication serializes the broker callback and fallback +// observer, but only consumes the publication after status and event +// persistence both succeed. A transient failure therefore remains retryable. +type askUserPendingPublication struct { + gate contextMutex + published bool +} + +func (p *askUserPendingPublication) publish(ctx context.Context, publish func() bool) bool { + if ctx == nil { + ctx = context.Background() + } + if err := p.gate.lock(ctx); err != nil { + return false + } + defer p.gate.unlock() + if p.published { + return true + } + if ctx.Err() != nil || !publish() { + return false + } + p.published = true + return true +} + +func (r *Runner) observeAskUserPending( + ctx context.Context, + runID string, + broker htools.AskUserQuestionBroker, + publish askUserPendingPublisher, +) func() { + if broker == nil || publish == nil { + return func() {} + } + observerCtx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + var lifecycleMu sync.Mutex + started := false + stopped := false + go func() { + defer close(done) + ticker := time.NewTicker(time.Millisecond) + defer ticker.Stop() + for { + if pending, ok := broker.Pending(runID); ok { + lifecycleMu.Lock() + if stopped { + lifecycleMu.Unlock() + return + } + started = true + lifecycleMu.Unlock() + + notifyCtx := observerCtx + notifyCancel := func() {} + if !pending.DeadlineAt.IsZero() { + notifyCtx, notifyCancel = context.WithDeadline(observerCtx, pending.DeadlineAt) + } + defer notifyCancel() + for { + if publish(notifyCtx, pending) { + return + } + select { + case <-notifyCtx.Done(): + return + case <-ticker.C: + } + } + } + select { + case <-observerCtx.Done(): + return + case <-ticker.C: + } + } + }() + return func() { + lifecycleMu.Lock() + if !started { + stopped = true + cancel() + } + lifecycleMu.Unlock() + <-done + cancel() + } +} diff --git a/internal/harness/tools/core/ask_user_question.go b/internal/harness/tools/core/ask_user_question.go index 8f249421..8532807f 100644 --- a/internal/harness/tools/core/ask_user_question.go +++ b/internal/harness/tools/core/ask_user_question.go @@ -77,6 +77,7 @@ func AskUserQuestionTool(broker tools.AskUserQuestionBroker, timeout time.Durati CallID: callID, Questions: questions, Timeout: timeout, + OnPending: tools.AskUserQuestionPendingNotifierFromContext(ctx), }) if err != nil { return "", err diff --git a/internal/harness/tools/core/core_test.go b/internal/harness/tools/core/core_test.go index a9c854de..f7bb76aa 100644 --- a/internal/harness/tools/core/core_test.go +++ b/internal/harness/tools/core/core_test.go @@ -949,6 +949,14 @@ func (s *askBrokerStub) Ask(_ context.Context, req tools.AskUserQuestionRequest) if s.askErr != nil { return nil, time.Time{}, s.askErr } + if req.OnPending != nil { + req.OnPending(context.Background(), tools.AskUserQuestionPending{ + RunID: req.RunID, + CallID: req.CallID, + Tool: tools.AskUserQuestionToolName, + Questions: req.Questions, + }) + } return s.askAnswers, time.Now().UTC(), nil } @@ -973,6 +981,10 @@ func TestAskUserQuestionToolReturnsQuestionsAndAnswers(t *testing.T) { ctx := context.WithValue(context.Background(), tools.ContextKeyRunID, "run_123") ctx = context.WithValue(ctx, tools.ContextKeyToolCallID, "call_123") + notified := false + ctx = tools.WithAskUserQuestionPendingNotifier(ctx, func(_ context.Context, pending tools.AskUserQuestionPending) { + notified = pending.RunID == "run_123" && pending.CallID == "call_123" + }) out, err := tool.Handler(ctx, json.RawMessage(`{"questions":[{"question":"Where next?","header":"Next","options":[{"label":"Docs","description":"Open docs"},{"label":"Code","description":"Open code"}],"multiSelect":false}]}`)) if err != nil { @@ -982,6 +994,9 @@ func TestAskUserQuestionToolReturnsQuestionsAndAnswers(t *testing.T) { if broker.lastReq.RunID != "run_123" || broker.lastReq.CallID != "call_123" { t.Fatalf("unexpected request ids: %+v", broker.lastReq) } + if !notified { + t.Fatal("AskUserQuestion tool did not forward the pending notifier") + } var payload map[string]any if err := json.Unmarshal([]byte(out), &payload); err != nil { diff --git a/internal/harness/tools/types.go b/internal/harness/tools/types.go index 6358b448..a60eae71 100644 --- a/internal/harness/tools/types.go +++ b/internal/harness/tools/types.go @@ -540,6 +540,14 @@ type AskUserQuestionRequest struct { CallID string Questions []AskUserQuestion Timeout time.Duration + // OnPending is started after the question is readable through Pending. Ask + // does not consume an answer until notification finishes, so wait-state + // publication cannot be overtaken by a quick submission. Its context + // expires with the question deadline. Implementations MUST honor its + // cancellation in every blocking persistence or publication step and + // return promptly; Ask deliberately preserves an accepted answer instead + // of synthesizing a timeout when publication is still running. + OnPending AskUserQuestionPendingNotifier } type AskUserQuestionPending struct { @@ -550,6 +558,8 @@ type AskUserQuestionPending struct { DeadlineAt time.Time `json:"deadline_at"` } +type AskUserQuestionPendingNotifier func(context.Context, AskUserQuestionPending) + type AskUserQuestionBroker interface { Ask(ctx context.Context, req AskUserQuestionRequest) (answers map[string]string, answeredAt time.Time, err error) Pending(runID string) (AskUserQuestionPending, bool) @@ -567,8 +577,21 @@ const ContextKeyOutputStreamer contextKey = "output_streamer" const ContextKeyMessageReplacer contextKey = "message_replacer" const ContextKeySandboxScope contextKey = "sandbox_scope" const ContextKeyPlanModeGate contextKey = "plan_mode_gate" +const contextKeyAskUserQuestionPendingNotifier contextKey = "ask_user_question_pending_notifier" const contextKeyForkDepth contextKey = "fork_depth" +func WithAskUserQuestionPendingNotifier(ctx context.Context, notifier AskUserQuestionPendingNotifier) context.Context { + return context.WithValue(ctx, contextKeyAskUserQuestionPendingNotifier, notifier) +} + +func AskUserQuestionPendingNotifierFromContext(ctx context.Context) AskUserQuestionPendingNotifier { + if ctx == nil { + return nil + } + notifier, _ := ctx.Value(contextKeyAskUserQuestionPendingNotifier).(AskUserQuestionPendingNotifier) + return notifier +} + // DefaultMaxForkDepth is the maximum recursion depth for spawned subagents. // Agents at depth >= DefaultMaxForkDepth may not spawn further children. const DefaultMaxForkDepth = 5 diff --git a/internal/server/http_checkpoints.go b/internal/server/http_checkpoints.go index bc9e42ff..4032a86c 100644 --- a/internal/server/http_checkpoints.go +++ b/internal/server/http_checkpoints.go @@ -3,6 +3,7 @@ package server import ( "context" "encoding/json" + "errors" "fmt" "net/http" "strings" @@ -164,6 +165,10 @@ func (s *Server) handleResumeCheckpoint(w http.ResponseWriter, r *http.Request, writeError(w, http.StatusNotFound, "not_found", fmt.Sprintf("checkpoint %q not found", checkpointID)) return } + if errors.Is(err, checkpoints.ErrAlreadyResolved) { + writeError(w, http.StatusConflict, "already_resolved", "checkpoint is already resolved") + return + } writeError(w, http.StatusInternalServerError, "internal_error", err.Error()) return } diff --git a/internal/server/http_checkpoints_test.go b/internal/server/http_checkpoints_test.go index c855fd64..f001757f 100644 --- a/internal/server/http_checkpoints_test.go +++ b/internal/server/http_checkpoints_test.go @@ -62,3 +62,78 @@ func TestHandleCheckpointResume(t *testing.T) { t.Fatalf("response status = %v, want resumed", body["status"]) } } + +func TestHandleCheckpointResumeAlreadyResolvedIsConflictAndDoesNotMutate(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + resolve func(*checkpoints.Service, string) error + }{ + {name: "resumed", resolve: func(s *checkpoints.Service, id string) error { + return s.Resume(context.Background(), id, map[string]any{"decision": "first"}) + }}, + {name: "expired", resolve: func(s *checkpoints.Service, id string) error { + return s.Expire(context.Background(), id) + }}, + {name: "denied", resolve: func(s *checkpoints.Service, id string) error { + return s.Deny(context.Background(), id) + }}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + now := time.Date(2026, 4, 5, 12, 0, 0, 0, time.UTC) + service := checkpoints.NewService(checkpoints.NewMemoryStore(), func() time.Time { return now }) + record, err := service.Create(context.Background(), checkpoints.CreateRequest{ + Kind: checkpoints.KindExternalResume, DeadlineAt: now.Add(time.Minute), + }) + if err != nil { + t.Fatalf("Create: %v", err) + } + if err := tt.resolve(service, record.ID); err != nil { + t.Fatalf("resolve: %v", err) + } + before, err := service.Get(context.Background(), record.ID) + if err != nil { + t.Fatalf("Get before: %v", err) + } + + ts := httptest.NewServer(NewWithOptions(ServerOptions{ + AuthDisabled: true, + Checkpoints: service, + })) + defer ts.Close() + resp, err := http.Post( + ts.URL+"/v1/checkpoints/"+record.ID+"/resume", + "application/json", + bytes.NewBufferString(`{"payload":{"decision":"second"}}`), + ) + if err != nil { + t.Fatalf("POST second resume: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusConflict { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusConflict) + } + var body struct { + Error struct { + Code string `json:"code"` + } `json:"error"` + } + if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { + t.Fatalf("decode conflict: %v", err) + } + if body.Error.Code != "already_resolved" { + t.Fatalf("code = %q, want already_resolved", body.Error.Code) + } + after, err := service.Get(context.Background(), record.ID) + if err != nil { + t.Fatalf("Get after: %v", err) + } + if after.Status != before.Status || after.ResumePayload != before.ResumePayload || !after.UpdatedAt.Equal(before.UpdatedAt) { + t.Fatalf("checkpoint mutated: before=%+v after=%+v", before, after) + } + }) + } +} diff --git a/internal/server/http_test.go b/internal/server/http_test.go index e4e373e5..4dc6d12f 100644 --- a/internal/server/http_test.go +++ b/internal/server/http_test.go @@ -14,7 +14,9 @@ import ( "testing" "time" + "go-agent-harness/internal/checkpoints" "go-agent-harness/internal/harness" + htools "go-agent-harness/internal/harness/tools" "go-agent-harness/internal/provider/catalog" "go-agent-harness/internal/store" ) @@ -23,6 +25,31 @@ type staticProvider struct { result harness.CompletionResult } +type alreadyResolvedInputBroker struct { + ready chan struct{} +} + +func (b *alreadyResolvedInputBroker) Ask(ctx context.Context, req htools.AskUserQuestionRequest) (map[string]string, time.Time, error) { + pending := htools.AskUserQuestionPending{ + RunID: req.RunID, CallID: req.CallID, Tool: htools.AskUserQuestionToolName, + Questions: req.Questions, DeadlineAt: time.Now().Add(req.Timeout), + } + if req.OnPending != nil { + req.OnPending(ctx, pending) + } + close(b.ready) + <-ctx.Done() + return nil, time.Time{}, ctx.Err() +} + +func (b *alreadyResolvedInputBroker) Pending(string) (htools.AskUserQuestionPending, bool) { + return htools.AskUserQuestionPending{}, false +} + +func (*alreadyResolvedInputBroker) Submit(string, map[string]string) error { + return checkpoints.ErrAlreadyResolved +} + func (s *staticProvider) Complete(_ context.Context, _ harness.CompletionRequest) (harness.CompletionResult, error) { return s.result, nil } @@ -50,6 +77,39 @@ type waitingProvider struct { done chan struct{} } +type transientWaitingEventStore struct { + *store.MemoryStore + mu sync.Mutex + attempts int + failed chan struct{} + retried chan struct{} +} + +func newTransientWaitingEventStore() *transientWaitingEventStore { + return &transientWaitingEventStore{ + MemoryStore: store.NewMemoryStore(), + failed: make(chan struct{}), + retried: make(chan struct{}), + } +} + +func (s *transientWaitingEventStore) AppendEvent(ctx context.Context, event *store.Event) error { + if event.EventType == string(harness.EventRunWaitingForUser) { + s.mu.Lock() + s.attempts++ + attempt := s.attempts + s.mu.Unlock() + switch attempt { + case 1: + close(s.failed) + return errors.New("transient waiting event append failure") + case 2: + close(s.retried) + } + } + return s.MemoryStore.AppendEvent(ctx, event) +} + func (w *waitingProvider) Complete(ctx context.Context, _ harness.CompletionRequest) (harness.CompletionResult, error) { select { case <-ctx.Done(): @@ -393,6 +453,65 @@ func TestRunInputEndpoints(t *testing.T) { } } +func TestRunInputAlreadyResolvedCheckpointReturnsStableConflict(t *testing.T) { + broker := &alreadyResolvedInputBroker{ready: make(chan struct{})} + provider := &scriptedProvider{turns: []harness.CompletionResult{{ + ToolCalls: []harness.ToolCall{{ + ID: "call_input_resolved", Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Where next?","header":"Route","options":[{"label":"Docs","description":"Read docs"},{"label":"Code","description":"Read code"}],"multiSelect":false}]}`, + }}, + }}} + registry := harness.NewDefaultRegistryWithOptions(t.TempDir(), harness.DefaultRegistryOptions{ + ApprovalMode: harness.ToolApprovalModeFullAuto, AskUserBroker: broker, AskUserTimeout: time.Second, + }) + runner := harness.NewRunner(provider, registry, harness.RunnerConfig{ + DefaultModel: "gpt-5-nano", MaxSteps: 2, AskUserBroker: broker, AskUserTimeout: time.Second, + }) + ts := httptest.NewServer(New(runner)) + defer ts.Close() + + createRes, err := http.Post(ts.URL+"/v1/runs", "application/json", bytes.NewBufferString(`{"prompt":"Need input"}`)) + if err != nil { + t.Fatalf("create run: %v", err) + } + defer createRes.Body.Close() + var created struct { + RunID string `json:"run_id"` + } + if err := json.NewDecoder(createRes.Body).Decode(&created); err != nil { + t.Fatalf("decode run: %v", err) + } + select { + case <-broker.ready: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for pending input") + } + + resp, err := http.Post( + ts.URL+"/v1/runs/"+created.RunID+"/input", + "application/json", + bytes.NewBufferString(`{"answers":{"Where next?":"Docs"}}`), + ) + if err != nil { + t.Fatalf("post input: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusConflict { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusConflict) + } + var body struct { + Error struct { + Code string `json:"code"` + } `json:"error"` + } + if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { + t.Fatalf("decode conflict: %v", err) + } + if body.Error.Code != "no_pending_input" { + t.Fatalf("code = %q, want no_pending_input", body.Error.Code) + } +} + func TestRunInputEndpointsMissingRunAndInvalidJSON(t *testing.T) { t.Parallel() @@ -955,6 +1074,121 @@ func TestLastEventIDSkipsSeenEvents(t *testing.T) { } } +func TestLastEventIDAfterTransientEventRetryReturnsOnlyUnseenEvents(t *testing.T) { + t.Parallel() + + provider := &scriptedProvider{turns: []harness.CompletionResult{ + { + ToolCalls: []harness.ToolCall{{ + ID: "call_retry_wait_event", + Name: htools.AskUserQuestionToolName, + Arguments: `{"questions":[{"question":"Continue?","header":"Continue","options":[{"label":"Yes","description":"Continue"},{"label":"No","description":"Stop"}],"multiSelect":false}]}`, + }}, + }, + {Content: "done after retry"}, + }} + broker := harness.NewInMemoryAskUserQuestionBroker(time.Now) + persistence := newTransientWaitingEventStore() + const timeout = 2 * time.Second + runner := harness.NewRunner(provider, harness.NewDefaultRegistryWithOptions(t.TempDir(), harness.DefaultRegistryOptions{ + ApprovalMode: harness.ToolApprovalModeFullAuto, + AskUserBroker: broker, + AskUserTimeout: timeout, + }), harness.RunnerConfig{ + DefaultModel: "gpt-4.1-mini", + MaxSteps: 4, + AskUserBroker: broker, + AskUserTimeout: timeout, + Store: persistence, + }) + + ts := httptest.NewServer(New(runner)) + defer ts.Close() + res, err := http.Post(ts.URL+"/v1/runs", "application/json", bytes.NewBufferString(`{"prompt":"retry pending event"}`)) + if err != nil { + t.Fatalf("create run: %v", err) + } + defer res.Body.Close() + var created struct { + RunID string `json:"run_id"` + } + if err := json.NewDecoder(res.Body).Decode(&created); err != nil { + t.Fatalf("decode create response: %v", err) + } + select { + case <-persistence.retried: + case <-time.After(timeout): + t.Fatal("waiting event append was not retried") + } + if err := runner.SubmitInput(created.RunID, map[string]string{"Continue?": "Yes"}); err != nil { + t.Fatalf("SubmitInput: %v", err) + } + + fullRes, err := http.Get(ts.URL + "/v1/runs/" + created.RunID + "/events") + if err != nil { + t.Fatalf("full events: %v", err) + } + fullBody, err := io.ReadAll(fullRes.Body) + fullRes.Body.Close() + if err != nil { + t.Fatalf("read full events: %v", err) + } + fullIDs, waitingIndex := sseIDsAndWaitingIndex(string(fullBody)) + if waitingIndex < 0 { + t.Fatalf("full replay has no waiting event:\n%s", fullBody) + } + for i, id := range fullIDs { + want := fmt.Sprintf("%s:%d", created.RunID, i) + if id != want { + t.Fatalf("event IDs are not contiguous at index %d: got %q, want %q; all=%v", i, id, want, fullIDs) + } + } + + reconnectReq, err := http.NewRequest(http.MethodGet, ts.URL+"/v1/runs/"+created.RunID+"/events", nil) + if err != nil { + t.Fatalf("new reconnect request: %v", err) + } + reconnectReq.Header.Set("Last-Event-ID", fullIDs[waitingIndex]) + reconnectRes, err := http.DefaultClient.Do(reconnectReq) + if err != nil { + t.Fatalf("reconnect: %v", err) + } + reconnectBody, err := io.ReadAll(reconnectRes.Body) + reconnectRes.Body.Close() + if err != nil { + t.Fatalf("read reconnect events: %v", err) + } + reconnectIDs, _ := sseIDsAndWaitingIndex(string(reconnectBody)) + wantReconnect := fullIDs[waitingIndex+1:] + if fmt.Sprint(reconnectIDs) != fmt.Sprint(wantReconnect) { + t.Fatalf("reconnect IDs = %v, want only unseen IDs %v; body:\n%s", reconnectIDs, wantReconnect, reconnectBody) + } +} + +func sseIDsAndWaitingIndex(body string) ([]string, int) { + ids := make([]string, 0) + waitingIndex := -1 + for _, frame := range strings.Split(body, "\n\n") { + var id, eventType string + for _, line := range strings.Split(frame, "\n") { + switch { + case strings.HasPrefix(line, "id: "): + id = strings.TrimPrefix(line, "id: ") + case strings.HasPrefix(line, "event: "): + eventType = strings.TrimPrefix(line, "event: ") + } + } + if id == "" { + continue + } + ids = append(ids, id) + if eventType == string(harness.EventRunWaitingForUser) { + waitingIndex = len(ids) - 1 + } + } + return ids, waitingIndex +} + // TestLastEventID_AdversarialValuesDoNotPanic is an ATTACK test (C1): a // crafted Last-Event-ID header must never crash the run-events handler. // handleRunEvents parses the sequence number out of Last-Event-ID and slices