diff --git a/desktop/frontend-next/src/i18n/en.ts b/desktop/frontend-next/src/i18n/en.ts index 813f068e9d..92a922ca72 100644 --- a/desktop/frontend-next/src/i18n/en.ts +++ b/desktop/frontend-next/src/i18n/en.ts @@ -1825,8 +1825,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 9a93e7d6cd..1e2dc924eb 100644 --- a/internal/tools/builtin/bgjobs.go +++ b/internal/tools/builtin/bgjobs.go @@ -318,21 +318,23 @@ func clampWaitSeconds(requested int) (seconds int, capped bool) { // 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