Merge pull request #14278 from tongsailong/fix/subagent-progress

fix(delegation): track background subagent progress
This commit is contained in:
YHH authored and GitHub committed 2026-10-23 05:12:26 -07:00
commit 5e8d0d704a
19 files changed
+612 -42

No files matched your search

+4 -2
View File
@@ -1825,8 +1825,10 @@ export const EN: Record<string, string> = {
"后台任务已结束:{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",
+13 -7
View File
@@ -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;
}
+1 -1
View File
@@ -43,7 +43,7 @@ export const NOTICE_TEXT: Record<string, string> = {
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}",
@@ -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", () => {
+5 -1
View File
@@ -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.
@@ -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
}
@@ -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")
}
}
@@ -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)
}
}
@@ -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 := &noteRecorder{}
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 := &noteRecorder{}
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")
}
@@ -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
+12 -10
View File
@@ -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 {
+54
View File
@@ -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)
}
}
})
}
}
+4
View File
@@ -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.
+43 -8
View File
@@ -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
+5 -5
View File
@@ -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.
+133
View File
@@ -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)
}
}
+6
View File
@@ -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)
+25 -2
View File
@@ -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)
}
+11
View File
@@ -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