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..d1e0f9ff3 --- /dev/null +++ b/internal/pipeline/liveness.go @@ -0,0 +1,196 @@ +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 + 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]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() + 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. 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() + 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 + seq uint64 + } + now := time.Now() + parts := make([]kindAge, 0, len(l.seen)) + 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 { + 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)) + } + 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..485f3b8f3 --- /dev/null +++ b/internal/pipeline/liveness_test.go @@ -0,0 +1,189 @@ +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 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"} + 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)