From c9a6a3b4fc876431eb79d73ce7cbd18b2ca379df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=98=BF=E7=A8=B3?= Date: Fri, 23 Oct 2026 18:19:23 +0800 Subject: [PATCH] fix(delegation): track background subagent progress MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A background sub-agent writes its work into its own transcript and its tool calls, never into the job that started it. The job's activity stamp therefore never moved, and the watchdog judged the run by bytes on streams that stay empty: a busy sub-agent read as a job that had been silent since it started, and a stalled notice went out about work that was progressing. Bind the child's progress to its job at beginExecution — the one boundary every delegated run crosses, foreground included, where a context with no job binds nothing. A finished tool call is one reported step; token deltas refresh the activity stamp at most once every two seconds, because the watchdog reads minutes and the child emits far more often than that. Reported steps are kept apart from output text rather than folded into it. They answer different questions — whether the child advanced, and whether it wrote here — and a caller decides from the pair. `wait` says both, and a stalled notice says what the job had reported before going quiet, since a job that worked and stopped needs a different next move than one that never got going. Tested: internal/tools/jobs (a job that writes nothing stays alive while it reports progress; wait reads steps separately from output), internal/tools/builtin (progress wording per reading), internal/runtime/delegation (binding at beginExecution, a finished tool call reports one step, deltas are paced, a foreground run stays unbound), internal/contract/event, and desktop frontend-next (typecheck; full vitest run, 6120 tests). The desktop stalled-notice sentence had no English for its new wording and the i18n gate caught it; the three entries are added here. --- desktop/frontend-next/src/i18n/en.ts | 6 +- desktop/frontend-next/src/i18n/job_notice.ts | 20 ++- desktop/frontend-next/src/i18n/notices.ts | 2 +- .../src/ui/cards/NoticeCard.test.tsx | 13 +- internal/contract/event/notice_payload.go | 6 +- .../delegation/delegation_lifecycle.go | 4 + .../delegation_liveness_binding_test.go | 70 +++++++++ .../runtime/delegation/subagent_liveness.go | 63 ++++++++ .../delegation/subagent_liveness_test.go | 142 ++++++++++++++++++ .../runtime/delegation/subagent_progress.go | 10 +- internal/tools/builtin/bgjobs.go | 22 +-- internal/tools/builtin/wait_steps_test.go | 54 +++++++ internal/tools/jobs/artifacts.go | 4 + internal/tools/jobs/jobs.go | 51 ++++++- internal/tools/jobs/jobs_test.go | 10 +- internal/tools/jobs/liveness_test.go | 133 ++++++++++++++++ internal/tools/jobs/progress.go | 6 + internal/tools/jobs/stalled_notice_test.go | 27 +++- internal/tools/jobs/watch.go | 11 ++ 19 files changed, 612 insertions(+), 42 deletions(-) create mode 100644 internal/runtime/delegation/delegation_liveness_binding_test.go create mode 100644 internal/runtime/delegation/subagent_liveness.go create mode 100644 internal/runtime/delegation/subagent_liveness_test.go create mode 100644 internal/tools/builtin/wait_steps_test.go create mode 100644 internal/tools/jobs/liveness_test.go diff --git a/desktop/frontend-next/src/i18n/en.ts b/desktop/frontend-next/src/i18n/en.ts index 3d5fb8017a..545a5bac42 100644 --- a/desktop/frontend-next/src/i18n/en.ts +++ b/desktop/frontend-next/src/i18n/en.ts @@ -1824,8 +1824,10 @@ export const EN: Record = { "后台任务已结束：{name}": "Background job finished: {name}", "后台任务已终止：{name}": "Background job killed: {name}", "后台任务 {name} 失败，需要处理": "Background job {name} failed: needs attention", - "后台任务 {name} 已有 {quiet} 没往自己的输出里写任何内容：可能是在写文件，也可能卡住了；可以用 wait 或 bash_output 查看，或用 kill_shell 结束它": - "Background job {name} has written nothing to its own output for {quiet}: it may be writing to a file, or stuck; check it with wait or bash_output, or stop it with kill_shell", + "后台任务 {name} 已有 {quiet} 没有活动（{work}）：可能在等某个东西，也可能卡住了；可以用 wait 或 bash_output 查看，或用 kill_shell 结束它": + "Background job {name} has been inactive for {quiet} ({work}): it may be waiting on something, or stuck; check it with wait or bash_output, or stop it with kill_shell", + "{n} 步在静默前完成": "{n} step(s) completed before going quiet", + "静默前没有任何进展": "no progress reported", "被信号 {signal} 终止": "Ended by signal {signal}", "退出码 {code}": "Exit code {code}", "已恢复 {n} 条未完成的指令。待发送已暂停，请先在输入框上方的队列里查看，再点“继续派发”": "Recovered {n} unfinished instruction(s). Sending is paused: review them in the queue above the input box, then press Release", diff --git a/desktop/frontend-next/src/i18n/job_notice.ts b/desktop/frontend-next/src/i18n/job_notice.ts index 4d37d3ce80..0538bd2c39 100644 --- a/desktop/frontend-next/src/i18n/job_notice.ts +++ b/desktop/frontend-next/src/i18n/job_notice.ts @@ -20,16 +20,22 @@ export function jobNoticeVars(detail: string | undefined): { name: string } | un } } -// A stalled job's sentence also says how long its own output has been silent: -// without that span the sentence would be half-filled, so it reads as no payload. -export function jobStalledVars(detail: string | undefined): { name: string; quiet: string } | undefined { +// A stalled job's sentence also says how long it has been inactive, and whether +// it had reported any steps by then: a job that worked and stopped is waiting on +// something, while one that never reported may be writing elsewhere. Without the +// span the sentence would be half-filled, so it reads as no payload. +export function jobStalledVars(detail: string | undefined): { name: string; quiet: string; work: string } | undefined { const named = jobNoticeVars(detail); if (!named || !detail) return undefined; try { - const p = JSON.parse(detail) as { quietSeconds?: unknown }; - return typeof p.quietSeconds === "number" && Number.isInteger(p.quietSeconds) && p.quietSeconds > 0 - ? { ...named, quiet: elapsed(p.quietSeconds * 1000) } - : undefined; + const p = JSON.parse(detail) as { quietSeconds?: unknown; steps?: unknown }; + if (typeof p.quietSeconds !== "number" || !Number.isInteger(p.quietSeconds) || p.quietSeconds <= 0) return undefined; + const steps = typeof p.steps === "number" && Number.isInteger(p.steps) && p.steps > 0 ? p.steps : 0; + return { + ...named, + quiet: elapsed(p.quietSeconds * 1000), + work: steps > 0 ? t("{n} 步在静默前完成", { n: steps }) : t("静默前没有任何进展"), + }; } catch { return undefined; } diff --git a/desktop/frontend-next/src/i18n/notices.ts b/desktop/frontend-next/src/i18n/notices.ts index 3760264735..30342d43fa 100644 --- a/desktop/frontend-next/src/i18n/notices.ts +++ b/desktop/frontend-next/src/i18n/notices.ts @@ -43,7 +43,7 @@ export const NOTICE_TEXT: Record = { job_finished: "后台任务已结束：{name}", job_killed: "后台任务已终止：{name}", job_failed: "后台任务 {name} 失败，需要处理", - job_stalled: "后台任务 {name} 已有 {quiet} 没往自己的输出里写任何内容：可能是在写文件，也可能卡住了；可以用 wait 或 bash_output 查看，或用 kill_shell 结束它", + job_stalled: "后台任务 {name} 已有 {quiet} 没有活动（{work}）：可能在等某个东西，也可能卡住了；可以用 wait 或 bash_output 查看，或用 kill_shell 结束它", compacted: "已压缩", compact_declined: "无需压缩：{why}", compact_failed: "压缩失败：{why}", diff --git a/desktop/frontend-next/src/ui/cards/NoticeCard.test.tsx b/desktop/frontend-next/src/ui/cards/NoticeCard.test.tsx index 296b021017..948916075d 100644 --- a/desktop/frontend-next/src/ui/cards/NoticeCard.test.tsx +++ b/desktop/frontend-next/src/ui/cards/NoticeCard.test.tsx @@ -221,10 +221,15 @@ describe("a background job notice", () => { expect(said).not.toContain("background bash"); }); - it("words a stalled job from its payload: the name and how long it has been quiet", () => { - const said = text({ level: "warn", code: "job_stalled", text: "background bash bash-16 has written nothing to its own output for 15m0s", detail: JSON.stringify({ kind: "bash", id: "bash-16", label: "make build", quietSeconds: 900 }) }); - expect(said).toBe(t("后台任务 {name} 已有 {quiet} 没往自己的输出里写任何内容：可能是在写文件，也可能卡住了；可以用 wait 或 bash_output 查看，或用 kill_shell 结束它", { name: "make build", quiet: "15m" })); - expect(said).not.toContain("has written nothing"); + it("words a stalled job from its payload: the name, the span, and what it had reported", () => { + const said = text({ level: "warn", code: "job_stalled", text: "background bash bash-16 has been inactive for 15m0s", detail: JSON.stringify({ kind: "bash", id: "bash-16", label: "make build", quietSeconds: 900, steps: 12 }) }); + expect(said).toBe(t("后台任务 {name} 已有 {quiet} 没有活动（{work}）：可能在等某个东西，也可能卡住了；可以用 wait 或 bash_output 查看，或用 kill_shell 结束它", { name: "make build", quiet: "15m", work: t("{n} 步在静默前完成", { n: 12 }) })); + expect(said).not.toContain("has been inactive"); + }); + + it("says a stalled job reported no progress when it reported no steps", () => { + const said = text({ level: "warn", code: "job_stalled", text: "x", detail: JSON.stringify({ kind: "bash", id: "bash-16", quietSeconds: 900 }) }); + expect(said).toContain(t("静默前没有任何进展")); }); it("keeps the kernel's text for a stalled job whose payload does not say how long", () => { diff --git a/internal/contract/event/notice_payload.go b/internal/contract/event/notice_payload.go index df67b0a88d..e3f4f83c27 100644 --- a/internal/contract/event/notice_payload.go +++ b/internal/contract/event/notice_payload.go @@ -103,8 +103,12 @@ type JobNotice struct { // knew: Signal is the signal number that ended it, zero when none did. ExitCode *int `json:"exitCode,omitempty"` Signal int `json:"signal,omitempty"` - // QuietSeconds is how long the job's own output has been silent, on a stalled notice only. + // QuietSeconds is how long the job has been inactive, on a stalled notice only. QuietSeconds int `json:"quietSeconds,omitempty"` + // Steps is what the job had reported before it went quiet, on a stalled + // notice only. Zero means it never reported any, which is a different + // situation from a job that worked and then stopped. + Steps int `json:"steps,omitempty"` } // Encode renders the payload as the notice's Detail. diff --git a/internal/runtime/delegation/delegation_lifecycle.go b/internal/runtime/delegation/delegation_lifecycle.go index 8369e2433b..d7d4d6a74b 100644 --- a/internal/runtime/delegation/delegation_lifecycle.go +++ b/internal/runtime/delegation/delegation_lifecycle.go @@ -134,6 +134,10 @@ func (t *TaskTool) beginExecution(ctx context.Context, life *delegationLifecycle if node, drawn := executionNode(ctx, life); drawn { publishGraph(node.sink, fanOutItemRunningDelta(node.id)) } + // The child's work reaches its transcript, never the job's own streams, so + // a background run that does not report here reads as silent for its whole + // life. A foreground context carries no job, so this is one path for both. + bindJobLiveness(ctx, trk) trk.running() return nil } diff --git a/internal/runtime/delegation/delegation_liveness_binding_test.go b/internal/runtime/delegation/delegation_liveness_binding_test.go new file mode 100644 index 0000000000..8da9af3422 --- /dev/null +++ b/internal/runtime/delegation/delegation_liveness_binding_test.go @@ -0,0 +1,70 @@ +package delegation + +import ( + "context" + "io" + "testing" + "time" + + "reasonix/internal/contract/event" + "reasonix/internal/tools/jobs" +) + +// A background sub-agent works in its own transcript, never in the job's own +// streams. beginExecution is the single boundary every delegated run crosses, +// and it is where the child's liveness gets bound to its job: without that, the +// job's activity stamp never moves and a busy run reads as stalled for its whole +// life. This asserts the binding itself, so removing that one call fails here. +func TestBeginningExecutionBindsTheChildToItsJob(t *testing.T) { + jm := jobs.NewManager(event.Discard) + defer jm.Close() + + ready := make(chan context.Context, 1) + j := jm.Start("task", "probe", func(jobCtx context.Context, _ io.Writer) (string, error) { + ready <- jobCtx + <-jobCtx.Done() + return "", jobCtx.Err() + }) + defer jm.Kill(j.ID) + + var jobCtx context.Context + select { + case jobCtx = <-ready: + case <-time.After(2 * time.Second): + t.Fatal("background job never started") + } + + trk := tracked(t, newFakeProgressClock(time.Unix(0, 0)), chanSink{ch: make(chan event.Event, 64)}) + if trk.noteProgress != nil { + t.Fatal("tracker acquired a notifier before execution began") + } + task := &TaskTool{} + if err := task.beginExecution(jobCtx, nil, true, trk, nil); err != nil { + t.Fatalf("beginExecution: %v", err) + } + if trk.noteProgress == nil { + t.Fatal("beginning execution left the child unbound: its job will read as silent") + } + + trk.noteProgress(1) + res, _ := jm.Wait(context.Background(), []string{j.ID}, jobs.WaitOptions{Timeout: time.Second}) + if len(res) != 1 || res[0].Progress == nil { + t.Fatalf("job state = %+v, want a running job with progress", res) + } + if got := res[0].Progress.Steps; got != 1 { + t.Fatalf("Steps = %d, want 1: the bound notifier never reached the job", got) + } +} + +// A foreground run crosses the same boundary but has no job to tell, so it must +// stay unbound rather than acquire a notifier with nothing to report to. +func TestBeginningExecutionLeavesAForegroundRunUnbound(t *testing.T) { + trk := tracked(t, newFakeProgressClock(time.Unix(0, 0)), chanSink{ch: make(chan event.Event, 64)}) + task := &TaskTool{} + if err := task.beginExecution(context.Background(), nil, true, trk, nil); err != nil { + t.Fatalf("beginExecution: %v", err) + } + if trk.noteProgress != nil { + t.Fatal("a foreground run acquired a background notifier") + } +} diff --git a/internal/runtime/delegation/subagent_liveness.go b/internal/runtime/delegation/subagent_liveness.go new file mode 100644 index 0000000000..0a63bd6b40 --- /dev/null +++ b/internal/runtime/delegation/subagent_liveness.go @@ -0,0 +1,63 @@ +package delegation + +import ( + "context" + "time" + + "reasonix/internal/contract/event" + "reasonix/internal/tools/jobs" +) + +// A child's work reaches its transcript and tool calls, never its job's own +// streams, so the job learns the child is alive through here or not at all. + +// noteProgressPace bounds how often a child refreshes its job's activity stamp. +// The watchdog reads minutes, so anything well below that is continuous activity. +const noteProgressPace = 2 * time.Second + +// shouldNoteLocked decides whether this event refreshes the owning job's +// liveness, and records the moment when it does. Called under t.mu; the caller +// invokes the notifier after releasing it, so the two locks never nest. +func (t *subagentProgressTracker) shouldNoteLocked(kind event.Kind) (steps int, note bool) { + if t.noteProgress == nil { + return 0, false + } + // A finished tool call is one completed unit of work, at whatever rate the + // child produces them. Everything else is the run still breathing, paced + // because token deltas arrive far more often than a watchdog needs. + if kind == event.ToolResult { + t.notedAt = t.merger.clock.Now() + return 1, true + } + now := t.merger.clock.Now() + if !t.notedAt.IsZero() && now.Sub(t.notedAt) < noteProgressPace { + return 0, false + } + t.notedAt = now + return 0, true +} + +// bindJobLiveness points a child's reports at the background job it runs under. +// beginExecution is the boundary every delegated run crosses, and a foreground +// context carries no job, so this needs no second call site. +func bindJobLiveness(ctx context.Context, trk *subagentProgressTracker) { + if !jobs.HasJob(ctx) { + return + } + trk.mu.Lock() + defer trk.mu.Unlock() + trk.noteProgress = func(steps int) { jobs.NoteProgress(ctx, steps) } +} + +// noteJobProgress reports one child event to the job, if this event deserves a +// report. It takes the tracker's lock and releases it before calling out, so the +// child and the job never hold their locks in an order either could reverse. +func (t *subagentProgressTracker) noteJobProgress(kind event.Kind) { + t.mu.Lock() + steps, note := t.shouldNoteLocked(kind) + notify := t.noteProgress + t.mu.Unlock() + if note && notify != nil { + notify(steps) + } +} diff --git a/internal/runtime/delegation/subagent_liveness_test.go b/internal/runtime/delegation/subagent_liveness_test.go new file mode 100644 index 0000000000..e07f4837ec --- /dev/null +++ b/internal/runtime/delegation/subagent_liveness_test.go @@ -0,0 +1,142 @@ +package delegation + +import ( + "context" + "io" + "sync" + "testing" + "time" + + "reasonix/internal/contract/event" + "reasonix/internal/runtime/agent" + "reasonix/internal/tools/jobs" +) + +// noteRecorder collects what a tracker told its job. +type noteRecorder struct { + mu sync.Mutex + steps []int + calls int +} + +func (r *noteRecorder) note(steps int) { + r.mu.Lock() + defer r.mu.Unlock() + r.calls++ + r.steps = append(r.steps, steps) +} + +func (r *noteRecorder) snapshot() (calls int, steps []int) { + r.mu.Lock() + defer r.mu.Unlock() + return r.calls, append([]int(nil), r.steps...) +} + +// tracked builds a tracker joined to a shared merger, so tests control the clock. +func tracked(t *testing.T, clock progressClock, sink event.Sink) *subagentProgressTracker { + t.Helper() + merger := newSubagentProgressMerger(clock, sink, "group-1") + t.Cleanup(merger.Close) + ctx := withSubagentProgressMerger(agent.WithCallContext(context.Background(), "child-1", sink, nil, false), merger) + return newSubagentProgressTracker(ctx, subSink(ctx)) +} + +// A finished tool call is a completed unit of work and is reported as such, at +// whatever rate the child produces them: it is the only signal a job whose +// streams stay empty ever gets. +func TestAFinishedToolCallReportsOneStepToTheJob(t *testing.T) { + clock := newFakeProgressClock(time.Unix(0, 0)) + trk := tracked(t, clock, chanSink{ch: make(chan event.Event, 256)}) + rec := ¬eRecorder{} + trk.noteProgress = rec.note + wrap := trk.wrap() + + for range 3 { + wrap.Emit(event.Event{Kind: event.ToolResult, Tool: event.Tool{ID: "bash_1", Name: "bash"}}) + } + calls, steps := rec.snapshot() + if calls != 3 { + t.Fatalf("notified %d times, want one per finished tool call", calls) + } + for i, s := range steps { + if s != 1 { + t.Fatalf("step report %d = %d, want 1", i, s) + } + } +} + +// Token deltas arrive far more often than a watchdog reading minutes needs, so +// they are paced. What matters is that the job keeps hearing the run is alive. +func TestTokenDeltasRefreshLivenessAtMostOncePerPace(t *testing.T) { + clock := newFakeProgressClock(time.Unix(0, 0)) + trk := tracked(t, clock, chanSink{ch: make(chan event.Event, 256)}) + rec := ¬eRecorder{} + trk.noteProgress = rec.note + wrap := trk.wrap() + + wrap.Emit(event.Event{Kind: event.Text, Text: "a"}) + if calls, _ := rec.snapshot(); calls != 1 { + t.Fatalf("first delta notified %d times, want 1", calls) + } + clock.Advance(noteProgressPace / 2) + wrap.Emit(event.Event{Kind: event.Text, Text: "b"}) + if calls, _ := rec.snapshot(); calls != 1 { + t.Fatalf("delta inside the window notified again: %d calls", calls) + } + clock.Advance(noteProgressPace) + wrap.Emit(event.Event{Kind: event.Text, Text: "c"}) + if calls, steps := rec.snapshot(); calls != 2 { + t.Fatalf("delta past the window notified %d times, want 2 (steps %v)", calls, steps) + } +} + +// A foreground run has no job to tell, and must not acquire a notifier. +func TestAForegroundRunReportsNothing(t *testing.T) { + trk := tracked(t, newFakeProgressClock(time.Unix(0, 0)), chanSink{ch: make(chan event.Event, 256)}) + bindJobLiveness(context.Background(), trk) + trk.wrap().Emit(event.Event{Kind: event.ToolResult, Tool: event.Tool{ID: "bash_1", Name: "bash"}}) + if trk.noteProgress != nil { + t.Fatal("a foreground tracker acquired a background notifier") + } +} + +// End to end through the real registry: child tool results reach the job's step +// count, which is what keeps a job that writes nothing from reading as idle. +func TestChildToolResultsReachTheOwningJob(t *testing.T) { + m := jobs.NewManager(event.Discard) + defer m.Close() + + release := make(chan struct{}) + defer close(release) + ctxSink := make(chan event.Event, 256) + started := make(chan struct{}) + j := m.Start("task", "audit", func(jobCtx context.Context, _ io.Writer) (string, error) { + trk := tracked(t, newFakeProgressClock(time.Unix(0, 0)), chanSink{ch: ctxSink}) + bindJobLiveness(jobCtx, trk) + if trk.noteProgress == nil { + t.Error("a background run did not acquire a notifier") + } + wrap := trk.wrap() + for range 4 { + wrap.Emit(event.Event{Kind: event.ToolResult, Tool: event.Tool{ID: "bash_1", Name: "bash"}}) + } + close(started) + <-release + return "", nil + }) + defer m.Kill(j.ID) + + <-started + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + res, _ := m.Wait(context.Background(), []string{j.ID}, jobs.WaitOptions{Timeout: 10 * time.Millisecond}) + if len(res) == 1 && res[0].Progress != nil && res[0].Progress.Steps == 4 { + if res[0].Progress.Produced != 0 { + t.Fatalf("Produced = %d, want no output text: the child writes none", res[0].Progress.Produced) + } + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("child tool results never reached the job's step count") +} diff --git a/internal/runtime/delegation/subagent_progress.go b/internal/runtime/delegation/subagent_progress.go index 1fd0c46381..92cfbc50e1 100644 --- a/internal/runtime/delegation/subagent_progress.go +++ b/internal/runtime/delegation/subagent_progress.go @@ -605,8 +605,8 @@ func utf8SafeTail(s string, maxBytes int) string { // subagentProgressTracker is the per-child state machine installed between a // sub-agent run and its parent sink. It converts the child's reasoning/text/ -// notice/retrying into preview slots on the group merger, forwards tool -// activity unchanged, and guarantees exactly one terminal status event. +// notice/retrying into preview slots, forwards tool activity, and guarantees +// exactly one terminal status event. type subagentProgressTracker struct { mu sync.Mutex merger *subagentProgressMerger @@ -618,6 +618,11 @@ type subagentProgressTracker struct { started time.Time ownsMerger bool done bool + + // noteProgress reports this run to the job it runs under; nil when it runs + // in the foreground. notedAt paces those reports. See subagent_liveness.go. + noteProgress func(steps int) + notedAt time.Time } // newSubagentProgressTracker creates (or joins) the group merger and returns a @@ -738,6 +743,7 @@ func (s trackedChildSink) Emit(e event.Event) { t.setPhaseLocked(subagentPhaseTool) } t.mu.Unlock() + t.noteJobProgress(e.Kind) switch e.Kind { case event.ToolDispatch, event.ToolResult, event.ToolProgress: e.Source = event.UsageSourceSubagent diff --git a/internal/tools/builtin/bgjobs.go b/internal/tools/builtin/bgjobs.go index a6e9bb187a..841f2b4c5b 100644 --- a/internal/tools/builtin/bgjobs.go +++ b/internal/tools/builtin/bgjobs.go @@ -295,21 +295,23 @@ func (waitJob) Execute(ctx context.Context, args json.RawMessage) (string, error // stillRunning states what happened while the caller waited. No thresholds and // no verdict: a task making slow progress and one that has stopped differ by -// whether anything arrived, and the caller is the one who knows which of those -// its own task should look like. +// what arrived, and the caller is the one who knows which of those its own task +// should look like. Output text and reported steps are separate readings because +// a sub-agent works in its own transcript: it advances without ever writing here. func stillRunning(p *jobs.Progress) string { if p == nil { return "" } - produced := "nothing on its own output so far" - if p.Produced > 0 { - produced = fmt.Sprintf("%s produced", byteCount(p.Produced)) + parts := []string{fmt.Sprintf("running %s", p.Running.Round(time.Second))} + switch { + case p.Produced > 0: + parts = append(parts, fmt.Sprintf("%s of output, %s of it while waiting", byteCount(p.Produced), byteCount(p.Delta))) + case p.Steps > 0: + parts = append(parts, fmt.Sprintf("no output text (it does not write here), %d step(s) reported, %d while waiting", p.Steps, p.StepDelta)) + default: + parts = append(parts, "no output text and no steps reported yet") } - arrived := "nothing arrived while waiting" - if p.Delta > 0 { - arrived = fmt.Sprintf("%s arrived while waiting", byteCount(p.Delta)) - } - return fmt.Sprintf(" — running %s, %s, %s", p.Running.Round(time.Second), produced, arrived) + return " — " + strings.Join(parts, ", ") } func byteCount(n int64) string { diff --git a/internal/tools/builtin/wait_steps_test.go b/internal/tools/builtin/wait_steps_test.go new file mode 100644 index 0000000000..3372bbf902 --- /dev/null +++ b/internal/tools/builtin/wait_steps_test.go @@ -0,0 +1,54 @@ +package builtin + +import ( + "strings" + "testing" + "time" + + "reasonix/internal/tools/jobs" +) + +// What a wait reports must separate the two readings a caller acts on: a job +// that produced output text, and one that advanced without writing here at all. +func TestWaitProgressDistinguishesOutputFromSteps(t *testing.T) { + cases := []struct { + name string + progress *jobs.Progress + want []string + absent []string + }{ + { + name: "no readings yet", + progress: &jobs.Progress{Running: 90 * time.Second}, + want: []string{"no output text and no steps reported yet"}, + absent: []string{"step(s) reported"}, + }, + { + name: "steps without output text", + progress: &jobs.Progress{Running: 90 * time.Second, Steps: 12, StepDelta: 3}, + want: []string{"no output text (it does not write here)", "12 step(s) reported", "3 while waiting"}, + absent: []string{"of output"}, + }, + { + name: "output text", + progress: &jobs.Progress{Running: 90 * time.Second, Produced: 2048, Delta: 1024}, + want: []string{"2.0 KB of output", "1.0 KB of it while waiting"}, + absent: []string{"step(s) reported"}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := stillRunning(tc.progress) + for _, w := range tc.want { + if !strings.Contains(got, w) { + t.Fatalf("stillRunning = %q, want %q", got, w) + } + } + for _, a := range tc.absent { + if strings.Contains(got, a) { + t.Fatalf("stillRunning = %q, want no %q", got, a) + } + } + }) + } +} diff --git a/internal/tools/jobs/artifacts.go b/internal/tools/jobs/artifacts.go index 8351338a03..314b2ffb11 100644 --- a/internal/tools/jobs/artifacts.go +++ b/internal/tools/jobs/artifacts.go @@ -21,6 +21,10 @@ type jobTimes struct { started int64 activity int64 finished int64 + // steps counts what the runner reported as done, which a job whose work + // never reaches its own streams has and a hung one does not. It is the + // other half of activity: a sub-agent writes a transcript, not job output. + steps int64 } // touch records that the job's output just moved. diff --git a/internal/tools/jobs/jobs.go b/internal/tools/jobs/jobs.go index d51badcc30..b43471f461 100644 --- a/internal/tools/jobs/jobs.go +++ b/internal/tools/jobs/jobs.go @@ -629,8 +629,9 @@ func (m *Manager) monitorStalled(parentSession string, j *Job) { idle := time.Since(time.UnixMilli(j.times.activity)) if idle >= m.stalledWarning && !j.stalled { j.stalled = true + steps := j.times.steps j.mu.Unlock() - m.recordStalled(parentSession, j.ID, j.Kind, j.Label) + m.recordStalled(parentSession, j.ID, j.Kind, j.Label, steps) return } wait := m.stalledWarning - idle @@ -643,21 +644,30 @@ func (m *Manager) monitorStalled(parentSession string, j *Job) { } } -func (m *Manager) recordStalled(parentSession, id, kind, label string) { +// recordStalled reports a job that has gone quiet. steps says how much work it +// had reported by then: a job with steps and no recent activity is waiting on +// something, while one with none never got going, and the two need different +// answers from the reader. +func (m *Manager) recordStalled(parentSession, id, kind, label string, steps int64) { tag := id if label != "" { tag = fmt.Sprintf("%s (%s)", id, label) } parentSession = strings.TrimSpace(parentSession) + quiet := m.stalledWarning.Round(time.Second) + // What went quiet is two different things: a job that reported work and + // stopped is waiting on something, while one that never reported may simply + // be writing to a file rather than its own streams. + work := "no progress has been reported, so it may be writing elsewhere rather than stuck" + if steps > 0 { + work = fmt.Sprintf("%d step(s) were reported before it went quiet, so it may be waiting on something", steps) + } m.mu.Lock() if parentSession != "" && m.destroying[parentSession] { m.mu.Unlock() return } - // What is true, not what it looks like: silence on a job's own streams is - // also what a task redirecting into a file looks like, and calling that - // stalled sends the reader after the wrong thing. - text := fmt.Sprintf("%s has written nothing to its own output for %s — it may be writing to a file instead, or it may be stuck. Check with wait or bash_output, or stop it with kill_shell.", tag, m.stalledWarning.Round(time.Second)) + text := fmt.Sprintf("%s has been inactive for %s — %s. It may be waiting on something, or it may be stuck. Check with wait or bash_output, or stop it with kill_shell.", tag, quiet, work) m.completed = append(m.completed, completion{sessionID: parentSession, text: text}) active := m.active shouldEmit := active == "" || parentSession == "" || active == parentSession @@ -667,8 +677,8 @@ func (m *Manager) recordStalled(parentSession, id, kind, label string) { Kind: event.Notice, Level: event.LevelWarn, Code: event.NoticeCodeJobStalled, - Text: fmt.Sprintf("background %s %s has written nothing to its own output for %s — it may be writing to a file, or stuck; check with wait/bash_output or stop with kill_shell", kind, id, m.stalledWarning.Round(time.Second)), - Detail: event.JobNotice{Kind: kind, ID: id, Label: label, QuietSeconds: int(m.stalledWarning / time.Second)}.Encode(), + Text: fmt.Sprintf("background %s %s has been inactive for %s — %s; check with wait/bash_output or stop with kill_shell", kind, id, quiet, work), + Detail: event.JobNotice{Kind: kind, ID: id, Label: label, QuietSeconds: int(m.stalledWarning / time.Second), Steps: int(steps)}.Encode(), }) } } @@ -1704,6 +1714,31 @@ func PublishEvidence(ctx context.Context, summary evidence.ChildEvidenceSummary) j.mu.Unlock() } +// HasJob reports whether ctx belongs to a running background job. A runner whose +// work never reaches the job's own streams asks this before promising to report. +func HasJob(ctx context.Context) bool { + j, _ := ctx.Value(jobCtxKey{}).(*Job) + return j != nil +} + +// NoteProgress records that the job's runner is alive: it refreshes the job's +// activity stamp, and counts the work unit when steps is non-zero. A sub-agent +// writes a transcript and calls tools; none of that is job output, so counting +// bytes alone reports a busy job as silent for its whole life. This claims +// liveness and step count, never that any output text was produced. +func NoteProgress(ctx context.Context, steps int) { + j, _ := ctx.Value(jobCtxKey{}).(*Job) + if j == nil { + return + } + j.mu.Lock() + if steps > 0 { + j.times.steps += int64(steps) + } + j.times.touch() + j.mu.Unlock() +} + // LeaseEvidenceForSession returns a copy of a terminal job's evidence without // consuming it. Collection is only provisional: the receipts merge into the // collecting turn's ledger, but that ledger is discarded if the turn is diff --git a/internal/tools/jobs/jobs_test.go b/internal/tools/jobs/jobs_test.go index a52ab57dce..3e2ccc64aa 100644 --- a/internal/tools/jobs/jobs_test.go +++ b/internal/tools/jobs/jobs_test.go @@ -288,7 +288,7 @@ func TestStalledWarningIgnoresReturnedJobBeforeTerminalStatusPublished(t *testin note := m.DrainCompletedNote() // Negative, so it has to track the wording it is guarding against: matching // a sentence nothing says any more passes for the wrong reason. - if strings.Contains(note, "nothing to its own output") { + if strings.Contains(note, "has been inactive for") { t.Fatalf("got false stalled warning for already-returned job %s: %q", j.ID, note) } if !strings.Contains(note, j.ID) || !strings.Contains(note, string(Done)) { @@ -412,9 +412,9 @@ func TestStalledWarningEmitsNoticeAndDrainNote(t *testing.T) { waitFor(t, func() bool { for _, text := range sink.texts() { - // The claim, not the sentence: a job that has said nothing on its - // own streams, named so the reader knows which one. - if strings.Contains(text, "nothing to its own output") && strings.Contains(text, j.ID) { + // The claim, not the sentence: a job that has gone inactive, + // named so the reader knows which one. + if strings.Contains(text, "has been inactive for") && strings.Contains(text, j.ID) { return true } } @@ -424,7 +424,7 @@ func TestStalledWarningEmitsNoticeAndDrainNote(t *testing.T) { t.Fatalf("stalled job output status = %q ok=%v, want running", st, ok) } note := m.DrainCompletedNote() - if !strings.Contains(note, "nothing to its own output") || !strings.Contains(note, j.ID) { + if !strings.Contains(note, "has been inactive for") || !strings.Contains(note, j.ID) { t.Fatalf("stalled drain note = %q, want stalled update for %s", note, j.ID) } // The warning is once per job. diff --git a/internal/tools/jobs/liveness_test.go b/internal/tools/jobs/liveness_test.go new file mode 100644 index 0000000000..0d38b82233 --- /dev/null +++ b/internal/tools/jobs/liveness_test.go @@ -0,0 +1,133 @@ +package jobs + +import ( + "context" + "io" + "strings" + "testing" + "time" + + "reasonix/internal/contract/event" +) + +// A sub-agent's work reaches its transcript and tool calls, never its job's own +// streams. Judging liveness by bytes alone therefore calls a busy job silent for +// its whole life; what it reports instead is that it is still working. +func TestAJobThatWritesNothingStaysAliveWhileItReportsProgress(t *testing.T) { + sink := &recordingSink{} + m := NewManager(sink, WithStalledWarningAfter(60*time.Millisecond)) + defer m.Close() + + j := m.Start("task", "silent worker", func(ctx context.Context, _ io.Writer) (string, error) { + // Six ticks over ~180ms: three warning windows, none of them quiet + // enough to deserve a report. + for range 6 { + NoteProgress(ctx, 1) + time.Sleep(30 * time.Millisecond) + } + return "done", nil + }) + res, _ := m.Wait(context.Background(), []string{j.ID}, WaitOptions{Timeout: 5 * time.Second}) + if len(res) != 1 || res[0].Status != Done { + t.Fatalf("result = %+v, want one finished job", res) + } + for _, text := range sink.texts() { + if strings.Contains(text, "has been inactive for") { + t.Fatalf("a job reporting progress was reported inactive: %q", text) + } + } +} + +// The reading the caller gets must separate "wrote nothing" from "did nothing": +// a sub-agent has steps and no output text, and that pair is not silence. +func TestWaitReadsStepsSeparatelyFromOutputText(t *testing.T) { + m := NewManager(event.Discard) + defer m.Close() + + release := make(chan struct{}) + defer close(release) + j := m.Start("task", "busy", func(ctx context.Context, _ io.Writer) (string, error) { + NoteProgress(ctx, 3) + <-release + return "", nil + }) + defer m.Kill(j.ID) + + var got *Progress + waitFor(t, func() bool { + res, _ := m.Wait(context.Background(), []string{j.ID}, WaitOptions{Timeout: time.Millisecond}) + if len(res) == 1 && res[0].Progress != nil && res[0].Progress.Steps == 3 { + got = res[0].Progress + return true + } + return false + }) + if got.Produced != 0 || got.Spoke { + t.Fatalf("progress = %+v, want no output text: the job wrote none", got) + } + if got.Silent > time.Second { + t.Fatalf("Silent = %s, want recent activity from the reported steps", got.Silent) + } +} + +// A job that stops reporting is still reported as inactive, so the fix does not +// trade a false positive for a false negative. +func TestAJobThatStopsReportingIsStillCalledInactive(t *testing.T) { + sink := &recordingSink{} + m := NewManager(sink, WithStalledWarningAfter(40*time.Millisecond)) + defer m.Close() + + release := make(chan struct{}) + defer close(release) + j := m.Start("task", "stops", func(ctx context.Context, _ io.Writer) (string, error) { + NoteProgress(ctx, 2) // one step, then silence + <-release + return "", nil + }) + defer m.Kill(j.ID) + + waitFor(t, func() bool { + for _, text := range sink.texts() { + if strings.Contains(text, "has been inactive for") && strings.Contains(text, j.ID) { + return true + } + } + return false + }) + note := m.DrainCompletedNote() + if !strings.Contains(note, "2 step(s) were reported") { + t.Fatalf("drain note = %q, want the steps it reported before going quiet", note) + } +} + +// A cancelled job settles at once: reporting progress must not keep a killed +// job's status from being observable. +func TestKilledJobStopsBeingReportedAsRunning(t *testing.T) { + m := NewManager(event.Discard) + defer m.Close() + + j := m.Start("task", "cancel me", func(ctx context.Context, _ io.Writer) (string, error) { + for { + select { + case <-ctx.Done(): + return "", ctx.Err() + case <-time.After(5 * time.Millisecond): + NoteProgress(ctx, 1) + } + } + }) + time.Sleep(20 * time.Millisecond) + if !m.Kill(j.ID) { + t.Fatal("Kill reported the job was not running") + } + res, _ := m.Wait(context.Background(), []string{j.ID}, WaitOptions{Timeout: 2 * time.Second}) + if len(res) != 1 { + t.Fatalf("results = %+v, want one", res) + } + if res[0].Status != Killed { + t.Fatalf("status = %q, want %q", res[0].Status, Killed) + } + if res[0].Progress != nil { + t.Fatalf("progress = %+v, want none for a settled job", res[0].Progress) + } +} diff --git a/internal/tools/jobs/progress.go b/internal/tools/jobs/progress.go index 86c7d94dd5..93f91e68d8 100644 --- a/internal/tools/jobs/progress.go +++ b/internal/tools/jobs/progress.go @@ -18,6 +18,11 @@ type Progress struct { // differ here, and which one matters is the caller's to know. Produced int64 Delta int64 + // Steps is how many steps the runner reported; StepDelta is how many + // arrived while the caller waited. A sub-agent works in its own transcript, + // so steps without output text are progress, not silence. + Steps int64 + StepDelta int64 } func (m *Manager) results(targets []*Job) []Result { @@ -44,6 +49,7 @@ func (m *Manager) results(targets []*Job) []Result { Silent: sinceMs(j.times.activity), Spoke: j.written > 0, Produced: j.written, + Steps: j.times.steps, } } out = append(out, r) diff --git a/internal/tools/jobs/stalled_notice_test.go b/internal/tools/jobs/stalled_notice_test.go index 324cf764ae..a19529608f 100644 --- a/internal/tools/jobs/stalled_notice_test.go +++ b/internal/tools/jobs/stalled_notice_test.go @@ -13,7 +13,7 @@ func TestStalledJobEmitsCodedNoticeWithTheQuietSpan(t *testing.T) { m := NewManager(sink) defer m.Close() m.stalledWarning = 15 * time.Minute - m.recordStalled("", "bash-16", "bash", "make build") + m.recordStalled("", "bash-16", "bash", "make build", 0) sink.mu.Lock() defer sink.mu.Unlock() for _, ev := range sink.events { @@ -24,10 +24,33 @@ func TestStalledJobEmitsCodedNoticeWithTheQuietSpan(t *testing.T) { if !ok || p.Kind != "bash" || p.ID != "bash-16" || p.Label != "make build" || p.QuietSeconds != 900 || ev.Level != event.LevelWarn { t.Fatalf("stalled notice = %+v payload %+v", ev, p) } - if !strings.Contains(ev.Text, "has written nothing to its own output for 15m0s") || !event.DetailIsPayload(ev.Code) { + if !strings.Contains(ev.Text, "has been inactive for 15m0s") || !strings.Contains(ev.Text, "no progress has been reported") || !event.DetailIsPayload(ev.Code) { t.Fatalf("English fallback = %q, payload flag %v", ev.Text, event.DetailIsPayload(ev.Code)) } return } t.Fatalf("no %s notice in %+v", event.NoticeCodeJobStalled, sink.events) } + +// A job that reported steps and then went quiet is a different report from one +// that never reported any: the first was working and is now waiting, the second +// never got going, and the reader's next move differs. +func TestStalledNoticeSaysWhatTheJobHadReported(t *testing.T) { + sink := &recordingSink{} + m := NewManager(sink) + defer m.Close() + m.stalledWarning = 15 * time.Minute + m.recordStalled("", "task-7", "task", "audit", 42) + sink.mu.Lock() + defer sink.mu.Unlock() + for _, ev := range sink.events { + if ev.Code != event.NoticeCodeJobStalled { + continue + } + if !strings.Contains(ev.Text, "42 step(s) were reported") { + t.Fatalf("notice = %q, want the step count it reported", ev.Text) + } + return + } + t.Fatalf("no %s notice in %+v", event.NoticeCodeJobStalled, sink.events) +} diff --git a/internal/tools/jobs/watch.go b/internal/tools/jobs/watch.go index 4181ce39f6..0996247afc 100644 --- a/internal/tools/jobs/watch.go +++ b/internal/tools/jobs/watch.go @@ -112,6 +112,14 @@ func (j *Job) producedBytes() int64 { return j.written } +// reportedSteps is how many steps the runner had reported when it was asked. +// A caller compares two readings to see whether a job advanced without writing. +func (j *Job) reportedSteps() int64 { + j.mu.Lock() + defer j.mu.Unlock() + return j.times.steps +} + // WaitForSession blocks until the jobs finish, one of them writes something // opts.Match matches, the timeout expires, or ctx ends. Each result carries // what the job produced while this call was waiting, which is the only reading @@ -122,14 +130,17 @@ func (m *Manager) WaitForSession(ctx context.Context, parentSession string, ids return nil, WaitFinished } before := make(map[string]int64, len(targets)) + beforeSteps := make(map[string]int64, len(targets)) for _, j := range targets { before[j.ID] = j.producedBytes() + beforeSteps[j.ID] = j.reportedSteps() } finish := func(outcome WaitOutcome) ([]Result, WaitOutcome) { out := m.results(targets) for i := range out { if out[i].Progress != nil { out[i].Progress.Delta = maxInt64(0, out[i].Progress.Produced-before[out[i].ID]) + out[i].Progress.StepDelta = maxInt64(0, out[i].Progress.Steps-beforeSteps[out[i].ID]) } } return out, outcome