From 5d3f29e062868be78084748a947a2ca6ef822379 Mon Sep 17 00:00:00 2001 From: 010351 Date: Wed, 26 Aug 2026 03:47:50 +0800 Subject: [PATCH 1/2] fix(agent): replace hard agent deadline with pi-session-aware silence watchdog The daemon killed healthy Pi agents as "silent 30m" because it used a fixed per-context deadline while pi in --mode json buffers stdout when piped, so a healthy turn showed zero stdout bytes. Forensics on run 01M0WRW741G2YPQFXYE5TG6DZD proved the fixer's session JSONL advanced until 23s before SIGTERM. And one review-step budget spanned the fix turn and the rereview turn, so run 01M0WF7T6DEKM614JT3DM4ZT2F's new reviewer inherited ~20m of spent budget and died ~10 minutes into its own turn. Fold all liveness evidence into one monotonic last-activity clock per invocation (internal/pipeline/liveness.go): stdout bytes from the shared native-command pipe, native process lifecycle (a start re-arms the clock, giving every new process a fresh full budget), and bound pi-session JSONL advancement. The watchdog cancels with the whole process tree only after the full configured budget passes with no activity from any source, so the real timeout is preserved unchanged for a genuinely frozen agent while a healthy long turn is never cut off. The pi session binding (internal/agent/pi_liveness.go) is narrow and fail-closed: resume binds only the one existing _.jsonl in the launched cwd's session dir (pi's encoding, resolved physical like pi's process.cwd()); a fresh durable session binds only when exactly one new file appears after launch and unbinds if the late stdout header names a different id; --no-session turns, relocated session storage, ambiguity, and unresolvable dirs all keep the conservative stdout/lifecycle-only behavior. Review and Test now pass their configured budgets per agent turn through the seam (review_agent_timeout, test_agent_timeout) instead of pre-installing one hard context, so every turn gets its own fresh silence budget. Timeout diagnostics name per-kind last-activity ages ("pi session events 21.995s ago, process lifecycle 24s ago") with no content or paths, and bind/unbind transitions land in the step log as liveness notes. Managed-server adapters (opencode, rovodev) report no activity and behave exactly as the legacy fixed deadline. Docs: global-config.md owns the changed agent_timeout / review_agent_timeout / test_agent_timeout semantics (silence budget, not wall-clock cap; review is per-turn, no longer per-round). --- AGENTS.md | 11 +- .../content/docs/reference/global-config.md | 20 +- internal/agent/acpx.go | 2 +- internal/agent/activity.go | 43 ++ internal/agent/agent.go | 7 + internal/agent/antigravity.go | 2 +- internal/agent/claude.go | 2 +- internal/agent/codex.go | 2 +- internal/agent/copilot.go | 2 +- internal/agent/grok.go | 2 +- internal/agent/native_command.go | 23 +- internal/agent/pi.go | 42 +- internal/agent/pi_liveness.go | 347 ++++++++++++++++ internal/agent/pi_liveness_test.go | 384 ++++++++++++++++++ internal/agent/pi_test.go | 4 +- internal/agent/reap_unix_test.go | 2 +- internal/config/config.go | 24 +- internal/pipeline/agent_run.go | 105 +++-- internal/pipeline/agent_run_pi_test.go | 207 ++++++++++ internal/pipeline/agent_run_test.go | 9 +- internal/pipeline/liveness.go | 168 ++++++++ internal/pipeline/liveness_test.go | 161 ++++++++ internal/pipeline/pipeline.go | 2 +- internal/pipeline/steps/common_fix.go | 15 +- internal/pipeline/steps/review.go | 55 +-- internal/pipeline/steps/review_test.go | 88 ++-- internal/pipeline/steps/test.go | 42 +- internal/pipeline/steps/test_test.go | 13 +- 28 files changed, 1605 insertions(+), 179 deletions(-) create mode 100644 internal/agent/activity.go create mode 100644 internal/agent/pi_liveness.go create mode 100644 internal/agent/pi_liveness_test.go create mode 100644 internal/pipeline/agent_run_pi_test.go create mode 100644 internal/pipeline/liveness.go create mode 100644 internal/pipeline/liveness_test.go diff --git a/AGENTS.md b/AGENTS.md index b4b242482..3b761320c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -243,9 +243,18 @@ Safest local verification sequence after non-trivial changes: This repository dogfoods an empty `commands.test` so the agent-driven targeted path is the default; do not reintroduce `go test -race ./...` as a local Test override. Process-group reaping on clean/error exit (#357) and Unix WaitDelay remain the lifecycle safety net when agents spawn test workers - restoring the agent-driven path must not revive the daemon OOM leak. Those agent turns are bounded by `test_agent_timeout` (default 30m, global-only): a stalled evidence or repair agent is cancelled and the run fails instead of waiting forever. Native adapters already honor that deadline through `CommandContext`; the missing piece was the Test step never setting one. Docs owner is `docs/src/content/docs/reference/global-config.md`. - Every other pipeline agent invocation is bounded by `agent_timeout` (default 30m, global-only) at `pipeline.RunAgent` / the executor `timeoutAgent` seam, so a new agent-spawning step cannot hang a run by forgetting a deadline. Review keeps `review_agent_timeout` as a per-round budget; an existing sooner deadline is honored rather than capped. The invocation context is scoped only to `Agent.Run`; a late successful return after the deadline is rejected. Docs owner is `docs/src/content/docs/reference/global-config.md`. + Every other pipeline agent invocation is bounded by `agent_timeout` (default 30m, global-only) at `pipeline.RunAgent` / the executor `timeoutAgent` seam, so a new agent-spawning step cannot hang a run by forgetting a deadline. Review keeps `review_agent_timeout` as a per-turn budget (the fix turn and the rereview turn each get a fresh full budget); an existing sooner deadline is honored rather than capped. The invocation context is scoped only to `Agent.Run`; a late successful return after the deadline is rejected. Docs owner is `docs/src/content/docs/reference/global-config.md`. Regressions: `TestTestStep_InitialAgent_TargetedValidationContract`, `TestTestStep_FixMode_TargetedVerificationContract`, `TestTestStep_FixMode_DriverFullSuiteInstructionDoesNotOverrideContract`, `TestTestStep_InitialAgent_NoTargetedEvidenceRequiresHonestFinding`, `TestTestStep_HangingEvidenceAgentFailsRunAfterTimeout`, `TestCodexAgent_RunCancelsSilentHang`, `TestDogfoodConfig_NoBroadLocalTestCommand`, `TestCIWorkflow_RetainsFullRaceSuiteAsBroadRegressionOwner`, plus the existing #357 reap/WaitDelay tests, `TestRunAgent_*`, `TestExecutor_DirectAgentRunIsDeadlineBounded`, `TestDocumentStep_HangingAgentFailsRunAfterTimeout`, `TestLintStep_HangingAgentFailsRunAfterTimeout`, `TestCIStep_HangingFixAgentFailsAfterTimeout`, `TestRebaseStep_HangingConflictAgentFailsAfterTimeout`. +**Agent Liveness, Not stdout Bytes (`internal/pipeline/liveness.go`, `internal/agent/pi_liveness.go`)** + +- Agent timeouts are SILENCE budgets, not hard wall-clock caps: one monotonic last-activity clock per invocation, cancelled with the whole process tree only after the full configured budget (`agent_timeout` / `review_agent_timeout` / `test_agent_timeout`) passes with no activity. The fix turned the old fixed deadline into this watchdog because pi in `--mode json` buffers stdout when piped, so a healthy fixer was killed as "silent 30m" 23s after its last real session event, and a review rereview inherited the fix turn's spent budget (killed ~10 minutes into its own turn). +- Activity folds three kinds into the one clock (`agent.ActivityKind`): stdout bytes (all native adapters, wired at `startNativeAgentCommand`), native process lifecycle (a start re-arms the clock, which is what makes every new process's budget fresh), and bound pi-session JSONL advancement. Managed-server adapters (opencode, rovodev) report nothing and behave exactly as the legacy fixed deadline. +- The pi session binding is narrow and fail-closed (`startPiSessionWatcher`): `--no-session` turns have no file and stay stdout/lifecycle-only; a resume binds only to the one existing `_.jsonl` in the launched cwd's session dir (pi's `-- -->--` encoding under `PI_CODING_AGENT_DIR` or `~/.pi/agent`, cwd resolved physical like pi's `process.cwd()`); a fresh durable session binds only when exactly ONE new file appears after launch, and unbinds if the late stdout session header names a different id. Relocated storage (`--session-dir` / `PI_CODING_AGENT_SESSION_DIR`), multiple matches, or an unresolvable dir all mean NO binding - never credit unrelated activity. +- The timeout diagnostic (`AgentTimeoutError`, re-labeled per step) names per-kind last-activity ages ("last activity: pi session events 30m0s ago" / "no stdout, lifecycle, or session-event activity observed") with no content or paths; bind/unbind transitions also land in the step log via lifecycle `liveness` notes. +- Nested seams share one owner: `RunAgent` installs the watchdog and stores it in the ctx; the executor `timeoutAgent` backstop reuses it via `livenessFromContext` instead of stacking a second clock. A parent with an existing hard deadline still wins unchanged (legacy callers). +- Regressions: `internal/pipeline/liveness_test.go`, `internal/agent/pi_liveness_test.go`, `TestRunAgent_PiSessionActivityKeepsBufferedStdoutInvocationAlive`, `TestRunAgent_FrozenPiSessionIsKilledAfterTheFullBudget`, `TestRunAgent_SessionFreePiQuietStdoutIsKilledAtBudget` (`internal/pipeline/agent_run_pi_test.go`), `TestReviewStep_EachAgentTurnGetsItsOwnFreshBudget`. + **Intent Provenance & Conformance (`internal/pipeline/steps/intent_prompt.go`)** - Intent carries provenance: an explicit `axi run --intent` persists `Source==db.RunIntentSourceAgent` ("agent", score 1); a transcript match persists the agent name ("claude"/"codex"/...). The executor propagates it as `StepContext.IntentSource` alongside `UserIntent` (`executor.go`). diff --git a/docs/src/content/docs/reference/global-config.md b/docs/src/content/docs/reference/global-config.md index 0e687c9b3..adea74f71 100644 --- a/docs/src/content/docs/reference/global-config.md +++ b/docs/src/content/docs/reference/global-config.md @@ -391,10 +391,12 @@ For older active runs that do not yet have activity rows, AXI falls back to the ### agent_timeout -Maximum wall-clock time for one pipeline agent invocation that does not already have a more specific deadline. +Maximum silence time for one pipeline agent invocation that does not already have a more specific deadline. This is the default-by-construction budget: Document, Lint, Rebase conflict repair, PR drafting, CI auto-fix, and any future agent-spawning step are bounded even if they forget to install their own timer. -Review still uses [`review_agent_timeout`](#review_agent_timeout) as a per-round budget, Test still uses [`test_agent_timeout`](#test_agent_timeout) per invocation, and Intent keeps its five-minute extraction cap; any existing deadline is honored rather than capped. -When this deadline expires, the agent is cancelled and the invocation returns a timeout diagnostic instead of remaining active indefinitely. Agent-driven mutation steps fail the run, while PR drafting follows its existing agent-error fallback and continues with deterministic content. +The budget is a liveness budget, not a hard wall-clock cap: the invocation is cancelled only after this long with no reported activity. Activity means bytes on the agent's stdout pipe, native process lifecycle events, and - for the Pi adapter - advancement of the exact pi session file bound to that invocation (pi buffers stdout when piped, so a quiet pipe alone does not prove a quiet agent). Each newly launched agent process starts with a fresh full budget. +When no pi session can be bound unambiguously (session-free `--no-session` turns, relocated session storage, or ambiguous files), only stdout and lifecycle evidence govern the invocation, matching the previous behavior. +Review uses [`review_agent_timeout`](#review_agent_timeout) per agent turn, Test uses [`test_agent_timeout`](#test_agent_timeout) per invocation, and Intent keeps its five-minute extraction cap; any existing deadline is honored rather than capped. +When the silence budget expires with no activity from any source, the agent's whole process tree is terminated and the invocation returns a timeout diagnostic naming the last observed activity instead of remaining active indefinitely. Agent-driven mutation steps fail the run, while PR drafting follows its existing agent-error fallback and continues with deterministic content. A late successful return after the deadline is rejected, so post-agent commits and PR content cannot use work from a timed-out turn. | | | @@ -409,9 +411,9 @@ It is global-only: repository config and environment variables cannot override i ### review_agent_timeout -Maximum wall-clock time for the Review step's agent turns in one review round. -The budget starts at that round's first agent turn and covers its optional review-fix turn plus the rereview turn together; every later auto-fix round starts a fresh budget. -When the deadline expires, the review agent is cancelled and the run fails with a diagnostic naming the timeout instead of remaining active indefinitely. +Maximum silence time for one Review-step agent turn. +The optional review-fix turn and the rereview turn each get their own fresh full budget measured from that turn's own activity: a long fix turn cannot leave the rereview only the remainder. Liveness evidence is the same as [`agent_timeout`](#agent_timeout) (stdout bytes, process lifecycle, bound pi session advancement). +When the silence budget expires with no activity, the review agent's process tree is terminated and the run fails with a diagnostic naming the timeout and the last observed activity instead of remaining active indefinitely. | | | | ------- | ---------------------- | @@ -424,9 +426,9 @@ Raise it for repositories whose reviews legitimately run long; it bounds only th ### test_agent_timeout -Maximum wall-clock time for one Test-step agent invocation. -The budget covers the post-test evidence-gathering turn, and a Test-repair turn gets its own budget of the same length. -When the deadline expires, the test agent is cancelled and the run fails with a diagnostic naming the timeout instead of remaining active indefinitely. +Maximum silence time for one Test-step agent invocation. +The post-test evidence-gathering turn and a Test-repair turn each get their own budget of this size. Liveness evidence is the same as [`agent_timeout`](#agent_timeout). +When the silence budget expires with no activity, the test agent's process tree is terminated and the run fails with a diagnostic naming the timeout and the last observed activity instead of remaining active indefinitely. | | | | ------- | ---------------------- | diff --git a/internal/agent/acpx.go b/internal/agent/acpx.go index e5c820fe0..2d79f9ffc 100644 --- a/internal/agent/acpx.go +++ b/internal/agent/acpx.go @@ -53,7 +53,7 @@ func (a *acpxAgent) runOnce(ctx context.Context, opts RunOpts) (*Result, error) if err != nil { return nil, fmt.Errorf("acpx stdin pipe: %w", err) } - started, err := startNativeAgentCommand(cmd) + started, err := startNativeAgentCommand(cmd, opts.OnActivity) if err != nil { _ = stdin.Close() return nil, fmt.Errorf("acpx start: %w", err) diff --git a/internal/agent/activity.go b/internal/agent/activity.go new file mode 100644 index 000000000..1e8e07dc7 --- /dev/null +++ b/internal/agent/activity.go @@ -0,0 +1,43 @@ +package agent + +// ActivityKind identifies one source of liveness evidence for a running agent +// invocation. The pipeline's invocation watchdog folds every reported kind +// into a single monotonic last-activity clock: an invocation is killed only +// after its full configured silence budget elapses with no activity from any +// source. Kinds are deliberately coarse - they carry no content, no paths, +// and no session payloads - so timeout diagnostics can name the evidence +// class without leaking prompt or session data. +type ActivityKind string + +const ( + // ActivityStdout reports bytes visible on the agent's stdout pipe. This + // is the wrapper's traditional evidence source; some harnesses (pi in + // JSON mode) buffer stdout when piped, so a quiet pipe does not prove a + // quiet agent. + ActivityStdout ActivityKind = "stdout" + // ActivityLifecycle reports native process lifecycle transitions + // (process start, process exit). A start also resets the silence clock, + // which is what gives every freshly launched process its own full + // budget. Process existence alone is never activity. + ActivityLifecycle ActivityKind = "lifecycle" + // ActivitySession reports advancement of the exact adapter-native + // session file bound to this invocation (pi's session JSONL). It is + // credited only when the binding is unambiguous; otherwise the + // invocation falls back to stdout/lifecycle evidence only. + ActivitySession ActivityKind = "session" +) + +// ActivityKindLabel renders a kind for operator-facing diagnostics. The +// labels name evidence classes, never content. +func ActivityKindLabel(kind ActivityKind) string { + switch kind { + case ActivityStdout: + return "stdout bytes" + case ActivityLifecycle: + return "process lifecycle" + case ActivitySession: + return "pi session events" + default: + return string(kind) + } +} diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 8573c6132..ed6c41624 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -32,6 +32,13 @@ type RunOpts struct { JSONSchema json.RawMessage // structured output schema (optional) OnChunk func(text string) // streaming text callback (optional) OnLifecycle func(LifecycleEvent) // native agent lifecycle callback (optional) + // OnActivity reports coarse liveness evidence (bytes on the stdout pipe, + // native process lifecycle, bound adapter-native session advancement) to + // the pipeline's per-invocation silence watchdog. It carries kinds only, + // never content. Adapters that report nothing behave exactly as before: + // the watchdog then measures silence from invocation start, matching the + // legacy fixed-deadline behavior. + OnActivity func(ActivityKind) // Session, when non-nil, asks a session-capable adapter (see // SessionResumer) to start or resume a durable native session. Adapters // without session support ignore it and run cold; the caller detects the diff --git a/internal/agent/antigravity.go b/internal/agent/antigravity.go index 0dbee2d6e..25ef9aa8f 100644 --- a/internal/agent/antigravity.go +++ b/internal/agent/antigravity.go @@ -98,7 +98,7 @@ func (a *antigravityAgent) runOnce(ctx context.Context, opts RunOpts) (*Result, cmd.Env = a.gitSafeEnv(opts.CWD) shellenv.ConfigureShellCommand(cmd) - started, err := startNativeAgentCommand(cmd) + started, err := startNativeAgentCommand(cmd, opts.OnActivity) if err != nil { return nil, fmt.Errorf("antigravity start: %w", err) } diff --git a/internal/agent/claude.go b/internal/agent/claude.go index 82598173f..c89c3d38a 100644 --- a/internal/agent/claude.go +++ b/internal/agent/claude.go @@ -83,7 +83,7 @@ func (a *claudeAgent) runOnce(ctx context.Context, opts RunOpts) (*Result, error var stderrBuf []byte var stderrWG sync.WaitGroup - started, err := startNativeAgentCommand(cmd) + started, err := startNativeAgentCommand(cmd, opts.OnActivity) if err != nil { return nil, fmt.Errorf("claude start: %w", err) } diff --git a/internal/agent/codex.go b/internal/agent/codex.go index 9c83b3d1b..31f2adc3b 100644 --- a/internal/agent/codex.go +++ b/internal/agent/codex.go @@ -99,7 +99,7 @@ func (a *codexAgent) runOnce(ctx context.Context, opts RunOpts) (*Result, error) var stderrBuf []byte var stderrWG sync.WaitGroup - started, err := startNativeAgentCommand(cmd) + started, err := startNativeAgentCommand(cmd, opts.OnActivity) if err != nil { return nil, fmt.Errorf("codex start: %w", err) } diff --git a/internal/agent/copilot.go b/internal/agent/copilot.go index 86271511c..4f53e05d4 100644 --- a/internal/agent/copilot.go +++ b/internal/agent/copilot.go @@ -46,7 +46,7 @@ func (a *copilotAgent) runOnce(ctx context.Context, opts RunOpts) (*Result, erro var stderrBuf []byte var stderrWG sync.WaitGroup - started, err := startNativeAgentCommand(cmd) + started, err := startNativeAgentCommand(cmd, opts.OnActivity) if err != nil { return nil, fmt.Errorf("copilot start: %w", err) } diff --git a/internal/agent/grok.go b/internal/agent/grok.go index f712edafa..8eea168fd 100644 --- a/internal/agent/grok.go +++ b/internal/agent/grok.go @@ -103,7 +103,7 @@ func (a *grokAgent) runOnce(ctx context.Context, opts RunOpts) (*Result, error) var stderrBuf []byte var stderrWG sync.WaitGroup - started, err := startNativeAgentCommand(cmd) + started, err := startNativeAgentCommand(cmd, opts.OnActivity) if err != nil { return nil, fmt.Errorf("grok start: %w", err) } diff --git a/internal/agent/native_command.go b/internal/agent/native_command.go index 2faf035f9..df579a648 100644 --- a/internal/agent/native_command.go +++ b/internal/agent/native_command.go @@ -38,10 +38,17 @@ type nativeAgentPipe struct { file *os.File done func() doneOnce sync.Once + // onData, when set, is called after every successful read. It feeds the + // invocation silence watchdog: bytes visible on the pipe are liveness + // evidence. Kinds only - the bytes themselves never leave the parse path. + onData func() } func (p *nativeAgentPipe) Read(b []byte) (int, error) { n, err := p.file.Read(b) + if n > 0 && p.onData != nil { + p.onData() + } if err != nil { p.markDone() } @@ -58,7 +65,12 @@ func (p *nativeAgentPipe) markDone() { p.doneOnce.Do(p.done) } -func startNativeAgentCommand(cmd *exec.Cmd) (*nativeAgentCommand, error) { +// startNativeAgentCommand launches cmd with piped stdout/stderr and reports +// coarse liveness evidence through onActivity: one lifecycle signal at process +// start (which resets the silence clock for the fresh process) and one stdout +// signal per successful pipe read. onActivity may be nil, in which case the +// caller's watchdog keeps its legacy invocation-start-only behavior. +func startNativeAgentCommand(cmd *exec.Cmd, onActivity func(ActivityKind)) (*nativeAgentCommand, error) { stdoutR, stdoutW, err := os.Pipe() if err != nil { return nil, fmt.Errorf("stdout pipe: %w", err) @@ -88,8 +100,15 @@ func startNativeAgentCommand(cmd *exec.Cmd) (*nativeAgentCommand, error) { remainingPipes: 2, pipesDone: make(chan struct{}), } - started.stdout = &nativeAgentPipe{file: stdoutR, done: started.markPipeDone} + var onStdoutData func() + if onActivity != nil { + onStdoutData = func() { onActivity(ActivityStdout) } + } + started.stdout = &nativeAgentPipe{file: stdoutR, done: started.markPipeDone, onData: onStdoutData} started.stderr = &nativeAgentPipe{file: stderrR, done: started.markPipeDone} + if onActivity != nil { + onActivity(ActivityLifecycle) + } go func() { err := cmd.Wait() started.terminate() diff --git a/internal/agent/pi.go b/internal/agent/pi.go index 479ba7f20..1edfaaefa 100644 --- a/internal/agent/pi.go +++ b/internal/agent/pi.go @@ -10,6 +10,7 @@ import ( "os/exec" "strings" "sync" + "sync/atomic" "github.com/kunchenguid/no-mistakes/internal/shellenv" ) @@ -72,8 +73,26 @@ func (a *piAgent) runOnce(ctx context.Context, opts RunOpts) (*Result, error) { return nil, fmt.Errorf("pi stdin pipe: %w", err) } - started, err := startNativeAgentCommand(cmd) + pp := &piParser{onChunk: opts.OnChunk} + // Bind this exact process to its pi session JSONL so buffered stdout does + // not read as silence. The watcher credits only unambiguous session + // advancement; when no binding exists it is nil and stdout/lifecycle + // evidence alone governs. It snapshots pre-existing session files here, + // strictly before the process can create its own, so the fresh-session + // binding cannot race pi's startup. It stops when this invocation's + // context ends or runOnce returns, whichever comes first. + watch := startPiSessionWatcher(ctx, a.extraArgs, cmd.Env, opts.CWD, opts.Session, pp.sessionIDString, opts.OnActivity, func(msg string) { + emitLifecycle(opts, LifecycleEvent{Agent: "pi", Phase: "liveness", Message: msg}) + }) + if watch != nil { + defer watch.shutdown() + } + + started, err := startNativeAgentCommand(cmd, opts.OnActivity) if err != nil { + if watch != nil { + watch.shutdown() + } _ = stdin.Close() return nil, fmt.Errorf("pi start: %w", err) } @@ -92,7 +111,6 @@ func (a *piAgent) runOnce(ctx context.Context, opts RunOpts) (*Result, error) { stderrBuf, _ = io.ReadAll(started.stderr) }() - pp := &piParser{onChunk: opts.OnChunk} if err := pp.parse(ctx, started.stdout); err != nil { err = started.waitAfterParseError(err) stderrWG.Wait() @@ -137,14 +155,14 @@ func (a *piAgent) runOnce(ctx context.Context, opts RunOpts) (*Result, error) { res.ModelProvider = pp.provider } if err == nil && opts.Session != nil { - if pp.sessionID == "" { + if pp.sessionIDString() == "" { // A durable invocation without Pi's required JSON-mode session header // cannot be resumed safely on a later fixer turn. err = fmt.Errorf("pi did not report a session identity") - } else if opts.Session.ID != "" && pp.sessionID != opts.Session.ID { + } else if opts.Session.ID != "" && pp.sessionIDString() != opts.Session.ID { err = fmt.Errorf("pi did not confirm the requested session") } else { - res.SessionID = pp.sessionID + res.SessionID = pp.sessionIDString() res.Resumed = opts.Session.ID != "" // Pi's agent_end event contains only the messages generated by this // invocation, including after --session, so usage is not cumulative. @@ -275,7 +293,10 @@ type piParser struct { streamText map[int]string completeText map[int]string finalAssistant map[string]any - sessionID string + // sessionID is written by the parse goroutine and read by the liveness + // watcher goroutine (late header confirmation), so it is atomic (stores + // string; unset means no header parsed yet). + sessionID atomic.Value model string provider string usage TokenUsage @@ -283,6 +304,11 @@ type piParser struct { assistantError string } +func (p *piParser) sessionIDString() string { + id, _ := p.sessionID.Load().(string) + return id +} + func (p *piParser) parse(ctx context.Context, r io.Reader) error { scanner := bufio.NewScanner(r) scanner.Buffer(make([]byte, 0, 64*1024), 256*1024*1024) @@ -319,9 +345,9 @@ func (p *piParser) handleEvent(event map[string]any) { case "session": // The JSON-mode header is emitted before agent_start. Preserve only its // first valid full UUID so later arbitrary events cannot replace it. - if p.sessionID == "" { + if p.sessionIDString() == "" { if id, _ := event["id"].(string); isPiSessionID(id) { - p.sessionID = id + p.sessionID.Store(id) } } case "message_update": diff --git a/internal/agent/pi_liveness.go b/internal/agent/pi_liveness.go new file mode 100644 index 000000000..328470045 --- /dev/null +++ b/internal/agent/pi_liveness.go @@ -0,0 +1,347 @@ +package agent + +import ( + "context" + "os" + "path/filepath" + "strings" + "sync" + "time" +) + +// pi_liveness.go binds one launched pi process to its exact adapter-native +// session JSONL so the pipeline's silence watchdog can credit real session +// advancement while pi's own stdout is buffered (pi in --mode json emits on a +// pipe that the wrapper may not drain for long stretches; the session file is +// the authoritative per-event record, as incident forensics proved when a +// healthy fixer was killed as "silent 30m" 23s after its last session event). +// +// Binding is deliberately narrow and fail-closed: +// - --no-session invocations persist nothing, so there is no file to watch; +// stdout/lifecycle evidence alone governs them. +// - A resumed session (--session ) binds only to the one existing +// file in the launched cwd's session dir whose name ends _.jsonl. +// - A fresh durable session binds only when exactly one new session file +// appears in that dir after process start, and it unbinds if pi's +// JSON-mode session header later names a different session id. +// - Anything ambiguous (multiple matches, --session-dir or +// PI_CODING_AGENT_SESSION_DIR relocating storage, an unresolvable agent +// dir) disables the watcher: the invocation keeps the conservative +// stdout/lifecycle behavior instead of crediting unrelated activity. +// +// The watcher polls exactly one directory listing or one file stat per tick, +// never reads session content, and never credits broad filesystem changes. + +// piSessionWatchPollInterval is the watcher's tick. It is a package-level var +// so tests can shorten it, mirroring transientBackoff. +var piSessionWatchPollInterval = 2 * time.Second + +// piSessionPathReplacer mirrors pi's session-dir encoding (verified against +// pi's dist/core/session-manager.js getDefaultSessionDirPath): strip one +// leading path separator, then replace every '/', '\', and ':' with '-'. +var piSessionPathReplacer = strings.NewReplacer("/", "-", "\\", "-", ":", "-") + +// piSessionDirForCWD computes the directory pi persists this cwd's sessions +// to: /sessions/----. Pi encodes process.cwd(), which +// the kernel resolves to the physical path, so symlinked spellings (macOS +// /var -> /private/var, symlinked homes) must be resolved here too; when +// resolution fails the launch spelling is used and binding simply stays +// unavailable (fail-closed to stdout/lifecycle liveness). +func piSessionDirForCWD(agentDir, cwd string) string { + abs, err := filepath.EvalSymlinks(cwd) + if err != nil { + abs, err = filepath.Abs(cwd) + if err != nil { + abs = cwd + } + } + return filepath.Join(agentDir, "sessions", piEncodeSessionDirName(abs)) +} + +// piEncodeSessionDirName is the pure cwd→directory-name encoding, kept +// separate from path resolution so it stays testable on every platform. +func piEncodeSessionDirName(abs string) string { + enc := abs + if len(enc) > 0 && (enc[0] == '/' || enc[0] == '\\') { + enc = enc[1:] + } + return "--" + piSessionPathReplacer.Replace(enc) + "--" +} + +// piAgentDirFromEnv resolves pi's agent config dir the way pi does +// (dist/config.js getAgentDir): PI_CODING_AGENT_DIR when set, else +// /.pi/agent. Home is read from the child environment the adapter +// actually launches with, falling back to the process home. +func piAgentDirFromEnv(env []string) string { + if dir := envValue(env, "PI_CODING_AGENT_DIR"); dir != "" { + return dir + } + home := envValue(env, "HOME") + if home == "" { + home = envValue(env, "USERPROFILE") + } + if home == "" { + if h, err := os.UserHomeDir(); err == nil { + home = h + } + } + if home == "" { + return "" + } + return filepath.Join(home, ".pi", "agent") +} + +func envValue(env []string, key string) string { + prefix := key + "=" + value := "" + for _, entry := range env { + if strings.HasPrefix(entry, prefix) { + value = entry[len(prefix):] + } + } + return value +} + +// piSessionFileID extracts the session UUID from pi's session file name +// shape _.jsonl, rejecting any other shape. +func piSessionFileID(name string) string { + const suffix = ".jsonl" + if !strings.HasSuffix(name, suffix) { + return "" + } + stem := strings.TrimSuffix(name, suffix) + // Shape: _ - pi's timestamp prefix is never empty. + if len(stem) < 38 || stem[len(stem)-37] != '_' { + return "" + } + id := stem[len(stem)-36:] + if !isPiSessionID(id) { + return "" + } + return id +} + +// piSessionFileState is the advancement fingerprint of one session file. +type piSessionFileState struct { + size int64 + mod time.Time +} + +// piSessionWatcher watches at most one bound session file and reports its +// advancement as ActivitySession. Create it with startPiSessionWatcher; a nil +// watcher means binding was impossible and the invocation keeps conservative +// stdout/lifecycle liveness. +type piSessionWatcher struct { + dir string + resumeID string + sessionID func() string // late-arriving JSON-mode header id, "" until parsed + onActivity func(ActivityKind) + onNote func(string) + baseline map[string]piSessionFileState // session files present at launch, set once before run + + stopCh chan struct{} + doneCh chan struct{} + stop sync.Once +} + +// startPiSessionWatcher resolves the binding policy for this invocation and, +// when a binding is possible, starts the watch goroutine. It returns nil when +// the invocation cannot have a watchable session (session-free, relocated +// session storage, or an unresolvable agent dir). +func startPiSessionWatcher(ctx context.Context, extraArgs, childEnv []string, cwd string, session *SessionRef, sessionID func() string, onActivity func(ActivityKind), onNote func(string)) *piSessionWatcher { + if session == nil || onActivity == nil { + // --no-session persists no file; nothing authoritative to watch. + return nil + } + if envValue(childEnv, "PI_CODING_AGENT_SESSION_DIR") != "" || piArgsHaveSessionDir(extraArgs) { + // Session storage is relocated to a flat operator-chosen directory we + // cannot cwd-scope; binding there could credit an unrelated session. + if onNote != nil { + onNote("pi liveness: session storage relocated by --session-dir/PI_CODING_AGENT_SESSION_DIR; stdout-only liveness") + } + return nil + } + agentDir := piAgentDirFromEnv(childEnv) + if agentDir == "" { + return nil + } + w := &piSessionWatcher{ + dir: piSessionDirForCWD(agentDir, cwd), + sessionID: sessionID, + onActivity: onActivity, + onNote: onNote, + stopCh: make(chan struct{}), + doneCh: make(chan struct{}), + } + if session.ID != "" { + w.resumeID = session.ID + } + // Snapshot pre-existing session files synchronously: a fresh durable + // session may bind only to a file created after this point, and taking + // the baseline inside the goroutine would race the file pi creates at + // process start. The snapshot carries size/mtime so a resumed session + // that already advanced between launch and the first poll is credited. + w.baseline = w.snapshotSessionFiles() + go w.run(ctx) + return w +} + +func piArgsHaveSessionDir(args []string) bool { + for _, arg := range args { + if arg == "--session-dir" || strings.HasPrefix(arg, "--session-dir=") { + return true + } + } + return false +} + +// shutdown stops the watch goroutine and waits for it to exit. +func (w *piSessionWatcher) shutdown() { + w.stop.Do(func() { close(w.stopCh) }) + <-w.doneCh +} + +func (w *piSessionWatcher) note(msg string) { + if w.onNote != nil { + w.onNote(msg) + } +} + +func (w *piSessionWatcher) run(ctx context.Context) { + defer close(w.doneCh) + + bound := false + disabled := false + boundID := "" + boundPath := "" + var lastSize int64 + var lastMod time.Time + + bind := func(name, id string, advanced bool) { + bound = true + boundID = id + boundPath = filepath.Join(w.dir, name) + if info, err := os.Stat(boundPath); err == nil { + lastSize = info.Size() + lastMod = info.ModTime() + } + w.note("pi liveness: watching session " + id) + // Advancement between process launch and the bind poll is real session + // activity (a fresh session's creation, or a resumed session's new + // events): credit it so the silence clock reflects the true last + // activity instead of waiting one extra poll interval. + if advanced { + w.onActivity(ActivitySession) + } + } + disable := func(reason string) { + disabled = true + w.note(reason + "; stdout-only liveness") + } + + ticker := time.NewTicker(piSessionWatchPollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-w.stopCh: + return + case <-ticker.C: + } + if disabled { + return + } + + if !bound { + if w.resumeID != "" { + matches := w.matchResumeFiles() + if len(matches) > 1 { + disable("pi liveness: multiple session files match the resumed session id") + continue + } + if len(matches) == 1 { + launchState, existed := w.baseline[matches[0].name] + advanced := !existed || launchState.size != matches[0].state.size || !launchState.mod.Equal(matches[0].state.mod) + bind(matches[0].name, w.resumeID, advanced) + } + continue + } + current := w.snapshotSessionFiles() + var fresh []string + for name := range current { + if _, ok := w.baseline[name]; !ok { + fresh = append(fresh, name) + } + } + if len(fresh) > 1 { + disable("pi liveness: multiple new session files for this worktree, binding ambiguous") + continue + } + if len(fresh) == 1 { + candidateID := piSessionFileID(fresh[0]) + if headerID := w.sessionID(); headerID != "" && headerID != candidateID { + disable("pi liveness: session header does not match the new session file") + continue + } + // A fresh session file did not exist at launch: its creation is + // this invocation's own session advancement. + bind(fresh[0], candidateID, true) + } + continue + } + + // Bound: a late-arriving stdout header that names a different session + // proves the file we bound is not this invocation's; fail closed. + if headerID := w.sessionID(); headerID != "" && headerID != boundID { + disable("pi liveness: session header does not match the bound session file") + continue + } + info, err := os.Stat(boundPath) + if err != nil { + continue + } + if info.Size() != lastSize || !info.ModTime().Equal(lastMod) { + lastSize = info.Size() + lastMod = info.ModTime() + w.onActivity(ActivitySession) + } + } +} + +// snapshotSessionFiles returns the name→advancement-state map of pi-shaped +// session files in the watch dir. A missing or unreadable dir is empty, +// never an error: absence of evidence keeps the conservative path. +func (w *piSessionWatcher) snapshotSessionFiles() map[string]piSessionFileState { + entries, err := os.ReadDir(w.dir) + if err != nil { + return map[string]piSessionFileState{} + } + out := make(map[string]piSessionFileState, len(entries)) + for _, entry := range entries { + if entry.IsDir() || piSessionFileID(entry.Name()) == "" { + continue + } + info, err := entry.Info() + if err != nil { + continue + } + out[entry.Name()] = piSessionFileState{size: info.Size(), mod: info.ModTime()} + } + return out +} + +type piResumeMatch struct { + name string + state piSessionFileState +} + +func (w *piSessionWatcher) matchResumeFiles() []piResumeMatch { + var matches []piResumeMatch + for name, state := range w.snapshotSessionFiles() { + if piSessionFileID(name) == w.resumeID { + matches = append(matches, piResumeMatch{name: name, state: state}) + } + } + return matches +} diff --git a/internal/agent/pi_liveness_test.go b/internal/agent/pi_liveness_test.go new file mode 100644 index 000000000..b7818af1c --- /dev/null +++ b/internal/agent/pi_liveness_test.go @@ -0,0 +1,384 @@ +package agent + +import ( + "context" + "io" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestPiEncodeSessionDirName_MatchesPiConvention(t *testing.T) { + t.Parallel() + // Verified against pi's dist/core/session-manager.js + // getDefaultSessionDirPath: strip one leading separator, then every + // '/', '\', and ':' becomes '-', wrapped in double dashes. + cases := []struct { + in string + want string + }{ + {"/Users/x/.no-mistakes/worktrees/abc/01RUN", "--Users-x-.no-mistakes-worktrees-abc-01RUN--"}, + {"/", "----"}, + {`C:\work\repo`, "--C--work-repo--"}, + {`D:/mixed\seps/here`, "--D--mixed-seps-here--"}, + } + for _, tc := range cases { + if got := piEncodeSessionDirName(tc.in); got != tc.want { + t.Errorf("piEncodeSessionDirName(%q) = %q, want %q", tc.in, got, tc.want) + } + } +} + +func TestPiSessionFileID(t *testing.T) { + t.Parallel() + const id = "019ff2f3-5f31-744b-90b8-679074ff7686" + if got := piSessionFileID("2026-08-25T16-00-06-143Z_" + id + ".jsonl"); got != id { + t.Errorf("valid name parsed as %q, want %q", got, id) + } + for _, bad := range []string{ + "session.jsonl", + "_" + id + ".jsonl", + "2026-08-25T16-00-06-143Z_" + id + ".tmp", + "2026-08-25T16-00-06-143Z_not-a-uuid.jsonl", + id + ".jsonl", + "", + } { + if got := piSessionFileID(bad); got != "" { + t.Errorf("piSessionFileID(%q) = %q, want rejected", bad, got) + } + } +} + +func TestPiAgentDirFromEnv(t *testing.T) { + t.Parallel() + if got := piAgentDirFromEnv([]string{"PI_CODING_AGENT_DIR=/custom/agent", "HOME=/home/x"}); got != "/custom/agent" { + t.Errorf("PI_CODING_AGENT_DIR must win, got %q", got) + } + if got := piAgentDirFromEnv([]string{"HOME=/home/x"}); got != filepath.Join("/home/x", ".pi", "agent") { + t.Errorf("home-derived agent dir = %q", got) + } + if got := piAgentDirFromEnv([]string{"HOME=/first", "HOME=/last"}); got != filepath.Join("/last", ".pi", "agent") { + t.Errorf("last env entry must win, got %q", got) + } +} + +// piWatchFixture collects watcher callbacks for assertions. +type piWatchFixture struct { + mu sync.Mutex + activities []ActivityKind + notes []string + headerID atomic.Value // stores string; late stdout header simulation +} + +func (f *piWatchFixture) setHeaderID(id string) { f.headerID.Store(id) } + +func (f *piWatchFixture) headerIDString() string { + id, _ := f.headerID.Load().(string) + return id +} + +func (f *piWatchFixture) onActivity(kind ActivityKind) { + f.mu.Lock() + defer f.mu.Unlock() + f.activities = append(f.activities, kind) +} + +func (f *piWatchFixture) onNote(msg string) { + f.mu.Lock() + defer f.mu.Unlock() + f.notes = append(f.notes, msg) +} + +func (f *piWatchFixture) sessionActivityCount() int { + f.mu.Lock() + defer f.mu.Unlock() + n := 0 + for _, k := range f.activities { + if k == ActivitySession { + n++ + } + } + return n +} + +func (f *piWatchFixture) notesJoined() string { + f.mu.Lock() + defer f.mu.Unlock() + return strings.Join(f.notes, "\n") +} + +func shortPiWatchPoll(t *testing.T) { + t.Helper() + prev := piSessionWatchPollInterval + piSessionWatchPollInterval = 10 * time.Millisecond + t.Cleanup(func() { piSessionWatchPollInterval = prev }) +} + +func writePiSessionFile(t *testing.T, dir, name string, lines ...string) string { + t.Helper() + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + path := filepath.Join(dir, name) + if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o644); err != nil { + t.Fatal(err) + } + return path +} + +func appendToFile(t *testing.T, path, data string) { + t.Helper() + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + t.Fatal(err) + } + if _, err := f.WriteString(data); err != nil { + t.Fatal(err) + } + if err := f.Close(); err != nil { + t.Fatal(err) + } +} + +const piWatchTestSessionID = "019ff2f3-5f31-744b-90b8-679074ff7686" + +func piWatchEnv(home string) []string { return []string{"HOME=" + home} } + +func waitForCondition(t *testing.T, timeout time.Duration, what string, ok func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if ok() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) +} + +func TestPiSessionWatcher_ResumeBindsAndCreditsAdvancement(t *testing.T) { + shortPiWatchPoll(t) + home := t.TempDir() + cwd := t.TempDir() + dir := piSessionDirForCWD(piAgentDirFromEnv(piWatchEnv(home)), cwd) + path := writePiSessionFile(t, dir, "2026-08-25T16-00-06-143Z_"+piWatchTestSessionID+".jsonl", `{"type":"session","id":"`+piWatchTestSessionID+`"}`) + + fx := &piWatchFixture{} + w := startPiSessionWatcher(context.Background(), nil, piWatchEnv(home), cwd, + &SessionRef{ID: piWatchTestSessionID}, func() string { return piWatchTestSessionID }, fx.onActivity, fx.onNote) + if w == nil { + t.Fatal("resume with an exact session file must bind a watcher") + } + defer w.shutdown() + + waitForCondition(t, 2*time.Second, "resume binding note", func() bool { + return strings.Contains(fx.notesJoined(), "watching session "+piWatchTestSessionID) + }) + if got := fx.sessionActivityCount(); got != 0 { + t.Fatalf("pre-existing session content credited %d activities; only new advancement counts", got) + } + + appendToFile(t, path, `{"type":"message","message":{"role":"assistant","content":"working"}}`) + waitForCondition(t, 2*time.Second, "session advancement activity", func() bool { + return fx.sessionActivityCount() > 0 + }) +} + +func TestPiSessionWatcher_FreshSessionBindsToTheOneNewFile(t *testing.T) { + shortPiWatchPoll(t) + home := t.TempDir() + cwd := t.TempDir() + dir := piSessionDirForCWD(piAgentDirFromEnv(piWatchEnv(home)), cwd) + // A pre-existing unrelated session in the same dir must not count as new. + writePiSessionFile(t, dir, "2026-08-20T10-00-00-000Z_019fe058-c457-72a1-a555-fcfaa1a9f7bd.jsonl", `{"type":"session"}`) + + fx := &piWatchFixture{} + w := startPiSessionWatcher(context.Background(), nil, piWatchEnv(home), cwd, + &SessionRef{}, fx.headerIDString, fx.onActivity, fx.onNote) + if w == nil { + t.Fatal("durable session must start a watcher") + } + defer w.shutdown() + + path := writePiSessionFile(t, dir, "2026-08-25T16-00-06-143Z_"+piWatchTestSessionID+".jsonl", `{"type":"session","id":"`+piWatchTestSessionID+`"}`) + // The buffered stdout header arrives later and confirms the same id. + fx.setHeaderID(piWatchTestSessionID) + waitForCondition(t, 2*time.Second, "fresh binding note", func() bool { + return strings.Contains(fx.notesJoined(), "watching session "+piWatchTestSessionID) + }) + + appendToFile(t, path, `{"type":"message"}`) + waitForCondition(t, 2*time.Second, "fresh session advancement", func() bool { + return fx.sessionActivityCount() > 0 + }) +} + +func TestPiSessionWatcher_TwoNewFilesAreAmbiguousAndCreditNothing(t *testing.T) { + shortPiWatchPoll(t) + home := t.TempDir() + cwd := t.TempDir() + dir := piSessionDirForCWD(piAgentDirFromEnv(piWatchEnv(home)), cwd) + + fx := &piWatchFixture{} + w := startPiSessionWatcher(context.Background(), nil, piWatchEnv(home), cwd, + &SessionRef{}, func() string { return "" }, fx.onActivity, fx.onNote) + if w == nil { + t.Fatal("watcher must start so it can detect ambiguity") + } + defer w.shutdown() + + writePiSessionFile(t, dir, "2026-08-25T16-00-06-143Z_"+piWatchTestSessionID+".jsonl", `{"type":"session"}`) + writePiSessionFile(t, dir, "2026-08-25T16-00-07-000Z_019fe058-c457-72a1-a555-fcfaa1a9f7bd.jsonl", `{"type":"session"}`) + waitForCondition(t, 2*time.Second, "ambiguity note", func() bool { + return strings.Contains(fx.notesJoined(), "ambiguous") + }) + appendToFile(t, filepath.Join(dir, "2026-08-25T16-00-06-143Z_"+piWatchTestSessionID+".jsonl"), `{"type":"message"}`) + // Give the watcher several ticks to prove it never credits ambiguous files. + time.Sleep(100 * time.Millisecond) + if got := fx.sessionActivityCount(); got != 0 { + t.Fatalf("ambiguous binding credited %d session activities; want conservative stdout-only behavior", got) + } +} + +func TestPiSessionWatcher_HeaderMismatchUnbinds(t *testing.T) { + shortPiWatchPoll(t) + home := t.TempDir() + cwd := t.TempDir() + dir := piSessionDirForCWD(piAgentDirFromEnv(piWatchEnv(home)), cwd) + + fx := &piWatchFixture{} + w := startPiSessionWatcher(context.Background(), nil, piWatchEnv(home), cwd, + &SessionRef{}, fx.headerIDString, fx.onActivity, fx.onNote) + if w == nil { + t.Fatal("watcher must start") + } + defer w.shutdown() + + path := writePiSessionFile(t, dir, "2026-08-25T16-00-06-143Z_"+piWatchTestSessionID+".jsonl", `{"type":"session"}`) + waitForCondition(t, 2*time.Second, "binding note", func() bool { + return strings.Contains(fx.notesJoined(), "watching session") + }) + + // The late stdout header names a DIFFERENT session: the bound file is not + // this invocation's, so the watcher must fail closed and stop crediting. + fx.setHeaderID("019fe058-c457-72a1-a555-fcfaa1a9f7bd") + waitForCondition(t, 2*time.Second, "mismatch unbind note", func() bool { + return strings.Contains(fx.notesJoined(), "does not match") + }) + creditedAtUnbind := fx.sessionActivityCount() + appendToFile(t, path, `{"type":"message"}`) + time.Sleep(100 * time.Millisecond) + if got := fx.sessionActivityCount(); got != creditedAtUnbind { + t.Fatalf("unbound watcher credited %d further session activities", got-creditedAtUnbind) + } +} + +func TestPiSessionWatcher_SessionFreeInvocationHasNoWatcher(t *testing.T) { + fx := &piWatchFixture{} + if w := startPiSessionWatcher(context.Background(), nil, piWatchEnv(t.TempDir()), t.TempDir(), + nil, func() string { return "" }, fx.onActivity, fx.onNote); w != nil { + t.Fatal("--no-session invocations persist nothing and must not be watched") + } +} + +func TestPiSessionWatcher_RelocatedSessionStorageDisablesBinding(t *testing.T) { + fx := &piWatchFixture{} + if w := startPiSessionWatcher(context.Background(), nil, []string{"HOME=" + t.TempDir(), "PI_CODING_AGENT_SESSION_DIR=/flat/sessions"}, t.TempDir(), + &SessionRef{}, func() string { return "" }, fx.onActivity, fx.onNote); w != nil { + t.Fatal("PI_CODING_AGENT_SESSION_DIR relocation must disable binding") + } + if !strings.Contains(fx.notesJoined(), "stdout-only liveness") { + t.Fatalf("relocation note = %q, want stdout-only explanation", fx.notesJoined()) + } + + fx2 := &piWatchFixture{} + if w := startPiSessionWatcher(context.Background(), []string{"--session-dir", "/flat/sessions"}, piWatchEnv(t.TempDir()), t.TempDir(), + &SessionRef{}, func() string { return "" }, fx2.onActivity, fx2.onNote); w != nil { + t.Fatal("--session-dir override must disable binding") + } +} + +func TestPiSessionWatcher_StopsWithContext(t *testing.T) { + shortPiWatchPoll(t) + home := t.TempDir() + cwd := t.TempDir() + dir := piSessionDirForCWD(piAgentDirFromEnv(piWatchEnv(home)), cwd) + path := writePiSessionFile(t, dir, "2026-08-25T16-00-06-143Z_"+piWatchTestSessionID+".jsonl", `{"type":"session"}`) + + ctx, cancel := context.WithCancel(context.Background()) + fx := &piWatchFixture{} + w := startPiSessionWatcher(ctx, nil, piWatchEnv(home), cwd, + &SessionRef{ID: piWatchTestSessionID}, func() string { return piWatchTestSessionID }, fx.onActivity, fx.onNote) + if w == nil { + t.Fatal("resume must bind") + } + waitForCondition(t, 2*time.Second, "binding note", func() bool { + return strings.Contains(fx.notesJoined(), "watching session") + }) + cancel() + w.shutdown() // must return promptly: the goroutine is owned by ctx+shutdown + appendToFile(t, path, `{"type":"message"}`) + time.Sleep(50 * time.Millisecond) + if got := fx.sessionActivityCount(); got != 0 { + t.Fatalf("stopped watcher credited %d activities", got) + } +} + +func TestStartNativeAgentCommand_ReportsLifecycleAndStdoutActivity(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX shell fixture") + } + var mu sync.Mutex + var kinds []ActivityKind + record := func(kind ActivityKind) { + mu.Lock() + defer mu.Unlock() + kinds = append(kinds, kind) + } + cmd := exec.Command("sh", "-c", "printf hello; sleep 0.05") + started, err := startNativeAgentCommand(cmd, record) + if err != nil { + t.Fatalf("start: %v", err) + } + stderrDone := make(chan struct{}) + go func() { + defer close(stderrDone) + _, _ = io.ReadAll(started.stderr) + }() + buf := make([]byte, 16) + var out strings.Builder + for { + n, err := started.stdout.Read(buf) + if n > 0 { + out.Write(buf[:n]) + } + if err != nil { + break + } + } + <-stderrDone + _ = started.wait() + started.closePipes() + if out.String() != "hello" { + t.Fatalf("stdout = %q, want hello", out.String()) + } + mu.Lock() + defer mu.Unlock() + var sawLifecycle, sawStdout bool + for _, k := range kinds { + sawLifecycle = sawLifecycle || k == ActivityLifecycle + sawStdout = sawStdout || k == ActivityStdout + } + if !sawLifecycle { + t.Fatal("process start did not report lifecycle activity") + } + if !sawStdout { + t.Fatal("stdout bytes did not report stdout activity") + } +} diff --git a/internal/agent/pi_test.go b/internal/agent/pi_test.go index 630f6f383..0de131485 100644 --- a/internal/agent/pi_test.go +++ b/internal/agent/pi_test.go @@ -446,8 +446,8 @@ func TestPiParser_CapturesFirstValidSessionHeader(t *testing.T) { if err := pp.parse(context.Background(), strings.NewReader(stream)); err != nil { t.Fatalf("parse: %v", err) } - if pp.sessionID != sessionID { - t.Fatalf("session ID = %q, want first valid %q", pp.sessionID, sessionID) + if pp.sessionIDString() != sessionID { + t.Fatalf("session ID = %q, want first valid %q", pp.sessionIDString(), sessionID) } } diff --git a/internal/agent/reap_unix_test.go b/internal/agent/reap_unix_test.go index 68de75a4f..6d538eca5 100644 --- a/internal/agent/reap_unix_test.go +++ b/internal/agent/reap_unix_test.go @@ -33,7 +33,7 @@ func TestNativeAgentCommand_WaitDelayClosesEscapedPipeHolder(t *testing.T) { shellenv.ConfigureShellCommand(cmd) cmd.WaitDelay = 100 * time.Millisecond - started, err := startNativeAgentCommand(cmd) + started, err := startNativeAgentCommand(cmd, nil) if err != nil { t.Fatalf("startNativeAgentCommand: %v", err) } diff --git a/internal/config/config.go b/internal/config/config.go index cdce06ab2..e72e12fc6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -40,18 +40,22 @@ const ( // DefaultStepQuietWarning is how long a running/fixing step can go without // a new log or lifecycle activity before AXI status marks it quiet. DefaultStepQuietWarning = 10 * time.Minute - // DefaultAgentTimeout bounds one pipeline agent invocation that does not - // install a more specific deadline, so a stalled agent cannot leave a run - // active forever. Review and Test keep their own knobs; this is the - // default-by-construction budget for every other step. + // DefaultAgentTimeout is the silence budget for one pipeline agent + // invocation that does not install a more specific deadline: the + // invocation is killed only after this long with no reported activity + // (stdout bytes, native process lifecycle, or a bound adapter-native + // session), so a stalled agent cannot leave a run active forever while a + // healthy long turn is never cut off mid-work. Review and Test keep + // their own knobs; this is the default-by-construction budget for every + // other step. DefaultAgentTimeout = 30 * time.Minute - // DefaultReviewAgentTimeout bounds one review round, including its optional - // review-fix and rereview turns, so a stalled agent cannot leave a run - // active forever. + // DefaultReviewAgentTimeout is the per-turn silence budget for the Review + // step's agent turns: the optional review-fix turn and the rereview turn + // each get their own fresh budget measured from that turn's own activity. DefaultReviewAgentTimeout = 30 * time.Minute - // DefaultTestAgentTimeout bounds one Test-step agent invocation, including - // the post-test evidence-gathering turn and a Test-repair turn, so a stalled - // agent cannot leave a run active forever. + // DefaultTestAgentTimeout is the per-invocation silence budget for one + // Test-step agent invocation: the post-test evidence-gathering turn and a + // Test-repair turn each get their own budget of this size. DefaultTestAgentTimeout = 30 * time.Minute // DefaultDaemonConnectTimeout bounds client IPC connection attempts to a // daemon socket that exists but is not accepting connections. diff --git a/internal/pipeline/agent_run.go b/internal/pipeline/agent_run.go index da31ec1de..310735958 100644 --- a/internal/pipeline/agent_run.go +++ b/internal/pipeline/agent_run.go @@ -28,38 +28,58 @@ func AgentTimeout(cfg *config.Config) time.Duration { // call. The parent StepContext.Ctx is left unchanged so post-agent work // (commits, git, parsing) is not cancelled by the invocation budget. // -// If the parent context already has a deadline (review's round budget, Test's -// explicit wrap, intent extraction, caller cancellation), that bound is -// honored and no shorter default is stacked. Otherwise AgentTimeout is -// applied. A late successful return after the deadline is rejected. +// If the parent context already has a deadline (intent extraction, caller +// cancellation), that bound is honored and no shorter default is stacked. +// Otherwise AgentTimeout is applied as a silence budget: the invocation is +// cancelled only after that long with no reported activity (stdout bytes, +// native process lifecycle, or a bound adapter-native session). A late +// successful return after the deadline is rejected. func (sctx *StepContext) RunAgent(opts agent.RunOpts) (*agent.Result, error) { parent := context.Background() if sctx != nil { parent = sctx.Ctx } - return sctx.runAgent(parent, opts, "") + return sctx.runAgent(parent, 0, opts, "") } // RunAgentContext is RunAgent with an explicit parent, used when a step has -// already installed a more specific deadline (review round, Test invocation). +// already installed a more specific deadline. func (sctx *StepContext) RunAgentContext(parent context.Context, opts agent.RunOpts) (*agent.Result, error) { - return sctx.runAgent(parent, opts, "") + return sctx.runAgent(parent, 0, opts, "") +} + +// RunAgentBudget is RunAgentContext with an explicit silence budget in place +// of the default AgentTimeout. Steps with their own configured budget +// (review_agent_timeout, test_agent_timeout) pass it per invocation so every +// agent turn gets the full budget measured from that turn's own activity, +// never an earlier turn's remainder. +func (sctx *StepContext) RunAgentBudget(parent context.Context, budget time.Duration, opts agent.RunOpts) (*agent.Result, error) { + return sctx.runAgent(parent, budget, opts, "") } // RunAgentSessionContext is RunAgentSession with an explicit parent so a -// fixer turn can share a round budget (review) or a per-invocation wrap (Test). +// fixer turn can share a round budget or a per-invocation wrap. func (sctx *StepContext) RunAgentSessionContext(parent context.Context, role SessionRole, opts agent.RunOpts) (*agent.Result, error) { - return sctx.runAgent(parent, opts, role) + return sctx.runAgent(parent, 0, opts, role) +} + +// RunAgentSessionBudget is RunAgentSessionContext with an explicit silence +// budget, the session-bearing analogue of RunAgentBudget. +func (sctx *StepContext) RunAgentSessionBudget(parent context.Context, budget time.Duration, role SessionRole, opts agent.RunOpts) (*agent.Result, error) { + return sctx.runAgent(parent, budget, opts, role) } -func (sctx *StepContext) runAgent(parent context.Context, opts agent.RunOpts, sessionRole SessionRole) (*agent.Result, error) { +func (sctx *StepContext) runAgent(parent context.Context, budget time.Duration, opts agent.RunOpts, sessionRole SessionRole) (*agent.Result, error) { var ag agent.Agent timeout := AgentTimeout(nil) if sctx != nil { ag = sctx.Agent timeout = AgentTimeout(sctx.Config) } - return invokeAgent(parent, timeout, func(ctx context.Context) (*agent.Result, error) { + if budget > 0 { + timeout = budget + } + return invokeAgent(parent, timeout, &opts, func(ctx context.Context) (*agent.Result, error) { if sessionRole != "" && sctx != nil && sctx.Sessions != nil { return sctx.Sessions.Run(ctx, ag, sessionRole, opts, sctx.Log) } @@ -70,8 +90,30 @@ func (sctx *StepContext) runAgent(parent context.Context, opts agent.RunOpts, se }) } -func invokeAgent(parent context.Context, timeout time.Duration, run func(context.Context) (*agent.Result, error)) (*agent.Result, error) { - ctx, cancel, applied := bindAgentDeadline(parent, timeout) +// livenessContextKey carries the invocation's liveness owner so nested seams +// (RunAgent outside, the executor's timeoutAgent backstop inside) share one +// monotonic clock instead of stacking duplicate watchdogs. +type livenessContextKey struct{} + +func livenessFromContext(ctx context.Context) *invocationLiveness { + if ctx == nil { + return nil + } + l, _ := ctx.Value(livenessContextKey{}).(*invocationLiveness) + return l +} + +func invokeAgent(parent context.Context, timeout time.Duration, opts *agent.RunOpts, run func(context.Context) (*agent.Result, error)) (*agent.Result, error) { + ctx, cancel, applied, liveness := bindAgentLiveness(parent, timeout) + if liveness != nil && opts != nil { + previous := opts.OnActivity + opts.OnActivity = func(kind agent.ActivityKind) { + liveness.record(kind) + if previous != nil { + previous(kind) + } + } + } result, err := run(ctx) runErr := classifyAgentRun(ctx, applied, err) cancel() @@ -81,25 +123,39 @@ func invokeAgent(parent context.Context, timeout time.Duration, run func(context return result, nil } -func bindAgentDeadline(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc, time.Duration) { +// bindAgentLiveness installs the invocation's silence watchdog unless an +// outer layer already owns one. Precedence: an inherited liveness owner is +// reused (single monotonic clock); an existing parent deadline is honored +// unchanged (legacy hard bound); otherwise a fresh watchdog with the given +// budget governs. The returned duration is the budget actually applied at +// this layer (0 when an outer bound governs), and the liveness is non-nil +// exactly when activity should be reported to it. +func bindAgentLiveness(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc, time.Duration, *invocationLiveness) { if parent == nil { parent = context.Background() } + if liveness := livenessFromContext(parent); liveness != nil { + return parent, func() {}, 0, liveness + } if timeout <= 0 { - return parent, func() {}, 0 + return parent, func() {}, 0, nil } if _, ok := parent.Deadline(); ok { - return parent, func() {}, 0 + return parent, func() {}, 0, nil } - ctx, cancel := context.WithTimeoutCause(parent, timeout, ErrAgentTimeout) - return ctx, cancel, timeout + liveness := newInvocationLiveness() + ctx, cancel := watchSilence(parent, timeout, liveness) + return context.WithValue(ctx, livenessContextKey{}, liveness), cancel, timeout, liveness } func classifyAgentRun(ctx context.Context, applied time.Duration, err error) error { - if applied > 0 && errors.Is(context.Cause(ctx), ErrAgentTimeout) { - return fmt.Errorf("agent timed out after %s (agent silent for %s): %w", applied, applied, ErrAgentTimeout) - } if cause := context.Cause(ctx); cause != nil { + if applied > 0 && errors.Is(cause, ErrAgentTimeout) { + if ate := asAgentTimeout(cause); ate != nil { + return ate + } + return fmt.Errorf("agent timed out after %s (agent silent for %s): %w", applied, applied, ErrAgentTimeout) + } return cause } if err != nil { @@ -109,8 +165,9 @@ func classifyAgentRun(ctx context.Context, applied time.Duration, err error) err } // timeoutAgent is the executor backstop: every sctx.Agent.Run is bounded even -// if a future step forgets RunAgent. Nested with RunAgent it is a no-op when -// the incoming context already has a deadline. +// if a future step forgets RunAgent. Nested with RunAgent it shares the outer +// invocation's liveness owner instead of stacking a second watchdog, and it +// honors an incoming context's existing deadline as before. type timeoutAgent struct { inner agent.Agent timeout time.Duration @@ -121,7 +178,7 @@ func (a *timeoutAgent) Name() string { return a.inner.Name() } func (a *timeoutAgent) Close() error { return a.inner.Close() } func (a *timeoutAgent) Run(ctx context.Context, opts agent.RunOpts) (*agent.Result, error) { - return invokeAgent(ctx, a.timeout, func(runCtx context.Context) (*agent.Result, error) { + return invokeAgent(ctx, a.timeout, &opts, func(runCtx context.Context) (*agent.Result, error) { return a.inner.Run(runCtx, opts) }) } diff --git a/internal/pipeline/agent_run_pi_test.go b/internal/pipeline/agent_run_pi_test.go new file mode 100644 index 000000000..8409254f8 --- /dev/null +++ b/internal/pipeline/agent_run_pi_test.go @@ -0,0 +1,207 @@ +package pipeline + +import ( + "context" + "errors" + "os" + "path/filepath" + "runtime" + "strconv" + "strings" + "testing" + "time" + + "github.com/kunchenguid/no-mistakes/internal/agent" + "github.com/kunchenguid/no-mistakes/internal/config" + "github.com/kunchenguid/no-mistakes/internal/types" +) + +// agent_run_pi_test.go is process-level coverage of the pi liveness fix: a +// real piAgent adapter drives a fake pi subprocess whose stdout stays quiet +// (pi buffers it when piped) while its exact session JSONL keeps advancing. +// The fake redirects session storage through PI_CODING_AGENT_DIR so neither +// the adapter's watcher nor the fixture ever touches the operator's real +// ~/.pi. The budgets are sized above the watcher's 2s poll interval so the +// tests exercise the production polling path without test hooks. + +const piLivenessTestSessionID = "019ff2f3-5f31-744b-90b8-679074ff7686" + +// fakePiSessionScript appends one session event every 250ms for the given +// count while printing nothing, then emits the JSON-mode session header and +// agent_end. With count=0 it writes the header event to the session file and +// then freezes forever (until the watchdog kills the process tree). +func fakePiSessionScript(appendCount int) string { + return `#!/bin/sh +cat > /dev/null +# pi encodes process.cwd(), the physical path: pwd -P matches it. +enc=$(pwd -P | sed -e 's|^/||' -e 's|/|-|g') +session_dir="$PI_CODING_AGENT_DIR/sessions/--${enc}--" +mkdir -p "$session_dir" +session_file="$session_dir/2026-08-26T00-00-00-000Z_` + piLivenessTestSessionID + `.jsonl" +printf '%s\n' '{"type":"session","id":"` + piLivenessTestSessionID + `","timestamp":"2026-08-26T00:00:00Z","cwd":"'"$(pwd)"'"}' >> "$session_file" +i=0 +while [ "$i" -lt ` + strconv.Itoa(appendCount) + ` ]; do + printf '%s\n' '{"type":"message","message":{"role":"assistant","content":"working"}}' >> "$session_file" + sleep 0.25 + i=$((i+1)) +done +if [ ` + strconv.Itoa(appendCount) + ` -eq 0 ]; then + while true; do sleep 1; done +fi +printf '%s\n' '{"type":"session","id":"` + piLivenessTestSessionID + `"}' +printf '%s\n' '{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"done"}]}]}' +` +} + +func fmtInt(n int) string { + if n == 0 { + return "0" + } + digits := "" + for n > 0 { + digits = string(rune('0'+n%10)) + digits + n /= 10 + } + return digits +} + +func writeFakePiScript(t *testing.T, script string) string { + t.Helper() + bin := filepath.Join(t.TempDir(), "pi") + if err := os.WriteFile(bin, []byte(script), 0o755); err != nil { + t.Fatalf("write fake pi: %v", err) + } + return bin +} + +func newPiStepContext(t *testing.T, bin, agentDir string, budget time.Duration) *StepContext { + t.Helper() + ag, err := agent.NewWithOptions(types.AgentPi, bin, nil, agent.Options{}) + if err != nil { + t.Fatalf("new pi agent: %v", err) + } + return &StepContext{ + Ctx: context.Background(), + Agent: ag, + Config: &config.Config{AgentTimeout: budget}, + Env: []string{"PI_CODING_AGENT_DIR=" + agentDir}, + } +} + +// The incident class from run 01M0WRW741G2YPQFXYE5TG6DZD: the fixer's stdout +// was buffered into silence while its pi session stayed healthy to the end. +// The invocation must survive far past the configured silence budget on +// session-event evidence alone. +func TestRunAgent_PiSessionActivityKeepsBufferedStdoutInvocationAlive(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX shell fixture") + } + agentDir := t.TempDir() + workDir := t.TempDir() + bin := writeFakePiScript(t, fakePiSessionScript(26)) // ~6.5s of session activity + + sctx := newPiStepContext(t, bin, agentDir, 3*time.Second) + opts := agent.RunOpts{ + Prompt: "work", + CWD: workDir, + Env: sctx.Env, + Session: &agent.SessionRef{}, // durable: pi persists a session JSONL + } + start := time.Now() + result, err := sctx.RunAgent(opts) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("invocation with an advancing session must survive a quiet stdout, got %v (after %s)", err, elapsed) + } + if result == nil || result.SessionID != piLivenessTestSessionID { + t.Fatalf("result = %+v, want session %s", result, piLivenessTestSessionID) + } + if elapsed < 6*time.Second { + t.Fatalf("invocation ended after %s; the fake agent was active for ~6.5s, so the run was cut short", elapsed) + } + if !strings.Contains(result.Text, "done") { + t.Fatalf("result text = %q, want the agent's final output", result.Text) + } +} + +// The guard that must never weaken: when neither stdout nor the bound session +// makes progress for the full budget, the invocation is terminated with its +// process tree and the diagnostic names the evidence. +func TestRunAgent_FrozenPiSessionIsKilledAfterTheFullBudget(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX shell fixture") + } + agentDir := t.TempDir() + workDir := t.TempDir() + bin := writeFakePiScript(t, fakePiSessionScript(0)) // session header, then frozen + + sctx := newPiStepContext(t, bin, agentDir, 3*time.Second) + opts := agent.RunOpts{ + Prompt: "work", + CWD: workDir, + Env: sctx.Env, + Session: &agent.SessionRef{}, + } + start := time.Now() + _, err := sctx.RunAgent(opts) + elapsed := time.Since(start) + if err == nil { + t.Fatal("a frozen agent with no stdout and no session advancement must be killed") + } + if !errors.Is(err, ErrAgentTimeout) { + t.Fatalf("error = %v, want ErrAgentTimeout", err) + } + var ate *AgentTimeoutError + if !errors.As(err, &ate) { + t.Fatalf("error = %v, want AgentTimeoutError evidence", err) + } + if !strings.Contains(ate.Evidence, "pi session events") { + t.Fatalf("evidence = %q, want the bound session's last activity named", ate.Evidence) + } + // Killed by its own budget, not instantly and not never: last session + // event lands within one poll interval of launch, then the full budget + // of silence must elapse. + if elapsed < 3*time.Second { + t.Fatalf("killed after %s, before the full 3s silence budget elapsed", elapsed) + } + if elapsed > 30*time.Second { + t.Fatalf("kill took %s; the watchdog must bound a frozen invocation", elapsed) + } +} + +// A session-free pi invocation (--no-session) has no session file to bind: +// only stdout and lifecycle evidence govern it. With a quiet stdout it is +// genuinely silent and must be killed at the budget. +func TestRunAgent_SessionFreePiQuietStdoutIsKilledAtBudget(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX shell fixture") + } + agentDir := t.TempDir() + workDir := t.TempDir() + bin := writeFakePiScript(t, `#!/bin/sh +cat > /dev/null +while true; do sleep 1; done +`) + + sctx := newPiStepContext(t, bin, agentDir, 2*time.Second) + // No durable session: buildArgs adds --no-session, pi persists nothing. + opts := agent.RunOpts{Prompt: "work", CWD: workDir, Env: sctx.Env} + start := time.Now() + _, err := sctx.RunAgent(opts) + elapsed := time.Since(start) + if err == nil || !errors.Is(err, ErrAgentTimeout) { + t.Fatalf("error = %v, want ErrAgentTimeout", err) + } + var ate *AgentTimeoutError + if errors.As(err, &ate) { + if strings.Contains(ate.Evidence, "pi session events") { + t.Fatalf("evidence = %q; a --no-session invocation must never credit session activity", ate.Evidence) + } + if !strings.Contains(ate.Evidence, "process lifecycle") { + t.Fatalf("evidence = %q, want only lifecycle (process start) evidence", ate.Evidence) + } + } + if elapsed < 2*time.Second || elapsed > 20*time.Second { + t.Fatalf("killed after %s, want ~2s silence budget", elapsed) + } +} diff --git a/internal/pipeline/agent_run_test.go b/internal/pipeline/agent_run_test.go index a5d122d4d..397ce679f 100644 --- a/internal/pipeline/agent_run_test.go +++ b/internal/pipeline/agent_run_test.go @@ -91,9 +91,12 @@ func TestRunAgent_SuccessfulOutputUnchanged(t *testing.T) { want := json.RawMessage(`{"summary":"ok"}`) ag := &hangingAgent{ name: "ok", - runFn: func(ctx context.Context, _ agent.RunOpts) (*agent.Result, error) { - if _, ok := ctx.Deadline(); !ok { - t.Fatal("successful invocation ran without a deadline") + runFn: func(ctx context.Context, opts agent.RunOpts) (*agent.Result, error) { + // The invocation is bounded by the silence watchdog, not a fixed + // deadline: it must hand the adapter an activity sink so stdout, + // lifecycle, and session evidence can re-arm the clock. + if opts.OnActivity == nil { + t.Fatal("successful invocation ran without a liveness sink") } if err := ctx.Err(); err != nil { t.Fatalf("live invocation context: %v", err) diff --git a/internal/pipeline/liveness.go b/internal/pipeline/liveness.go new file mode 100644 index 000000000..013693f95 --- /dev/null +++ b/internal/pipeline/liveness.go @@ -0,0 +1,168 @@ +package pipeline + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + "sync" + "time" + + "github.com/kunchenguid/no-mistakes/internal/agent" +) + +// liveness.go owns the per-invocation silence watchdog. One invocation gets +// one monotonic last-activity clock; every reported activity kind (stdout +// bytes, native process lifecycle, bound adapter-native session advancement) +// folds into that single clock. The watchdog cancels the invocation only +// after the full configured budget elapses with no activity from any source, +// so a healthy turn whose stdout is buffered (pi JSON mode) stays alive on +// session evidence, while a genuinely frozen agent is still terminated with +// its whole process tree after the same budget the fixed deadline used to +// impose. The clock starts at invocation start and every process start +// re-arms it, which is what gives each newly launched agent a fresh full +// budget instead of inheriting an earlier turn's quiet time. + +// invocationLiveness is the single monotonic last-activity owner for one +// agent invocation. +type invocationLiveness struct { + mu sync.Mutex + last time.Time + seen map[agent.ActivityKind]time.Time +} + +func newInvocationLiveness() *invocationLiveness { + return &invocationLiveness{ + last: time.Now(), + seen: make(map[agent.ActivityKind]time.Time), + } +} + +func (l *invocationLiveness) record(kind agent.ActivityKind) { + l.mu.Lock() + defer l.mu.Unlock() + now := time.Now() + l.last = now + l.seen[kind] = now +} + +// evidence renders the per-kind activity summary for the timeout diagnostic. +// It names evidence classes and ages only - never content, paths, or session +// payloads - so operators can distinguish stdout-active, session-active, and +// genuinely quiet invocations. +func (l *invocationLiveness) evidence() string { + l.mu.Lock() + defer l.mu.Unlock() + if len(l.seen) == 0 { + return "no stdout, lifecycle, or session-event activity observed this invocation" + } + type kindAge struct { + label string + age time.Duration + at time.Time + } + now := time.Now() + parts := make([]kindAge, 0, len(l.seen)) + for kind, at := range l.seen { + parts = append(parts, kindAge{label: agent.ActivityKindLabel(kind), age: now.Sub(at).Round(time.Millisecond), at: at}) + } + sort.Slice(parts, func(i, j int) bool { return parts[i].at.After(parts[j].at) }) + rendered := make([]string, 0, len(parts)) + for _, part := range parts { + rendered = append(rendered, fmt.Sprintf("%s %s ago", part.label, part.age)) + } + return "last activity: " + strings.Join(rendered, ", ") +} + +// AgentTimeoutError is the watchdog's cancellation cause and diagnostic. It +// carries the applied budget plus the liveness evidence at kill time so step +// error mapping can re-render the same evidence under a step-specific label +// without re-deriving it. It unwraps to ErrAgentTimeout so existing +// errors.Is callers keep working. +type AgentTimeoutError struct { + Budget time.Duration + Evidence string +} + +func (e *AgentTimeoutError) Error() string { + return fmt.Sprintf("agent timed out after %s (agent silent for %s%s): %s", + e.Budget, e.Budget, evidenceClause(e.Evidence), ErrAgentTimeout) +} + +func (e *AgentTimeoutError) Unwrap() error { return ErrAgentTimeout } + +// StepError re-renders the timeout under a step-owned label, preserving the +// budget and evidence: "agent review timed out after 30m0s (review agent +// silent for 30m0s: ): agent timeout". +func (e *AgentTimeoutError) StepError(prefix, silentLabel string) error { + return fmt.Errorf("%s timed out after %s (%s silent for %s%s): %w", + prefix, e.Budget, silentLabel, e.Budget, evidenceClause(e.Evidence), ErrAgentTimeout) +} + +func evidenceClause(evidence string) string { + if evidence == "" { + return "" + } + return ": " + evidence +} + +// watchSilence derives a cancellation context from parent and starts the +// watchdog goroutine for l. When l stays silent for the whole budget, the +// watchdog cancels the context with an *AgentTimeoutError cause. The +// goroutine is owned by the returned context: it exits when the watchdog +// fires or when the caller cancels, whichever comes first. +func watchSilence(parent context.Context, budget time.Duration, l *invocationLiveness) (context.Context, context.CancelFunc) { + ctx, cancelCause := context.WithCancelCause(parent) + cancel := func() { cancelCause(nil) } + go func() { + ticker := time.NewTicker(silenceWatchInterval(budget)) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + l.mu.Lock() + silentFor := time.Since(l.last) + l.mu.Unlock() + if silentFor >= budget { + cancelCause(&AgentTimeoutError{Budget: budget, Evidence: l.evidence()}) + return + } + } + }() + return ctx, cancel +} + +// silenceWatchInterval scales the watchdog tick to the budget so short test +// budgets still fire promptly while production budgets poll cheaply. +func silenceWatchInterval(budget time.Duration) time.Duration { + interval := budget / 10 + if interval < 2*time.Millisecond { + return 2 * time.Millisecond + } + if interval > 10*time.Second { + return 10 * time.Second + } + return interval +} + +// asAgentTimeout extracts the watchdog's timeout diagnostic from an error +// chain, returning nil when the failure was something else. +func asAgentTimeout(err error) *AgentTimeoutError { + return AsAgentTimeout(err) +} + +// AsAgentTimeout extracts the silence watchdog's timeout diagnostic from an +// error chain, returning nil when the failure was something else. Steps with +// a step-labeled timeout message use it to re-render the budget and evidence +// under their own label. +func AsAgentTimeout(err error) *AgentTimeoutError { + var ate *AgentTimeoutError + if errors.As(err, &ate) { + return ate + } + return nil +} diff --git a/internal/pipeline/liveness_test.go b/internal/pipeline/liveness_test.go new file mode 100644 index 000000000..8af0a6561 --- /dev/null +++ b/internal/pipeline/liveness_test.go @@ -0,0 +1,161 @@ +package pipeline + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/kunchenguid/no-mistakes/internal/agent" +) + +func TestWatchSilence_ActivityReArmsTheClock(t *testing.T) { + t.Parallel() + l := newInvocationLiveness() + ctx, cancel := watchSilence(context.Background(), 120*time.Millisecond, l) + defer cancel() + + // Keep the invocation alive past the budget with periodic activity, then + // go quiet: the watchdog must fire only after a full silent budget. + deadline := time.Now().Add(3 * 120 * time.Millisecond) + for time.Now().Before(deadline) { + l.record(agent.ActivityStdout) + select { + case <-ctx.Done(): + t.Fatalf("watchdog fired during active period (cause: %v)", context.Cause(ctx)) + case <-time.After(30 * time.Millisecond): + } + } + select { + case <-ctx.Done(): + case <-time.After(2 * time.Second): + t.Fatal("watchdog did not fire after a full silent budget") + } + var ate *AgentTimeoutError + if !errors.As(context.Cause(ctx), &ate) { + t.Fatalf("cause = %v, want AgentTimeoutError", context.Cause(ctx)) + } + if ate.Budget != 120*time.Millisecond { + t.Fatalf("budget = %s, want 120ms", ate.Budget) + } + if !strings.Contains(ate.Evidence, "stdout bytes") { + t.Fatalf("evidence = %q, want stdout activity named", ate.Evidence) + } +} + +func TestWatchSilence_QuietInvocationFiresAfterFullBudget(t *testing.T) { + t.Parallel() + l := newInvocationLiveness() + start := time.Now() + ctx, cancel := watchSilence(context.Background(), 80*time.Millisecond, l) + defer cancel() + select { + case <-ctx.Done(): + case <-time.After(2 * time.Second): + t.Fatal("watchdog did not fire") + } + elapsed := time.Since(start) + if elapsed < 70*time.Millisecond { + t.Fatalf("watchdog fired after %s, before the 80ms budget elapsed", elapsed) + } + var ate *AgentTimeoutError + if !errors.As(context.Cause(ctx), &ate) { + t.Fatalf("cause = %v, want AgentTimeoutError", context.Cause(ctx)) + } + if !strings.Contains(ate.Evidence, "no stdout, lifecycle, or session-event activity observed") { + t.Fatalf("evidence = %q, want genuinely-quiet wording", ate.Evidence) + } +} + +func TestWatchSilence_ParentCancellationIsNotATimeout(t *testing.T) { + t.Parallel() + parent, parentCancel := context.WithCancel(context.Background()) + l := newInvocationLiveness() + ctx, cancel := watchSilence(parent, time.Hour, l) + defer cancel() + parentCancel() + select { + case <-ctx.Done(): + case <-time.After(2 * time.Second): + t.Fatal("parent cancellation did not propagate") + } + if asAgentTimeout(context.Cause(ctx)) != nil { + t.Fatalf("parent cancellation misread as agent timeout: %v", context.Cause(ctx)) + } +} + +func TestLivenessEvidence_NamesEveryKindSeen(t *testing.T) { + t.Parallel() + l := newInvocationLiveness() + l.record(agent.ActivityLifecycle) + l.record(agent.ActivitySession) + evidence := l.evidence() + if !strings.Contains(evidence, "process lifecycle") || !strings.Contains(evidence, "pi session events") { + t.Fatalf("evidence = %q, want lifecycle and session named", evidence) + } + if strings.Contains(evidence, "stdout bytes") { + t.Fatalf("evidence = %q, stdout was never observed and must not be claimed", evidence) + } + // Most recent first: the session event is newer than the process start. + if strings.Index(evidence, "pi session events") > strings.Index(evidence, "process lifecycle") { + t.Fatalf("evidence = %q, want most recent activity first", evidence) + } +} + +func TestAgentTimeoutError_MessageAndStepLabel(t *testing.T) { + t.Parallel() + ate := &AgentTimeoutError{Budget: 30 * time.Minute, Evidence: "last activity: pi session events 30m0s ago"} + msg := ate.Error() + for _, want := range []string{"agent timed out after 30m0s", "agent silent for 30m0s", "pi session events"} { + if !strings.Contains(msg, want) { + t.Fatalf("message %q missing %q", msg, want) + } + } + if !errors.Is(ate, ErrAgentTimeout) { + t.Fatal("AgentTimeoutError must unwrap to ErrAgentTimeout") + } + stepErr := ate.StepError("agent review", "review agent") + for _, want := range []string{"agent review timed out after 30m0s", "review agent silent for 30m0s", "pi session events"} { + if !strings.Contains(stepErr.Error(), want) { + t.Fatalf("step message %q missing %q", stepErr.Error(), want) + } + } + if !errors.Is(stepErr, ErrAgentTimeout) { + t.Fatal("step-labeled timeout must stay errors.Is-compatible with ErrAgentTimeout") + } +} + +func TestBindAgentLiveness_NestedSeamSharesOneOwner(t *testing.T) { + t.Parallel() + parent := context.Background() + outer, outerCancel, outerApplied, outerLiveness := bindAgentLiveness(parent, time.Hour) + defer outerCancel() + if outerApplied != time.Hour || outerLiveness == nil { + t.Fatalf("outer bind = (%s, %v), want watchdog installed", outerApplied, outerLiveness != nil) + } + inner, innerCancel, innerApplied, innerLiveness := bindAgentLiveness(outer, time.Hour) + defer innerCancel() + if innerLiveness != outerLiveness { + t.Fatal("nested seam stacked a second liveness owner instead of sharing one monotonic clock") + } + if innerApplied != 0 { + t.Fatalf("nested seam applied its own budget %s on top of the outer owner", innerApplied) + } + if _, ok := inner.Deadline(); ok { + t.Fatal("liveness-governed context must not carry a fixed deadline") + } +} + +func TestBindAgentLiveness_ExistingDeadlineStaysAHardBound(t *testing.T) { + t.Parallel() + parent, cancel := context.WithTimeout(context.Background(), time.Hour) + defer cancel() + ctx, _, applied, liveness := bindAgentLiveness(parent, time.Minute) + if liveness != nil || applied != 0 { + t.Fatalf("existing deadline must be honored unchanged, got liveness=%v applied=%s", liveness != nil, applied) + } + if ctx != parent { + t.Fatal("existing-deadline parent must pass through unchanged") + } +} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 73449c543..7f05289d8 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -91,7 +91,7 @@ type StepContext struct { // independent of the session that prescribed the fixes under review - goes // through RunAgent and stays session-isolated. func (sctx *StepContext) RunAgentSession(role SessionRole, opts agent.RunOpts) (*agent.Result, error) { - return sctx.runAgent(sctx.Ctx, opts, role) + return sctx.runAgent(sctx.Ctx, 0, opts, role) } // StepOutcome is the result of executing a pipeline step. diff --git a/internal/pipeline/steps/common_fix.go b/internal/pipeline/steps/common_fix.go index df662f85a..a0d763de4 100644 --- a/internal/pipeline/steps/common_fix.go +++ b/internal/pipeline/steps/common_fix.go @@ -8,6 +8,7 @@ import ( "log/slog" "os" "strings" + "time" "unicode/utf8" "github.com/kunchenguid/no-mistakes/internal/agent" @@ -26,6 +27,12 @@ type fixExecutionOptions struct { FallbackSummary string AfterAgentRun func(*agent.Result) error AgentContext context.Context + // AgentBudget, when positive, is the fix turn's silence budget (per + // invocation, measured from that turn's own activity). Steps with their + // own configured agent budget (review_agent_timeout, test_agent_timeout) + // pass it so the fix turn never inherits another turn's remainder; zero + // keeps the shared default AgentTimeout. + AgentBudget time.Duration // SessionRole, when set, runs the fix turn in that durable review-loop // session (the review step's fixer role). Steps outside the review loop // leave it empty and stay session-isolated. @@ -276,7 +283,13 @@ func executeFixMode(sctx *pipeline.StepContext, stepName types.StepName, opts fi if opts.AgentContext != nil { agentCtx = opts.AgentContext } - result, err := sctx.RunAgentSessionContext(agentCtx, opts.SessionRole, runOpts) + var result *agent.Result + var err error + if opts.AgentBudget > 0 { + result, err = sctx.RunAgentSessionBudget(agentCtx, opts.AgentBudget, opts.SessionRole, runOpts) + } else { + result, err = sctx.RunAgentSessionContext(agentCtx, opts.SessionRole, runOpts) + } if err != nil { return "", fmt.Errorf("%s: %w", opts.ErrorPrefix, err) } diff --git a/internal/pipeline/steps/review.go b/internal/pipeline/steps/review.go index b14304a96..a7154f766 100644 --- a/internal/pipeline/steps/review.go +++ b/internal/pipeline/steps/review.go @@ -1,9 +1,7 @@ package steps import ( - "context" "encoding/json" - "errors" "fmt" "strings" "time" @@ -23,26 +21,11 @@ func (s *ReviewStep) Name() types.StepName { return types.StepReview } func (s *ReviewStep) Execute(sctx *pipeline.StepContext) (*pipeline.StepOutcome, error) { ctx := sctx.Ctx - var cancel context.CancelFunc - var timeout time.Duration - var restoreContext func() - startReviewTimeout := func() { - if cancel != nil { - return - } - parentCtx := sctx.Ctx - ctx, cancel, timeout = reviewAgentContext(sctx) - sctx.Ctx = ctx - restoreContext = func() { - cancel() - sctx.Ctx = parentCtx - } - } - defer func() { - if restoreContext != nil { - restoreContext() - } - }() + // Every agent turn of this step - the optional fix turn and the rereview + // turn - gets its own fresh silence budget of this size. A shared round + // budget let a long healthy fix turn leave its successor only the + // remainder, killing a newly started reviewer minutes into its turn. + budget := reviewAgentBudget(sctx) baseSHA := resolveBranchBaseSHA(ctx, sctx.WorkDir, sctx.Run.BaseSHA, sctx.Repo.DefaultBranch) branch := sctx.Run.Branch ignorePatterns := "none" @@ -83,7 +66,6 @@ func (s *ReviewStep) Execute(sctx *pipeline.StepContext) (*pipeline.StepOutcome, // regression tests guard the wording, not the runtime. var fixSummary string if sctx.Fixing && !sctx.SkipFixExecution { - startReviewTimeout() previousFindings := sanitizedPreviousFindingsForPrompt(sctx.PreviousFindings) historySection := executionContextPromptSection() + roundHistoryPromptSection(sctx) + userIntentPromptSection(sctx) + testguidance.Rule fixPrompt := fmt.Sprintf( @@ -133,9 +115,10 @@ Previous review findings to address: SessionRole: pipeline.SessionRoleFixer, Purpose: "review-fix", Workload: workload, + AgentBudget: budget, }) if err != nil { - return nil, reviewAgentError(ctx, timeout, "agent fix", err) + return nil, reviewAgentError(err, "agent fix") } fixSummary = summary } @@ -172,7 +155,6 @@ Previous review findings to address: // Ask agent to review sctx.Log("reviewing changes...") - startReviewTimeout() // The review turn (initial and every post-fix rereview) carries the intent // conformance obligation: when the intent is authoritative acceptance @@ -275,7 +257,7 @@ Risk assessment (after listing all findings): // cross-round context a rereview legitimately needs travels in the // explicit sanitized round-history section above; only the fixer keeps a // durable session (executeFixMode), because it certifies nothing. - result, err := sctx.RunAgentContext(ctx, agent.RunOpts{ + result, err := sctx.RunAgentBudget(ctx, budget, agent.RunOpts{ Prompt: prompt, CWD: sctx.WorkDir, Env: sctx.Env, @@ -285,7 +267,7 @@ Risk assessment (after listing all findings): Workload: workload, }) if err != nil { - return nil, reviewAgentError(ctx, timeout, "agent review", err) + return nil, reviewAgentError(err, "agent review") } // Parse structured findings @@ -403,20 +385,23 @@ func sanitizePromptMultilineText(text string) string { return strings.TrimSpace(strings.Join(lines, "\n")) } -func reviewAgentContext(sctx *pipeline.StepContext) (context.Context, context.CancelFunc, time.Duration) { +// reviewAgentBudget resolves the Review step's per-turn silence budget: a +// positive review_agent_timeout wins, otherwise the default. +func reviewAgentBudget(sctx *pipeline.StepContext) time.Duration { timeout := config.DefaultReviewAgentTimeout if sctx != nil && sctx.Config != nil && sctx.Config.ReviewAgentTimeout > 0 { timeout = sctx.Config.ReviewAgentTimeout } - ctx, cancel := context.WithTimeoutCause(sctx.Ctx, timeout, errReviewAgentTimeout) - return ctx, cancel, timeout + return timeout } -var errReviewAgentTimeout = errors.New("review agent timeout") - -func reviewAgentError(ctx context.Context, timeout time.Duration, prefix string, err error) error { - if timeout > 0 && errors.Is(context.Cause(ctx), errReviewAgentTimeout) { - return fmt.Errorf("%s timed out after %s (review agent silent for %s): %w", prefix, timeout, timeout, err) +// reviewAgentError renders a review-turn failure, re-labeling the shared +// silence watchdog's timeout with the review-specific diagnostic (preserving +// its budget and liveness evidence) so operators can tell a genuinely quiet +// review agent from other failures. +func reviewAgentError(err error, prefix string) error { + if ate := pipeline.AsAgentTimeout(err); ate != nil { + return ate.StepError(prefix, "review agent") } return fmt.Errorf("%s: %w", prefix, err) } diff --git a/internal/pipeline/steps/review_test.go b/internal/pipeline/steps/review_test.go index ad10ac2b2..bb1ac331b 100644 --- a/internal/pipeline/steps/review_test.go +++ b/internal/pipeline/steps/review_test.go @@ -51,75 +51,69 @@ func TestReviewStep_HangingAgentFailsRunAfterTimeout(t *testing.T) { } } -// TestReviewStep_EachRoundGetsItsOwnAgentBudget pins the documented -// review_agent_timeout contract: the deadline bounds ONE review round - -// its optional fix turn plus the rereview turn share a single budget - and -// every later auto-fix round is derived fresh from the step's parent context. -// Without the fresh derivation, a step context reused across rounds would -// carry round 1's already-spent deadline into round 2 and fail a healthy agent. -func TestReviewStep_EachRoundGetsItsOwnAgentBudget(t *testing.T) { +// TestReviewStep_EachAgentTurnGetsItsOwnFreshBudget pins the +// review_agent_timeout contract: the budget is a per-turn silence budget, so +// a long fix turn cannot leave the rereview turn only the remainder. The +// pre-fix behavior installed one budget at the fix turn's start and killed +// the rereview minutes into its own turn (a real incident: the reviewer got +// ~10 minutes of a 30m budget). Here the fix turn quietly consumes two +// thirds of the budget and returns; the rereview must still get the full +// budget measured from its own start before the silence watchdog kills it. +func TestReviewStep_EachAgentTurnGetsItsOwnFreshBudget(t *testing.T) { dir, baseSHA, headSHA := setupGitRepo(t) gitCmd(t, dir, "checkout", "--detach", headSHA) - const timeout = time.Hour - type call struct { - fixTurn bool - deadline time.Time - } - var calls []call - + const budget = 300 * time.Millisecond findings := `{"findings":[{"file":"a.txt","line":1,"severity":"warning","action":"auto-fix","description":"tidy"}]}` + + var rereviewStartedAt time.Time + var rereviewSilentFor time.Duration ag := &mockAgent{ name: "budget-probe", runFn: func(ctx context.Context, opts agent.RunOpts) (*agent.Result, error) { - dl, ok := ctx.Deadline() - if !ok { - t.Errorf("agent call %d ran with no deadline", len(calls)+1) - } isFix := strings.Contains(opts.Prompt, "Investigate previous review findings") - calls = append(calls, call{fixTurn: isFix, deadline: dl}) - if isFix { - return &agent.Result{Output: json.RawMessage("fixed it")}, nil - } - // Round 1 raises an auto-fixable finding; later rounds are clean. - if len(calls) == 1 { + switch { + case isFix: + // Quiet wall-time consumption below the budget: allowed, but it + // must not shrink the next turn's budget. + time.Sleep(2 * budget / 3) + return &agent.Result{Output: json.RawMessage(`{"summary":"fixed it"}`)}, nil + case strings.Contains(opts.Prompt, "Review the code changes"): + rereviewStartedAt = time.Now() + <-ctx.Done() + rereviewSilentFor = time.Since(rereviewStartedAt) + return nil, ctx.Err() + default: return &agent.Result{Output: json.RawMessage(findings)}, nil } - return &agent.Result{Output: json.RawMessage(`{"findings":[]}`)}, nil }, } sctx := newTestContextWithDBRecords(t, ag, dir, baseSHA, headSHA, config.Commands{}) - sctx.Config.ReviewAgentTimeout = timeout + sctx.Config.ReviewAgentTimeout = budget sctx.Config.AutoFix.Review = 1 exec := pipeline.NewExecutor(sctx.DB, paths.WithRoot(t.TempDir()), sctx.Config, ag, []pipeline.Step{&ReviewStep{}}, nil) - if err := exec.Execute(context.Background(), sctx.Run, sctx.Repo, dir); err != nil { - t.Fatalf("execute: %v", err) + if err := exec.Execute(context.Background(), sctx.Run, sctx.Repo, dir); err == nil { + t.Fatal("expected the quiet rereview turn to hit its silence budget") } - // round 1: review. round 2: fix + rereview. - if len(calls) != 3 { - t.Fatalf("agent calls = %d, want 3 (review, fix, rereview); got %+v", len(calls), calls) - } - if calls[0].fixTurn || !calls[1].fixTurn || calls[2].fixTurn { - t.Fatalf("turn order = %+v, want review, fix, rereview", calls) + // The rereview was killed by its own full silence budget, not by the + // remainder after the fix turn (which would have been ~budget/3). + if rereviewSilentFor < 4*budget/5 { + t.Fatalf("rereview survived only %s of its %s silence budget; it inherited the fix turn's spent time", rereviewSilentFor, budget) } - // The fix turn and the rereview turn of round 2 share one round budget. - if !calls[1].deadline.Equal(calls[2].deadline) { - t.Errorf("round 2 fix and rereview deadlines differ (%v vs %v); one round must share one budget", - calls[1].deadline, calls[2].deadline) - } - // Round 2 is derived fresh, so its budget starts after round 1's. - if !calls[1].deadline.After(calls[0].deadline) { - t.Errorf("round 2 deadline %v is not later than round 1 deadline %v; the round budget leaked across rounds", - calls[1].deadline, calls[0].deadline) + run, err := sctx.DB.GetRun(sctx.Run.ID) + if err != nil { + t.Fatalf("get run: %v", err) } - // Each round's budget is the configured timeout, not a shrinking remainder. - if remaining := time.Until(calls[2].deadline); remaining <= timeout/2 { - t.Errorf("round 2 budget remaining %v is far below the configured %v; the round did not get a full budget", - remaining, timeout) + if run.Error == nil || !strings.Contains(*run.Error, "review agent silent for 300ms") { + var got string + if run.Error != nil { + got = *run.Error + } + t.Fatalf("run error = %q, want review silence diagnostic naming the per-turn budget", got) } } diff --git a/internal/pipeline/steps/test.go b/internal/pipeline/steps/test.go index 3a7a22792..978d0aea5 100644 --- a/internal/pipeline/steps/test.go +++ b/internal/pipeline/steps/test.go @@ -1,9 +1,7 @@ package steps import ( - "context" "encoding/json" - "errors" "fmt" "os" "time" @@ -76,21 +74,19 @@ Rules: Previous test findings to address: ` + sanitizedPreviousFindingsForPrompt(sctx.PreviousFindings) } - fixCtx, cancelFix, fixTimeout := testAgentContext(sctx) summary, err := executeFixMode(sctx, s.Name(), fixExecutionOptions{ LogMessage: "asking agent to fix test failures...", Prompt: fixPrompt, ErrorPrefix: "agent fix tests", FallbackSummary: "fix test failures", - AgentContext: fixCtx, + AgentBudget: testAgentBudget(sctx), AfterAgentRun: func(*agent.Result) error { newTestsFromFix = detectNewTestFiles(ctx, sctx.WorkDir) return nil }, }) - cancelFix() if err != nil { - return nil, testAgentError(fixCtx, fixTimeout, "agent fix tests", err) + return nil, testAgentError(err, "agent fix tests") } fixSummary = summary } @@ -150,8 +146,7 @@ Previous test findings to address: if testCmd != "" { configuredTestCommand = fmt.Sprintf("\nConfigured test command already ran successfully as baseline: `%s`\n", testCmd) } - evidenceCtx, cancelEvidence, evidenceTimeout := testAgentContext(sctx) - result, err := sctx.RunAgentContext(evidenceCtx, agent.RunOpts{ + result, err := sctx.RunAgentBudget(ctx, testAgentBudget(sctx), agent.RunOpts{ Prompt: fmt.Sprintf( `You are validating a code change by testing it. Examine the repository and run the smallest relevant tests yourself. @@ -208,10 +203,8 @@ Rules: JSONSchema: testFindingsSchema, OnChunk: sctx.LogChunk, }) - runErr := testAgentError(evidenceCtx, evidenceTimeout, "agent run tests", err) - cancelEvidence() - if runErr != nil { - return nil, runErr + if err != nil { + return nil, testAgentError(err, "agent run tests") } var findings Findings @@ -279,23 +272,24 @@ Rules: return &pipeline.StepOutcome{Findings: string(findingsJSON), FixSummary: fixSummary}, nil } -func testAgentContext(sctx *pipeline.StepContext) (context.Context, context.CancelFunc, time.Duration) { +// testAgentBudget resolves the Test step's per-invocation silence budget: a +// positive test_agent_timeout wins, otherwise the default. Each agent turn +// (evidence gathering, Test repair) gets its own full budget measured from +// that turn's own activity. +func testAgentBudget(sctx *pipeline.StepContext) time.Duration { timeout := config.DefaultTestAgentTimeout if sctx != nil && sctx.Config != nil && sctx.Config.TestAgentTimeout > 0 { timeout = sctx.Config.TestAgentTimeout } - ctx, cancel := context.WithTimeoutCause(sctx.Ctx, timeout, errTestAgentTimeout) - return ctx, cancel, timeout + return timeout } -var errTestAgentTimeout = errors.New("test agent timeout") - -func testAgentError(ctx context.Context, timeout time.Duration, prefix string, err error) error { - if timeout > 0 && errors.Is(context.Cause(ctx), errTestAgentTimeout) { - return fmt.Errorf("%s timed out after %s (test agent silent for %s): %w", prefix, timeout, timeout, context.Cause(ctx)) - } - if err != nil { - return fmt.Errorf("%s: %w", prefix, err) +// testAgentError renders a test-turn failure, re-labeling the shared silence +// watchdog's timeout with the test-specific diagnostic (preserving its +// budget and liveness evidence). +func testAgentError(err error, prefix string) error { + if ate := pipeline.AsAgentTimeout(err); ate != nil { + return ate.StepError(prefix, "test agent") } - return nil + return fmt.Errorf("%s: %w", prefix, err) } diff --git a/internal/pipeline/steps/test_test.go b/internal/pipeline/steps/test_test.go index cfb52dc76..70cef9fd5 100644 --- a/internal/pipeline/steps/test_test.go +++ b/internal/pipeline/steps/test_test.go @@ -53,11 +53,14 @@ func TestTestStep_HangingEvidenceAgentFailsRunAfterTimeout(t *testing.T) { func TestTestStep_EvidenceAgentCallIsDeadlineBounded(t *testing.T) { t.Parallel() dir, baseSHA, headSHA := setupGitRepo(t) - var sawDeadline bool + var sawLivenessSink bool ag := &mockAgent{ name: "test", - runFn: func(ctx context.Context, _ agent.RunOpts) (*agent.Result, error) { - _, sawDeadline = ctx.Deadline() + runFn: func(ctx context.Context, opts agent.RunOpts) (*agent.Result, error) { + // The evidence turn is bounded by the per-invocation silence + // watchdog (test_agent_timeout), which hands the adapter an + // activity sink rather than installing a fixed deadline. + sawLivenessSink = opts.OnActivity != nil return &agent.Result{Output: json.RawMessage(`{"findings":[],"summary":"","tested":["ok"],"testing_summary":"ok"}`)}, nil }, } @@ -66,8 +69,8 @@ func TestTestStep_EvidenceAgentCallIsDeadlineBounded(t *testing.T) { if err != nil { t.Fatal(err) } - if !sawDeadline { - t.Fatal("evidence agent ran without a deadline") + if !sawLivenessSink { + t.Fatal("evidence agent ran without a liveness sink") } if outcome == nil || outcome.NeedsApproval { t.Fatalf("successful evidence gathering should still complete, got %+v", outcome) From 2e9a8019f100f9cbcc74504c163e35a36f26b559 Mon Sep 17 00:00:00 2001 From: 010351 Date: Wed, 26 Aug 2026 08:41:23 +0800 Subject: [PATCH 2/2] fix(agent): order liveness evidence deterministically at clock ties evidence() sorted per-kind activity with an unstable sort keyed only on the observation timestamp, so two kinds stamped within one clock tick (the Windows CI timer granularity) rendered in Go map iteration order: the "most recent first" diagnostic contract was nondeterministic at exact ties, which is what failed TestLivenessEvidence_NamesEveryKindSeen on the windows-core leg with both ages rendering as 0s. Fold a per-invocation record sequence into the sort: most recent first, exact clock ties break in record order. Records are mutex-serialized, so the later record is the genuinely more recent activity and the ordering is a deterministic total order on every platform. Names, ages, and the no-content/no-paths diagnostic contract are unchanged, as are the watchdog budget, pi session binding, process-tree termination, per-invocation clock reset, and conservative fallback. Regression: TestLivenessEvidence_OrdersByRecencyAndBreaksClockTiesInRecordOrder constructs an exact tie through the recordAt seam (a coarse timer can no longer make the case vacuous) alongside a real ordering difference, and TestLivenessEvidence_NamesEveryKindSeen is now stable on every platform. --- internal/pipeline/liveness.go | 46 ++++++++++++++++++++++++------ internal/pipeline/liveness_test.go | 28 ++++++++++++++++++ 2 files changed, 65 insertions(+), 9 deletions(-) diff --git a/internal/pipeline/liveness.go b/internal/pipeline/liveness.go index 013693f95..d1e0f9ff3 100644 --- a/internal/pipeline/liveness.go +++ b/internal/pipeline/liveness.go @@ -29,28 +29,50 @@ import ( type invocationLiveness struct { mu sync.Mutex last time.Time - seen map[agent.ActivityKind]time.Time + seq uint64 + seen map[agent.ActivityKind]activityMark +} + +// activityMark is one kind's latest observation: when it happened and its +// place in this invocation's record sequence. The sequence makes the +// evidence ordering a total order even when two kinds share one clock +// reading, which a coarse platform timer can produce for back-to-back +// records. +type activityMark struct { + at time.Time + seq uint64 } func newInvocationLiveness() *invocationLiveness { return &invocationLiveness{ last: time.Now(), - seen: make(map[agent.ActivityKind]time.Time), + seen: make(map[agent.ActivityKind]activityMark), } } func (l *invocationLiveness) record(kind agent.ActivityKind) { + l.recordAt(kind, time.Now()) +} + +// recordAt is the single activity-mutation path: record passes the current +// time, and tests pass explicit instants to construct the exact clock ties a +// coarse platform timer can produce implicitly. +func (l *invocationLiveness) recordAt(kind agent.ActivityKind, at time.Time) { l.mu.Lock() defer l.mu.Unlock() - now := time.Now() - l.last = now - l.seen[kind] = now + l.seq++ + if at.After(l.last) { + l.last = at + } + l.seen[kind] = activityMark{at: at, seq: l.seq} } // evidence renders the per-kind activity summary for the timeout diagnostic. // It names evidence classes and ages only - never content, paths, or session // payloads - so operators can distinguish stdout-active, session-active, and -// genuinely quiet invocations. +// genuinely quiet invocations. Kinds render most recent first; an exact +// clock tie breaks in record order (the later record is the genuinely more +// recent activity), so the ordering is deterministic on every platform. func (l *invocationLiveness) evidence() string { l.mu.Lock() defer l.mu.Unlock() @@ -61,13 +83,19 @@ func (l *invocationLiveness) evidence() string { label string age time.Duration at time.Time + seq uint64 } now := time.Now() parts := make([]kindAge, 0, len(l.seen)) - for kind, at := range l.seen { - parts = append(parts, kindAge{label: agent.ActivityKindLabel(kind), age: now.Sub(at).Round(time.Millisecond), at: at}) + for kind, mark := range l.seen { + parts = append(parts, kindAge{label: agent.ActivityKindLabel(kind), age: now.Sub(mark.at).Round(time.Millisecond), at: mark.at, seq: mark.seq}) } - sort.Slice(parts, func(i, j int) bool { return parts[i].at.After(parts[j].at) }) + sort.Slice(parts, func(i, j int) bool { + if !parts[i].at.Equal(parts[j].at) { + return parts[i].at.After(parts[j].at) + } + return parts[i].seq > parts[j].seq + }) rendered := make([]string, 0, len(parts)) for _, part := range parts { rendered = append(rendered, fmt.Sprintf("%s %s ago", part.label, part.age)) diff --git a/internal/pipeline/liveness_test.go b/internal/pipeline/liveness_test.go index 8af0a6561..485f3b8f3 100644 --- a/internal/pipeline/liveness_test.go +++ b/internal/pipeline/liveness_test.go @@ -103,6 +103,34 @@ func TestLivenessEvidence_NamesEveryKindSeen(t *testing.T) { } } +func TestLivenessEvidence_OrdersByRecencyAndBreaksClockTiesInRecordOrder(t *testing.T) { + t.Parallel() + base := time.Now().Add(-time.Minute) + l := newInvocationLiveness() + l.recordAt(agent.ActivityStdout, base) + l.recordAt(agent.ActivityLifecycle, base.Add(2*time.Second)) + // A coarse platform timer (Windows CI) can stamp two back-to-back records + // with one clock reading; the later record is the genuinely more recent + // activity, and the tie must resolve the same way on every render. + l.recordAt(agent.ActivitySession, base.Add(2*time.Second)) + + assertOrder := func(evidence string) { + t.Helper() + sessionIdx := strings.Index(evidence, "pi session events") + lifecycleIdx := strings.Index(evidence, "process lifecycle") + stdoutIdx := strings.Index(evidence, "stdout bytes") + if sessionIdx < 0 || lifecycleIdx < 0 || stdoutIdx < 0 { + t.Fatalf("evidence = %q, want every recorded kind named", evidence) + } + if !(sessionIdx < lifecycleIdx && lifecycleIdx < stdoutIdx) { + t.Fatalf("evidence = %q, want the tied later record first and the genuinely older kind last", evidence) + } + } + assertOrder(l.evidence()) + // The ordering is a deterministic total order: repeated renders agree. + assertOrder(l.evidence()) +} + func TestAgentTimeoutError_MessageAndStepLabel(t *testing.T) { t.Parallel() ate := &AgentTimeoutError{Budget: 30 * time.Minute, Evidence: "last activity: pi session events 30m0s ago"}