diff --git a/internal/pipeline/implstage.go b/internal/pipeline/implstage.go index f356e6d2..5dd47ad0 100644 --- a/internal/pipeline/implstage.go +++ b/internal/pipeline/implstage.go @@ -411,13 +411,40 @@ func impRunRetryLoop(env impRetryEnv, plan ImplementationPlan, cfg StageConfig, KillStaleProcessTree(s.Pid) } }() - log.Info("implement", fmt.Sprintf("Attempt %d finished", displayAttempt), []ledger.KV{ + finished := []ledger.KV{ {Key: "exitCode", Value: result.ExitCode}, {Key: "timedOut", Value: result.TimedOut}, {Key: "durationMs", Value: result.DurationMs}, {Key: "events", Value: len(result.Events)}, - }) + } + if result.TimedOut { + // Issue #248: the kill cause belongs on the row — a + // productive wall-kill and a zero-event burn must not share + // an indistinguishable "timedOut true, events 0" line. + finished = append(finished, + ledger.KV{Key: "watchdogFired", Value: result.WatchdogFired}, + ledger.KV{Key: "coldStart", Value: result.ColdStart}, + ledger.KV{Key: "clockResets", Value: result.ClockResets}, + ledger.KV{Key: "meaningfulBytes", Value: result.MeaningfulBytes}, + ) + } + log.Info("implement", fmt.Sprintf("Attempt %d finished", displayAttempt), finished) if result.TimedOut || result.ExitCode != 0 { + if impIsProductiveWallKill(result) { + // Own classification (issue #248): the attempt streamed + // real work before the wall killed it. Retry semantics + // stay identical to the transient path — backoff, then + // the next dispatch continues the session via -c — but + // the row says what actually happened. + infraRetries++ + data := []ledger.KV{ + {Key: "clockResets", Value: result.ClockResets}, + {Key: "meaningfulBytes", Value: result.MeaningfulBytes}, + } + log.Warn("implement", fmt.Sprintf("wall-timeout (productive): attempt %d streamed %d clock resets / %d meaningful bytes before the kill; retrying with resume (infra retry %d)", displayAttempt, result.ClockResets, result.MeaningfulBytes, infraRetries), data) + time.Sleep(time.Duration(backoff(infraRetries)) * time.Millisecond) + continue + } if impIsInfraTransient(result) { infraRetries++ func() { @@ -482,10 +509,44 @@ func impRunRetryLoop(env impRetryEnv, plan ImplementationPlan, cfg StageConfig, return ImplementResult{OK: false, Worker: env.workerName, Attempts: attempts, WorktreePath: env.worktreePath, FailureClass: lastFailureClass}, false, nil } -// impIsInfraTransient mirrors the isInfraTransient closure (Q31): a cold-start -// kill is transient infra alongside the watchdog timeout — a wedged CLI init -// is not the worker's fault. +// impMeaningfulBytesThreshold: a wall-killed attempt counts as +// productive when the stream classifier saw meaningful output above this +// byte floor (Q33), or at least one classified clock reset. The +// "did the worker actually do anything" bar from the issue #248 +// evidence run (a 72-minute productive attempt carried tool activity +// and hundreds of KB of meaningful bytes; a wedged provider burn +// streams thinking-only or nothing at all). +const impMeaningfulBytesThreshold = 1024 + +// impIsProductiveWallKill reports whether a wall/watchdog kill landed on +// an attempt that demonstrably made progress (issue #248 required +// change 3): meaningful stream bytes above threshold, or at least one +// adapter-classified clock reset. Such a kill is NOT "transient infra" +// — it is its own outcome, retried via the -c resume path but classified +// and logged as what happened. +func impIsProductiveWallKill(r workers.WorkerResult) bool { + if !r.TimedOut { + return false + } + if r.ColdStart { + return false + } + if r.ClockResets > 0 || r.MeaningfulBytes >= impMeaningfulBytesThreshold { + return true + } + return false +} + +// impIsInfraTransient mirrors the isInfraTransient closure (Q31): a +// cold-start kill is transient infra alongside a zero-progress watchdog +// timeout — a wedged CLI init is not the worker's fault. A productive +// wall-kill (progress present before the kill) is deliberately NOT +// transient: it surfaces as its own "wall-timeout (productive)" row and +// still retries through the same resume loop. func impIsInfraTransient(r workers.WorkerResult) bool { + if r.TimedOut && impIsProductiveWallKill(r) { + return false + } if r.TimedOut || r.ColdStart { return true } diff --git a/internal/pipeline/implstage_test.go b/internal/pipeline/implstage_test.go index ee093269..c8b407bd 100644 --- a/internal/pipeline/implstage_test.go +++ b/internal/pipeline/implstage_test.go @@ -755,3 +755,79 @@ func TestImpDropOrcaWorkspaceIfRequestedNoop(t *testing.T) { t.Fatalf("plain repo must not drop anything: %+v", log.Entries) } } + +// Issue #248 required change 3: a wall kill on an attempt that streamed +// real work must NOT be classified "Transient infra failure" — it gets +// its own wall-timeout (productive) row, still retries with the same +// backoff, and does not consume the logic-attempt budget. +func TestImpRetryLoopProductiveWallKillIsNotInfraTransient(t *testing.T) { + repo := t.TempDir() + log := &impCaptureLog{} + spawns := 0 + env := impBaseRetryEnv(repo) + env.spawn = func(workers.WorkerSpawnOptions) workers.WorkerResult { + spawns++ + if spawns == 1 { + // The loop-176 attempt-1 signature: killed at the wall, but + // the stream classifier saw real work before the kill. + return workers.WorkerResult{ + ExitCode: 124, + TimedOut: true, + DurationMs: 4_316_000, + ClockResets: 112, + MeaningfulBytes: 631_882, + } + } + return workers.WorkerResult{ExitCode: 0, ResultText: "recovered"} + } + env.testGate = func(string, int) (gates.GateResult, error) { + return gates.GateResult{Passed: true}, nil + } + env.backoff = func(int) int { return 0 } + res, succeeded, err := impRunRetryLoop(env, impTestPlan(), StageConfig{TimeoutMs: 1000}, log) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !succeeded || !res.OK || res.Attempts != 1 { + t.Fatalf("productive wall-kill retries must not consume the logic budget: %+v", res) + } + if spawns != 2 { + t.Fatalf("expected 2 spawns, got %d", spawns) + } + if !log.has("warn", "wall-timeout (productive)") { + t.Fatalf("missing productive wall-kill row: %+v", log.Entries) + } + if log.has("warn", "Transient infra failure") { + t.Fatalf("productive wall-kill must not be classified infra-transient: %+v", log.Entries) + } + // The attempt-finished row must carry the kill evidence. + if v, ok := log.kvOf("info", "Attempt 1 finished", "clockResets"); !ok || v != 112 { + t.Fatalf("missing clockResets KV on attempt row: %v %v", v, ok) + } + // No transient class may be recorded for a productive wall-kill. + if _, err := os.Stat(impProxyStatePath(repo)); !os.IsNotExist(err) { + t.Fatalf("productive wall-kill must not write proxy-state.json: %v", err) + } +} + +// A zero-progress wall kill (no meaningful bytes, no clock resets) stays +// transient infra — the wedged-provider burn class must keep its +// existing classification. Cold-start kills stay transient regardless of +// stream bytes: nothing productive ever started. +func TestImpIsInfraTransient_WallKillClasses(t *testing.T) { + productive := workers.WorkerResult{TimedOut: true, ClockResets: 3, MeaningfulBytes: 4096} + if impIsInfraTransient(productive) { + t.Fatalf("productive wall-kill must not be infra-transient") + } + burn := workers.WorkerResult{TimedOut: true} + if !impIsInfraTransient(burn) { + t.Fatalf("zero-progress wall-kill must stay infra-transient") + } + coldStart := workers.WorkerResult{TimedOut: true, ColdStart: true, ClockResets: 0, MeaningfulBytes: 99999} + if !impIsInfraTransient(coldStart) { + t.Fatalf("cold-start kill must stay infra-transient even with bytes") + } + if impIsProductiveWallKill(workers.WorkerResult{TimedOut: false, ClockResets: 9}) { + t.Fatalf("a non-timed-out result is never a wall kill") + } +} diff --git a/internal/workers/adapter_common.go b/internal/workers/adapter_common.go index ce0edbbc..8043dd80 100644 --- a/internal/workers/adapter_common.go +++ b/internal/workers/adapter_common.go @@ -99,6 +99,19 @@ func derefSpawn(r *SpawnCliResult) SpawnCliResult { return SpawnCliResult{} } +// stampStreamMetrics copies the launch's Q34/Q33 stream evidence onto the +// finalized WorkerResult (issue #248 required change 3). The metrics of +// the FINAL attempt are the classification signal: a wall kill on an +// attempt that streamed work is a productive wall-kill; a no-progress +// watchdog firing on the last attempt is a burn regardless of earlier +// attempts. +func stampStreamMetrics(res WorkerResult, run SpawnCliResult) WorkerResult { + res.WatchdogFired = run.WatchdogFired + res.ClockResets = run.ClockResets + res.MeaningfulBytes = run.MeaningfulBytes + return res +} + // parseJSONAny parses s into any (TS JSON.parse); err swallowed by caller. func parseJSONAny(s string) (v any, ok bool) { if err := json.Unmarshal([]byte(s), &v); err != nil { diff --git a/internal/workers/claudecode.go b/internal/workers/claudecode.go index afe816e1..45317289 100644 --- a/internal/workers/claudecode.go +++ b/internal/workers/claudecode.go @@ -77,7 +77,7 @@ func (a *ClaudeCodeAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { } spawnOpts := SpawnCliOptions{ Dir: opts.Cwd, - TimeoutMs: opts.TimeoutMs, + TimeoutMs: launchBudgetMs(wallDeadline, a.nowMs(), int64(opts.TimeoutMs)), Env: opts.Env, NoProgressTimeoutMs: intPtrIf(noProgressTimeoutMs != 0, noProgressTimeoutMs), ColdStartTimeoutMs: opts.ColdStartTimeoutMs, @@ -153,7 +153,7 @@ func (a *ClaudeCodeAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { } } - return claudeFinalize(derefSpawn(last), a.nowMs()-start) + return stampStreamMetrics(claudeFinalize(derefSpawn(last), a.nowMs()-start), derefSpawn(last)) } // claudeBaseArgs mirrors the TS baseArgs. claude-code ignores variant; diff --git a/internal/workers/grok.go b/internal/workers/grok.go index d271c51f..bcc12466 100644 --- a/internal/workers/grok.go +++ b/internal/workers/grok.go @@ -528,7 +528,7 @@ func (a *GrokAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { } prepared, err := a.prepare("grok", args, SpawnCliOptions{ Dir: opts.Cwd, - TimeoutMs: opts.TimeoutMs, + TimeoutMs: launchBudgetMs(wallDeadline, a.nowMs(), int64(opts.TimeoutMs)), Env: spawnEnv, NoProgressTimeoutMs: &noProgressTimeoutMs, WatchdogLedger: opts.WatchdogLedger, @@ -575,7 +575,7 @@ func (a *GrokAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { fallback := grokFallbackEmpty() last = &fallback } - result := grokFinalize(*last, sessionId, a.nowMs()-start) + result := stampStreamMetrics(grokFinalize(*last, sessionId, a.nowMs()-start), *last) // FR-GROK-03: persist the exact cost onto the run ledger for every grok // worker run the provider priced. Best-effort — a ledger write must // never fail the run. Identity comes from the dispatcher's watchdog diff --git a/internal/workers/omp.go b/internal/workers/omp.go index adbb5de2..09dd9136 100644 --- a/internal/workers/omp.go +++ b/internal/workers/omp.go @@ -86,7 +86,10 @@ func BuildOmpArgs(opts WorkerSpawnOptions, o OmpArgsOptions) []string { return args } -// OmpOutcome mirrors TS OmpOutcome. +// OmpOutcome mirrors TS OmpOutcome. Events collects the NDJSON lifecycle +// records (session/turn_end/message_end/result) parsed from the stream — +// the evidence a success-without-result-envelope (or a wall-killed +// partial stream) actually streamed work. type OmpOutcome struct { IsError bool SessionId string @@ -94,6 +97,7 @@ type OmpOutcome struct { ResultText string Parsed map[string]any TimedOut bool + Events []WorkerEvent } // InterpretOmpForTest is the test seam re-export of the parser. @@ -115,6 +119,7 @@ func interpretOmp(run SpawnCliResult) OmpOutcome { sessionId := "" var terminalMessage map[string]any streamError := "" + var events []WorkerEvent // Try single-JSON first (legacy callers). If it parses as object/array, // skip the NDJSON walk. @@ -173,6 +178,13 @@ func interpretOmp(run SpawnCliResult) OmpOutcome { if !ok { continue } + // The events metric counts the records that evidence the + // session actually streamed (issue #248): session headers, + // turn/message lifecycle and result envelopes. + switch event["type"] { + case "session", "turn_end", "message_end", "result": + events = append(events, event) + } if event["type"] == "session" { if id, ok := event["id"].(string); ok { sessionId = id @@ -242,6 +254,7 @@ func interpretOmp(run SpawnCliResult) OmpOutcome { ResultText: resultText, Parsed: parsed, TimedOut: run.TimedOut, + Events: events, } } @@ -255,18 +268,28 @@ func ompFallbackEmpty() SpawnCliResult { } } -// ompFinalize mirrors TS finalize. +// ompFinalize maps the last spawn result to WorkerResult (issue #248 +// required change 2): the events metric is populated for every outcome +// class — a success without a result envelope synthesizes its events +// from the NDJSON lifecycle records, and a wall/watchdog kill reports +// whatever the partial stream delivered (0 for a truly silent kill, +// > 0 when the session had streamed before the kill). TimedOut keeps +// ResultText empty — the turn was never confirmed complete — but the +// events count stays honest either way. func ompFinalize(run SpawnCliResult, sessionId string, durationMs int64) WorkerResult { outcome := interpretOmp(run) if run.TimedOut { result := WorkerResult{ - ExitCode: run.ExitCode, - Events: []WorkerEvent{}, - ResultText: "", - SessionId: sessionId, - DurationMs: durationMs, - TimedOut: true, - ErrorText: strings.TrimSpace(run.Stderr), + ExitCode: run.ExitCode, + Events: outcome.Events, + ResultText: "", + SessionId: sessionId, + DurationMs: durationMs, + TimedOut: true, + ErrorText: strings.TrimSpace(run.Stderr), + WatchdogFired: run.WatchdogFired, + ClockResets: run.ClockResets, + MeaningfulBytes: run.MeaningfulBytes, } if run.ColdStart { result.ColdStart = true @@ -274,19 +297,27 @@ func ompFinalize(run SpawnCliResult, sessionId string, durationMs int64) WorkerR return result } var events []WorkerEvent - if outcome.Parsed != nil { + switch { + case outcome.Parsed != nil: events = []WorkerEvent{eventFromResult(outcome.Parsed)} - } else { + case len(outcome.Events) > 0: + // NDJSON success without a result envelope (the real omp shape): + // the lifecycle records ARE the events evidence. + events = outcome.Events + default: events = []WorkerEvent{} } return WorkerResult{ - ExitCode: run.ExitCode, - Events: events, - ResultText: outcome.ResultText, - SessionId: sessionId, - DurationMs: durationMs, - TimedOut: false, - ErrorText: outcome.ErrorText, + ExitCode: run.ExitCode, + Events: events, + ResultText: outcome.ResultText, + SessionId: sessionId, + DurationMs: durationMs, + TimedOut: false, + ErrorText: outcome.ErrorText, + WatchdogFired: run.WatchdogFired, + ClockResets: run.ClockResets, + MeaningfulBytes: run.MeaningfulBytes, } } @@ -321,6 +352,10 @@ func (a *OmpAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { args := BuildOmpArgs(opts, OmpArgsOptions{}) sessionId := "" var last *SpawnCliResult + // Issue #248: coalesce the per-launch stream evidence across retries + // so the final WorkerResult reports what the whole spawn produced. + var totalClockResets, totalMeaningfulBytes int + var sawWatchdogFired bool // omp exposes -r/--resume , but using it requires carrying the // session_id between attempts. The adapter parses sessionId from the // prior attempt's output but does not feed it to -r because that would @@ -338,7 +373,7 @@ func (a *OmpAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { } prepared, err := a.prepare("omp", args, SpawnCliOptions{ Dir: opts.Cwd, - TimeoutMs: opts.TimeoutMs, + TimeoutMs: launchBudgetMs(wallDeadline, a.nowMs(), int64(opts.TimeoutMs)), Env: opts.Env, NoProgressTimeoutMs: &noProgressTimeoutMs, ColdStartTimeoutMs: opts.ColdStartTimeoutMs, @@ -354,6 +389,11 @@ func (a *OmpAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { Herdr: opts.Herdr, }) last = &raw + totalClockResets += raw.ClockResets + totalMeaningfulBytes += raw.MeaningfulBytes + if raw.WatchdogFired { + sawWatchdogFired = true + } outcome := interpretOmp(raw) if outcome.SessionId != "" { @@ -380,5 +420,36 @@ func (a *OmpAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { last = &SpawnCliResult{} *last = ompFallbackEmpty() } + // Issue #248: surface the coalesced stream evidence on the final + // result (the last attempt alone would under-report after retries). + last.ClockResets = totalClockResets + last.MeaningfulBytes = totalMeaningfulBytes + if sawWatchdogFired { + last.WatchdogFired = true + } return ompFinalize(*last, sessionId, a.nowMs()-start) } + +// launchBudgetMs computes the per-launch wall budget for one adapter +// attempt: the caller timeout capped by what remains of the run's +// overall wall deadline (issue #248 finding 4 — the last attempt of a +// retry loop must not restart with a fresh full wall). timeoutMs<=0 or +// an already-spent deadline yields a small floor so the wall clock +// still kills the child promptly instead of arming nothing. +func launchBudgetMs(wallDeadline, now, timeoutMs int64) int { + const minLaunchBudgetMs = 1000 + if timeoutMs <= 0 { + return 0 + } + remaining := wallDeadline - now + if remaining <= 0 { + return minLaunchBudgetMs + } + if remaining > timeoutMs { + return int(timeoutMs) + } + if remaining < minLaunchBudgetMs { + return minLaunchBudgetMs + } + return int(remaining) +} diff --git a/internal/workers/omp_test.go b/internal/workers/omp_test.go index c08a95e9..86216d9d 100644 --- a/internal/workers/omp_test.go +++ b/internal/workers/omp_test.go @@ -160,3 +160,95 @@ func assertArgv(t *testing.T, got, want []string) { } } } + +// Issue #248 required change 2: a successful NDJSON run without a result +// envelope must not report events: 0 — the lifecycle records are the +// evidence the session actually streamed. +func TestOmpFinalize_NdjsonSuccessHasEvents(t *testing.T) { + stdout, err := os.ReadFile(filepath.Join("testdata", "omp-smoke-2026-08-30.jsonl")) + if err != nil { + t.Fatal(err) + } + res := ompFinalize(SpawnCliResult{ExitCode: 0, Stdout: string(stdout)}, "01a05127-5cc0-7680-9853-7dc3c80a1477", 1000) + if res.TimedOut || res.ExitCode != 0 { + t.Fatalf("result = %+v", res) + } + if len(res.Events) <= 0 { + t.Fatalf("success fixture must synthesize events > 0, got %d", len(res.Events)) + } + if res.ResultText != "OK" { + t.Fatalf("resultText = %q, want OK", res.ResultText) + } +} + +// TimedOut semantics (issue #248): the events count reports whatever the +// partial stream delivered before the kill — 0 for a truly silent burn, +// > 0 when the session had streamed. ResultText stays empty: the turn was +// never confirmed complete. +func TestOmpFinalize_TimeoutKeepsPartialStreamEvents(t *testing.T) { + stdout := "" + + `{"type":"session","id":"omp-timeout-1"}` + "\n" + + `{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"half done"}]}}` + "\n" + res := ompFinalize(SpawnCliResult{ExitCode: -1, Stdout: stdout, TimedOut: true}, "omp-timeout-1", 60_000) + if !res.TimedOut { + t.Fatalf("result = %+v", res) + } + if len(res.Events) != 2 { + t.Fatalf("partial stream events = %d, want 2", len(res.Events)) + } + if res.ResultText != "" { + t.Fatalf("timed-out resultText must stay empty, got %q", res.ResultText) + } + // The zero-event burn class: nothing streamed, nothing counted. + burn := ompFinalize(SpawnCliResult{ExitCode: -1, Stdout: "", TimedOut: true}, "", 60_000) + if len(burn.Events) != 0 { + t.Fatalf("silent burn events = %d, want 0", len(burn.Events)) + } +} + +// Issue #248 required change 3: the stream evidence must reach +// WorkerResult so the pipeline classifier can tell a productive wall-kill +// from a watchdog/cold-start kill. +func TestOmpFinalize_ThreadStreamMetrics(t *testing.T) { + run := SpawnCliResult{ + ExitCode: -1, + TimedOut: true, + WatchdogFired: true, + ClockResets: 7, + MeaningfulBytes: 1234, + } + res := ompFinalize(run, "", 1000) + if !res.WatchdogFired || res.ClockResets != 7 || res.MeaningfulBytes != 1234 { + t.Fatalf("metrics not threaded: %+v", res) + } + res = ompFinalize(SpawnCliResult{ExitCode: 0, Stdout: `{"type":"result","result":"ok"}`}, "", 1000) + if res.WatchdogFired || res.ClockResets != 0 || res.MeaningfulBytes != 0 { + t.Fatalf("clean run must carry zero metrics: %+v", res) + } +} + +// Issue #248 finding 4: per-launch walls hold — a late retry inside the +// adapter loop must not restart the full request budget. +func TestLaunchBudgetMs(t *testing.T) { + const full int64 = 60_000 + deadline := int64(100_000) + cases := []struct { + name string + now int64 + want int + }{ + {"first launch keeps the full budget", 40_000, 60_000}, + {"late retry gets the remaining budget", 70_000, 30_000}, + {"at the deadline gets the floor", 100_000, 1000}, + {"past the deadline gets the floor", 120_000, 1000}, + } + for _, tc := range cases { + if got := launchBudgetMs(deadline, tc.now, full); got != tc.want { + t.Errorf("%s: launchBudgetMs = %d, want %d", tc.name, got, tc.want) + } + } + // Unbounded requests stay unbounded (0 arms no wall clock). + if got := launchBudgetMs(1<<62, 0, 0); got != 0 { + t.Errorf("unbounded budget = %d, want 0", got) + } +} diff --git a/internal/workers/opencode.go b/internal/workers/opencode.go index 85c47d27..2f107dec 100644 --- a/internal/workers/opencode.go +++ b/internal/workers/opencode.go @@ -89,7 +89,7 @@ func (a *OpenCodeAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { } prepared, err := a.prepare(binary, args, SpawnCliOptions{ Dir: opts.Cwd, - TimeoutMs: opts.TimeoutMs, + TimeoutMs: launchBudgetMs(wallDeadline, a.nowMs(), int64(opts.TimeoutMs)), Env: opts.Env, NoProgressTimeoutMs: intPtrIf(noProgressTimeoutMs != 0, noProgressTimeoutMs), WatchdogLedger: opts.WatchdogLedger, @@ -111,7 +111,7 @@ func (a *OpenCodeAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { args = opencodeBaseArgs(opts, binary) fallbackPrepared, ferr := a.prepare(binary, args, SpawnCliOptions{ Dir: opts.Cwd, - TimeoutMs: opts.TimeoutMs, + TimeoutMs: launchBudgetMs(wallDeadline, a.nowMs(), int64(opts.TimeoutMs)), Env: opts.Env, NoProgressTimeoutMs: intPtrIf(noProgressTimeoutMs != 0, noProgressTimeoutMs), WatchdogLedger: opts.WatchdogLedger, @@ -219,10 +219,10 @@ func (a *OpenCodeAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { if final.ColdStart { result.ColdStart = true } - return result + return stampStreamMetrics(result, final) } - return WorkerResult{ + return stampStreamMetrics(WorkerResult{ ExitCode: final.ExitCode, Events: lastEvents, ResultText: lastResultText, @@ -232,7 +232,7 @@ func (a *OpenCodeAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { // Zero-event attempts (probe-dead bail or exhausted budget) surface // as noProgress, never as a false success. NoProgress: noProgress, - } + }, final) } // probeEndpointAlive: short-deadline trivial probe run before spending diff --git a/internal/workers/pi.go b/internal/workers/pi.go index 5405f8b2..2ae36b11 100644 --- a/internal/workers/pi.go +++ b/internal/workers/pi.go @@ -339,7 +339,7 @@ func (a *PiAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { prepared, err := a.prepare("pi", args, SpawnCliOptions{ Dir: opts.Cwd, - TimeoutMs: opts.TimeoutMs, + TimeoutMs: launchBudgetMs(wallDeadline, a.nowMs(), int64(opts.TimeoutMs)), Env: opts.Env, NoProgressTimeoutMs: intPtrIf(noProgressTimeoutMs != 0, noProgressTimeoutMs), ColdStartTimeoutMs: opts.ColdStartTimeoutMs, @@ -400,7 +400,7 @@ func (a *PiAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult { args = BuildPiArgs(opts, true) } - return piFinalize(derefSpawn(last), a.nowMs()-start) + return stampStreamMetrics(piFinalize(derefSpawn(last), a.nowMs()-start), derefSpawn(last)) } // TS Infinity proxy. diff --git a/internal/workers/spawnstream.go b/internal/workers/spawnstream.go index d2f0e77c..51fd16c6 100644 --- a/internal/workers/spawnstream.go +++ b/internal/workers/spawnstream.go @@ -18,6 +18,7 @@ import ( "time" "github.com/FreePeak/devagent/internal/config" + "github.com/FreePeak/devagent/internal/ledger" "github.com/FreePeak/devagent/internal/spawn" ) @@ -41,13 +42,72 @@ type SpawnCliOptions struct { // require an armed clock). Nil = no row. WatchdogLedger *WatchdogLedgerContext // WatchdogSink receives the watchdog-health row when a ledger context - // is present and a clock is armed. - // TODO(FR-GO-05 #190): replace with the orchestrator ledger port - // (appendWatchdogHealthRecord) once the ledger package lands. + // is present and a clock is armed. Nil = the default Q34 sink, which + // appends to the repo ledger at WatchdogLedger.RepoPath (the + // FR-GO-05 #190 wiring): every production spawn with an armed clock + // emits the watchdog-health row. Tests may inject a capture sink. WatchdogSink func(WatchdogHealthRecord) + // PostKillDrainWaitMs bounds the post-kill pipe drain: after cmd.Wait + // the stdout/stderr pumpers must EOF, but a grandchild that inherited + // the pipes can hold them open indefinitely and push the attempt's + // finish far past its wall clock (issue #248, +716s overshoot). 0 = + // the default cap; negative = wait unbounded (legacy behavior). + PostKillDrainWaitMs int } -// SpawnCliResult mirrors TS SpawnCliResult. +// DefaultPostKillDrainWaitMs is the default bounded post-kill drain wait +// (issue #248 required change 4): a few seconds — comfortably longer than +// any legitimate flush, short enough that a pipe-holding grandchild +// cannot push the finish past the attempt wall. +const DefaultPostKillDrainWaitMs = 3000 + +// watchdogLedgerSink is the production WatchdogSink (FR-GO-05 #190): +// one best-effort watchdog-health append into the ledger events file of +// repoPath. Mirrors the herdr-pane row writer (internal/herdr/herdr.go) +// — never throws. +func watchdogLedgerSink(repoPath string) func(WatchdogHealthRecord) { + return func(r WatchdogHealthRecord) { + runtime, visible, visibility := r.Runtime, r.Visible, r.Visibility + ledger.AppendWatchdogHealthRecord(repoPath, ledger.WatchdogHealthRecord{ + TS: r.Ts, + Kind: r.Kind, + TaskID: r.TaskId, + Attempt: r.Attempt, + Event: r.Event, + Site: r.Site, + Worker: r.Worker, + NoProgressTimeoutMs: int64(r.NoProgressTimeoutMs), + WatchdogFired: r.WatchdogFired, + ColdStartFired: r.ColdStartFired, + WallClockMs: r.WallClockMs, + ClockResets: r.ClockResets, + MeaningfulBytes: int64(r.MeaningfulBytes), + IdleMs: r.IdleMs, + Runtime: &runtime, + Visible: &visible, + Visibility: &visibility, + }) + } +} + +// watchdogSinkFor resolves the effective sink for one launch: the caller +// sink wins; nil = the default ledger-append sink when a ledger context +// is present. Returns nil when no context is set (rows are pointless +// without identity). +func watchdogSinkFor(opts SpawnCliOptions) func(WatchdogHealthRecord) { + if opts.WatchdogSink != nil { + return opts.WatchdogSink + } + if opts.WatchdogLedger == nil { + return nil + } + return watchdogLedgerSink(opts.WatchdogLedger.RepoPath) +} + +// SpawnCliResult mirrors TS SpawnCliResult. The trailing fields are the +// Q34/Q33 stream-progress evidence the streaming path observed (all zero +// on the unarmed runcli path) so WorkerResult can carry them to the +// pipeline classifier. type SpawnCliResult struct { ExitCode int Stdout string @@ -56,6 +116,13 @@ type SpawnCliResult struct { // ColdStart: true only when the cold-start (first-progress) deadline // killed the launch. ColdStart bool + // WatchdogFired: true only when the no-progress clock killed the + // launch (the wall-clock expiry leaves it false — herdr parity). + WatchdogFired bool + // ClockResets/MeaningfulBytes: adapter-classified progress evidence + // (Q33): count of meaningful chunks and their byte total. + ClockResets int + MeaningfulBytes int } // WatchdogHealthRecord mirrors the TS watchdog-health ledger row (Q34). @@ -214,39 +281,46 @@ func spawnCliStreaming(name string, args []string, opts SpawnCliOptions) SpawnCl mu.Unlock() watchdogStopped.Do(func() { close(watchdogStop) }) - // Q34: watchdog-health ledger row; requires an armed clock. - if opts.WatchdogLedger != nil && (noProgressMs > 0 || coldStartMs > 0) && opts.WatchdogSink != nil { - opts.WatchdogSink(WatchdogHealthRecord{ - Ts: time.Now().UTC().Format("2006-01-02T15:04:05.000Z07:00"), - Kind: "event", - Event: "watchdog-health", - TaskId: opts.WatchdogLedger.TaskId, - // FR-VIS: direct exec child — never operator-visible. When - // the operator asked for headless the row says so; otherwise - // this is a fallback from an attempted pane spawn - // (RunWorkerCli downgrades). - Runtime: "direct", - Visible: false, - Visibility: visibilityLabel(), - NoProgressTimeoutMs: noProgressMs, - WatchdogFired: wdFired, - ColdStartFired: csFired, - WallClockMs: wallClock.Milliseconds(), - ClockResets: resets, - MeaningfulBytes: mbytes, - IdleMs: idle.Milliseconds(), - Site: "spawn-cli", - Attempt: opts.WatchdogLedger.Attempt, - Worker: opts.WatchdogLedger.Worker, - }) + // Q34: watchdog-health ledger row; requires an armed clock and a + // ledger context. The default sink (FR-GO-05 #190 wiring) appends + // to the ledger events file; tests may inject a capture sink. + if opts.WatchdogLedger != nil && (noProgressMs > 0 || coldStartMs > 0) { + if sink := watchdogSinkFor(opts); sink != nil { + sink(WatchdogHealthRecord{ + Ts: time.Now().UTC().Format("2006-01-02T15:04:05.000Z07:00"), + Kind: "event", + Event: "watchdog-health", + TaskId: opts.WatchdogLedger.TaskId, + // FR-VIS: direct exec child — never operator-visible. + // When the operator asked for headless the row says so; + // otherwise this is a fallback from an attempted pane + // spawn (RunWorkerCli downgrades). + Runtime: "direct", + Visible: false, + Visibility: visibilityLabel(), + NoProgressTimeoutMs: noProgressMs, + WatchdogFired: wdFired, + ColdStartFired: csFired, + WallClockMs: wallClock.Milliseconds(), + ClockResets: resets, + MeaningfulBytes: mbytes, + IdleMs: idle.Milliseconds(), + Site: "spawn-cli", + Attempt: opts.WatchdogLedger.Attempt, + Worker: opts.WatchdogLedger.Worker, + }) + } } finishCh <- SpawnCliResult{ - ExitCode: exit, - Stdout: stdoutOut, - Stderr: stderrOut, - TimedOut: timedOutSnapshot, - ColdStart: csFired, + ExitCode: exit, + Stdout: stdoutOut, + Stderr: stderrOut, + TimedOut: timedOutSnapshot, + ColdStart: csFired, + WatchdogFired: wdFired, + ClockResets: resets, + MeaningfulBytes: mbytes, } } @@ -304,7 +378,6 @@ func spawnCliStreaming(name string, args []string, opts SpawnCliOptions) SpawnCl finish() return <-finishCh } - go func() { err := cmd.Wait() mu.Lock() @@ -317,8 +390,11 @@ func spawnCliStreaming(name string, args []string, opts SpawnCliOptions) SpawnCl mu.Unlock() // Node's 'close' event fires only after all stdio streams flush; // mirror that by waiting for the pipe drains so stdout/stderr are - // fully captured before the result snapshot. - drainWG.Wait() + // fully captured before the result snapshot — but bounded (issue + // #248): a pipe that never EOFs (grandchild inheriting the fds, + // a wedged flusher) must not push the attempt finish past its + // wall clock. + waitPipeDrain(&drainWG, opts.PostKillDrainWaitMs) finish() }() @@ -415,6 +491,30 @@ func drainProgress(r io.Reader, onChunk func(s string, meaningful bool)) { // loaded config, mirroring TS spawnVisibility(loadConfig()). var SpawnVisibilityConfig *config.Config +// waitPipeDrain blocks until all draining goroutines are done or the +// bounded wait expires: a clean call returns quickly; a never-draining +// pipe returns after waitMs (issue #248). The 0 sentinel uses the +// production cap; negative disables the cap (legacy callers that +// explicitly wait unbounded by contract). +func waitPipeDrain(wg *sync.WaitGroup, waitMs int) { + if waitMs < 0 { + wg.Wait() + return + } + if waitMs == 0 { + waitMs = DefaultPostKillDrainWaitMs + } + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(time.Duration(waitMs) * time.Millisecond): + } +} + func visibilityLabel() string { var cfg config.Config if SpawnVisibilityConfig != nil { diff --git a/internal/workers/spawnstream_watchdog_test.go b/internal/workers/spawnstream_watchdog_test.go index 5b702b2b..7b4c1b4a 100644 --- a/internal/workers/spawnstream_watchdog_test.go +++ b/internal/workers/spawnstream_watchdog_test.go @@ -5,10 +5,15 @@ package workers import ( + "encoding/json" "os" "path/filepath" + "strings" + "sync" "testing" "time" + + "github.com/FreePeak/devagent/internal/ledger" ) // fakeBin writes an executable shell script onto a fresh PATH dir and @@ -188,3 +193,134 @@ done`) t.Fatalf("watchdog fired too late: %v", time.Since(start)) } } + +// Issue #248 required change 1 (FR-GO-05 #190 TODO closure): a production +// spawn with an armed clock and NO explicit WatchdogSink still emits the +// Q34 watchdog-health row — the default sink appends it to the ledger +// events file of the WatchdogLedgerContext repo. +func TestSpawnCliStreaming_DefaultSinkAppendsLedgerRow(t *testing.T) { + repo := t.TempDir() + bin := fakeBin(t, "fake-worker-ledger", silentBody) + res := spawnCliStreaming(bin, nil, SpawnCliOptions{ + TimeoutMs: 30_000, + NoProgressTimeoutMs: intPtr(300), + WatchdogLedger: &WatchdogLedgerContext{RepoPath: repo, TaskId: "T-obs", Attempt: 3, Worker: "omp"}, + }) + if !res.TimedOut || !res.WatchdogFired { + t.Fatalf("expected no-progress kill, got %+v", res) + } + raw, err := os.ReadFile(filepath.Join(repo, ledger.LedgerDir, "events.jsonl")) + if err != nil { + t.Fatalf("watchdog-health row must land in the ledger: %v", err) + } + var row map[string]any + if err := json.Unmarshal(raw, &row); err != nil { + t.Fatalf("row = %s: %v", raw, err) + } + if row["event"] != "watchdog-health" || row["taskId"] != "T-obs" { + t.Fatalf("row identity = %v", row) + } + if row["site"] != "spawn-cli" || row["runtime"] != "direct" { + t.Fatalf("direct-exec row = %v", row) + } + if row["watchdogFired"] != true { + t.Fatalf("row must record the firing: %v", row) + } + // herdr-pane parity: attempt and worker come from the ledger context. + if row["attempt"] != float64(3) || row["worker"] != "omp" { + t.Fatalf("row identity fields = %v", row) + } +} + +// Issue #248 required change 1: no ledger context = no row (probe/one-off +// spawns are not orchestrated runs), even with a clock armed. +func TestSpawnCliStreaming_NoContextNoRow(t *testing.T) { + repo := t.TempDir() + bin := fakeBin(t, "fake-worker-noctx", silentBody) + spawnCliStreaming(bin, nil, SpawnCliOptions{ + TimeoutMs: 30_000, + NoProgressTimeoutMs: intPtr(300), + }) + if _, err := os.Stat(filepath.Join(repo, ledger.LedgerDir)); !os.IsNotExist(err) { + t.Fatalf("no-ledger-context spawn must not write any ledger: %v", err) + } +} + +// The drain cap must be comfortably under the grandchild lifetime: the +// legacy unbounded wait would block ~5s here (the grandchild holds the +// pipe until sleep 5 exits), the cap returns at ~500ms with everything +// the drain captured so far. +func TestSpawnCliStreaming_PostKillDrainBound(t *testing.T) { + body := `echo '{"type":"tool_execution_start","toolName":"read"}' +sleep 5 & +echo parent-done +exit 0` + bin := fakeBin(t, "fake-worker-drain", body) + start := time.Now() + res := spawnCliStreaming(bin, nil, SpawnCliOptions{ + TimeoutMs: 30_000, + NoProgressTimeoutMs: intPtr(5000), + PostKillDrainWaitMs: 500, + }) + elapsed := time.Since(start) + if res.TimedOut || res.ExitCode != 0 { + t.Fatalf("expected clean exit, got %+v", res) + } + if elapsed >= 3*time.Second { + t.Fatalf("drain wait must be bounded well under the grandchild lifetime, took %v", elapsed) + } + if !strings.Contains(res.Stdout, "parent-done") { + t.Fatalf("drained stdout must keep the captured prefix, got %q", res.Stdout) + } +} + +// Issue #248 required change 4: the bounded wait must actually bound. A +// WaitGroup whose drains never finish returns after ~waitMs with the cap. +func TestWaitPipeDrain_Bounded(t *testing.T) { + var wg sync.WaitGroup + wg.Add(1) // never Done: models a pipe held open past the wall + start := time.Now() + waitPipeDrain(&wg, 100) + elapsed := time.Since(start) + if elapsed < 90*time.Millisecond || elapsed > 3*time.Second { + t.Fatalf("bounded wait = %v, want ~100ms", elapsed) + } + // A clean drain returns immediately even with a huge cap. + var done sync.WaitGroup + done.Add(1) + done.Done() + start = time.Now() + waitPipeDrain(&done, 60_000) + if time.Since(start) > time.Second { + t.Fatalf("clean drain must not wait the cap: %v", time.Since(start)) + } +} + +// Negative wait = legacy unbounded opt-in: blocks until every drain +// finishes. +func TestWaitPipeDrain_UnboundedOptIn(t *testing.T) { + var wg sync.WaitGroup + wg.Add(1) + released := make(chan struct{}) + go func() { + time.Sleep(200 * time.Millisecond) + wg.Done() + close(released) + }() + waitPipeDrain(&wg, -1) + <-released +} + +// 0 = the production default cap (named const, default 3s per the issue). +func TestWaitPipeDrain_DefaultCapSentinel(t *testing.T) { + if DefaultPostKillDrainWaitMs != 3000 { + t.Fatalf("production cap = %d, want 3000", DefaultPostKillDrainWaitMs) + } + var wg sync.WaitGroup + wg.Add(1) + start := time.Now() + waitPipeDrain(&wg, 0) + if elapsed := time.Since(start); elapsed > time.Duration(DefaultPostKillDrainWaitMs)*time.Millisecond+3*time.Second { + t.Fatalf("default cap not applied: %v", elapsed) + } +} diff --git a/internal/workers/types.go b/internal/workers/types.go index 8571a1c6..f2ab21c4 100644 --- a/internal/workers/types.go +++ b/internal/workers/types.go @@ -68,6 +68,15 @@ type WorkerResult struct { // ColdStart: true when the launch was killed by the cold-start // (first-progress) deadline. Classified transient alongside NoProgress. ColdStart bool + // WatchdogFired: Q34 — true when the no-progress clock killed the + // launch (direct path). A wall-clock kill never sets it. + WatchdogFired bool + // ClockResets/MeaningfulBytes: Q33/Q34 stream progress metrics the + // spawn path observed for this launch. 0 for spawns without an armed + // streaming clock. The classifier (impIsInfraTransient) uses them to + // tell a productive wall-kill from a zero-progress burn. + ClockResets int + MeaningfulBytes int // CostUsdTicks: FR-GROK-03 xAI cost recorded verbatim for this run. // Nil = undefined (a missing cost is never coerced to 0). CostUsdTicks *float64