From 80bbdba617adf73891eaa8fa0da3d5bc81b081d2 Mon Sep 17 00:00:00 2001 From: Yichen Jiang Date: Fri, 4 Sep 2026 12:47:21 +0800 Subject: [PATCH] fix(jobs): restore the dropped-output notice with its spill files for the model The consuming shell read on master told the model where a truncated stream's complete text lived: [some output was dropped from memory; full output: ]. The ring-backed read rendered a notice only for the ring's own eviction and ignored the gapBefore mark the pump puts on a chunk whose source lost bytes between two polls, so a fast producer's tail read as continuous output and no path survived. A lossy source read now carries the source's spillPath onto its gap chunk (JobSourceRead, JobAppendOptions, JobChunk), the pump forwards it, the ring retains it, and the model renderer emits master's notice for a producer-side gap or a ring eviction, listing the distinct spill files in offset order. The bash and pwsh sources already returned the subprocess reader's spillPath; the type now admits it. --- docs/subsystems/jobs.i18n.yaml | 4 ++-- docs/subsystems/jobs.md | 6 ++++- docs/subsystems/jobs.zh.md | 6 ++++- packages/jobs/jobs-local/src/pump.ts | 5 +++- packages/jobs/jobs-local/src/ring.ts | 3 +++ packages/jobs/jobs-local/tests/pump.spec.ts | 22 +++++++++++++++-- packages/jobs/jobs-local/tests/ring.spec.ts | 4 ++-- packages/jobs/jobs/src/types.ts | 6 ++++- packages/jobs/jobs/src/view.ts | 2 ++ packages/jobs/tool-jobs/src/render.ts | 19 +++++++++------ .../jobs/tool-jobs/tests/tool-jobs.spec.ts | 24 +++++++++++++++---- .../shell/tool-bash/tests/background.spec.ts | 10 ++++++++ 12 files changed, 90 insertions(+), 21 deletions(-) diff --git a/docs/subsystems/jobs.i18n.yaml b/docs/subsystems/jobs.i18n.yaml index 381cd81097..c9cb38a0de 100644 --- a/docs/subsystems/jobs.i18n.yaml +++ b/docs/subsystems/jobs.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write docs/subsystems/jobs.md -jobs.md: e8884bae690188d8653fb02c0009be555a49f6f3 -jobs.zh.md: c75fae067e168a6be309febe888b7215e6edb591 +jobs.md: 8a13624c8269b63b18e80ec491f265c4bfcd0fc1 +jobs.zh.md: 39a1d4d0b42783154da96cd2a1515337c5a5cd1f diff --git a/docs/subsystems/jobs.md b/docs/subsystems/jobs.md index e8884bae69..8a13624c82 100644 --- a/docs/subsystems/jobs.md +++ b/docs/subsystems/jobs.md @@ -150,7 +150,7 @@ interface JobOutputSource { /** * Read everything captured since `fromByte` without consuming it. * @param fromByte - whole-stream offset to resume from (a prior read's `nextOffset`; 0 first). - * @returns the delta text, the next offset, and the lossy flag. + * @returns the delta text, the next offset, the lossy flag, and the spill path when one exists. */ read(fromByte: number): JobSourceRead } @@ -165,6 +165,8 @@ interface JobSourceRead { nextOffset: number /** True when the requested offset slid out of the source's retained window. */ lossy: boolean + /** Path to a file holding the complete stream, when the source keeps one and this read is lossy. */ + spillPath?: string } ``` @@ -183,6 +185,8 @@ interface JobChunk { readonly channel?: JobChannel /** Bytes immediately before this chunk were lost, at the producer or to retention. */ readonly gapBefore?: true + /** Host path of a file holding the complete stream before the gap, when the producer keeps one; present only with `gapBefore`. */ + readonly spillPath?: string } ``` diff --git a/docs/subsystems/jobs.zh.md b/docs/subsystems/jobs.zh.md index c75fae067e..39a1d4d0b4 100644 --- a/docs/subsystems/jobs.zh.md +++ b/docs/subsystems/jobs.zh.md @@ -150,7 +150,7 @@ interface JobOutputSource { /** * Read everything captured since `fromByte` without consuming it. * @param fromByte - whole-stream offset to resume from (a prior read's `nextOffset`; 0 first). - * @returns the delta text, the next offset, and the lossy flag. + * @returns the delta text, the next offset, the lossy flag, and the spill path when one exists. */ read(fromByte: number): JobSourceRead } @@ -165,6 +165,8 @@ interface JobSourceRead { nextOffset: number /** True when the requested offset slid out of the source's retained window. */ lossy: boolean + /** Path to a file holding the complete stream, when the source keeps one and this read is lossy. */ + spillPath?: string } ``` @@ -183,6 +185,8 @@ interface JobChunk { readonly channel?: JobChannel /** Bytes immediately before this chunk were lost, at the producer or to retention. */ readonly gapBefore?: true + /** Host path of a file holding the complete stream before the gap, when the producer keeps one; present only with `gapBefore`. */ + readonly spillPath?: string } ``` diff --git a/packages/jobs/jobs-local/src/pump.ts b/packages/jobs/jobs-local/src/pump.ts index e57cb02a09..ed6ae7a914 100644 --- a/packages/jobs/jobs-local/src/pump.ts +++ b/packages/jobs/jobs-local/src/pump.ts @@ -2,7 +2,9 @@ * The registry-owned pull pump: copies a job's {@link JobOutputSource}s into * its ring at a bounded cadence and drains them once more after the * producer settles, so the ring holds every byte before settlement trims and - * closes it. Pure utility — no cordis, no timers retained past settlement. + * closes it. A lossy source read lands as a gap chunk that names the source's + * spill file when it keeps one. Pure utility — no cordis, no timers retained + * past settlement. * @module @deepseek-ai/dsh-jobs-local/pump */ @@ -52,6 +54,7 @@ export function startPump( append(read.text, { ...source.channel !== undefined ? { channel: source.channel } : {}, ...read.lossy ? { gapBefore: true as const } : {}, + ...read.lossy && read.spillPath !== undefined ? { spillPath: read.spillPath } : {}, }) } } diff --git a/packages/jobs/jobs-local/src/ring.ts b/packages/jobs/jobs-local/src/ring.ts index fb4bcb3c5b..00e6c12096 100644 --- a/packages/jobs/jobs-local/src/ring.ts +++ b/packages/jobs/jobs-local/src/ring.ts @@ -14,6 +14,7 @@ interface RingChunk { bytes: number channel?: JobChannel gapBefore?: true + spillPath?: string } /** @@ -61,6 +62,7 @@ export class OutputRing { bytes, ...options?.channel !== undefined ? { channel: options.channel } : {}, ...options?.gapBefore !== undefined ? { gapBefore: options.gapBefore } : {}, + ...options?.spillPath !== undefined ? { spillPath: options.spillPath } : {}, }) this.total += bytes this.retainedBytes += bytes @@ -106,6 +108,7 @@ export class OutputRing { text: chunk.text, ...chunk.channel !== undefined ? { channel: chunk.channel } : {}, ...chunk.gapBefore !== undefined ? { gapBefore: chunk.gapBefore } : {}, + ...chunk.spillPath !== undefined ? { spillPath: chunk.spillPath } : {}, }) } return { chunks, next: this.total, lossy: from < this.earliest } diff --git a/packages/jobs/jobs-local/tests/pump.spec.ts b/packages/jobs/jobs-local/tests/pump.spec.ts index 31ee5d4d3f..c9580e4185 100644 --- a/packages/jobs/jobs-local/tests/pump.spec.ts +++ b/packages/jobs/jobs-local/tests/pump.spec.ts @@ -13,7 +13,7 @@ function sink() { /** A scripted source: each read() shifts the next scripted result. */ function scriptedSource( - reads: { text: string; lossy?: boolean }[], + reads: { text: string; lossy?: boolean; spillPath?: string }[], channel?: 'stdout' | 'stderr', ): { source: JobOutputSource; offsets: number[] } { const offsets: number[] = [] @@ -24,7 +24,12 @@ function scriptedSource( offsets.push(fromByte) const next = reads.shift() ?? { text: '' } offset += Buffer.byteLength(next.text, 'utf8') - return { text: next.text, nextOffset: offset, lossy: next.lossy ?? false } + return { + text: next.text, + nextOffset: offset, + lossy: next.lossy ?? false, + ...next.spillPath !== undefined ? { spillPath: next.spillPath } : {}, + } }, } return { source, offsets } @@ -66,6 +71,19 @@ describe('startPump', () => { expect(appends).toEqual([{ text: 'tail', options: { gapBefore: true } }]) }) + it("names the source's spill file on a lossy read and ignores one on a clean read", async () => { + const { appends, append } = sink() + const { source } = scriptedSource([ + { text: 'tail', lossy: true, spillPath: '/spill/out.log' }, + { text: 'more', spillPath: '/spill/out.log' }, + ], 'stdout') + await startPump([source], append, 1, Promise.resolve()).done + expect(appends).toEqual([ + { text: 'tail', options: { channel: 'stdout', gapBefore: true, spillPath: '/spill/out.log' } }, + { text: 'more', options: { channel: 'stdout' } }, + ]) + }) + it('treats a rejected settlement as settlement and still drains the final bytes', async () => { const { appends, append } = sink() const { source } = scriptedSource([{ text: 'last' }]) diff --git a/packages/jobs/jobs-local/tests/ring.spec.ts b/packages/jobs/jobs-local/tests/ring.spec.ts index 670e121d0b..da4aab5a63 100644 --- a/packages/jobs/jobs-local/tests/ring.spec.ts +++ b/packages/jobs/jobs-local/tests/ring.spec.ts @@ -6,14 +6,14 @@ describe('OutputRing', () => { const ring = new OutputRing() expect(ring.append('', undefined, 1024)).toBe(false) expect(ring.append('héllo', { channel: 'stdout' }, 1024)).toBe(true) - expect(ring.append('wörld', { channel: 'stderr', gapBefore: true }, 1024)).toBe(true) + expect(ring.append('wörld', { channel: 'stderr', gapBefore: true, spillPath: '/spill/err.log' }, 1024)).toBe(true) expect(ring.total).toBe(12) expect(ring.earliest).toBe(0) expect(ring.retainedBytes).toBe(12) expect(ring.readFrom(0)).toEqual({ chunks: [ { at: 0, text: 'héllo', channel: 'stdout' }, - { at: 6, text: 'wörld', channel: 'stderr', gapBefore: true }, + { at: 6, text: 'wörld', channel: 'stderr', gapBefore: true, spillPath: '/spill/err.log' }, ], next: 12, lossy: false, diff --git a/packages/jobs/jobs/src/types.ts b/packages/jobs/jobs/src/types.ts index b3abd3ac68..cfbfb49d92 100644 --- a/packages/jobs/jobs/src/types.ts +++ b/packages/jobs/jobs/src/types.ts @@ -38,6 +38,8 @@ export interface JobSourceRead { nextOffset: number /** True when the requested offset slid out of the source's retained window. */ lossy: boolean + /** Path to a file holding the complete stream, when the source keeps one and this read is lossy. */ + spillPath?: string } /** @@ -52,7 +54,7 @@ export interface JobOutputSource { /** * Read everything captured since `fromByte` without consuming it. * @param fromByte - whole-stream offset to resume from (a prior read's `nextOffset`; 0 first). - * @returns the delta text, the next offset, and the lossy flag. + * @returns the delta text, the next offset, the lossy flag, and the spill path when one exists. */ read(fromByte: number): JobSourceRead } @@ -67,6 +69,8 @@ export interface JobAppendOptions { * silent splice. */ gapBefore?: true + /** Path to a file holding the complete stream the gap can be recovered from; meaningful with `gapBefore`. */ + spillPath?: string } /** diff --git a/packages/jobs/jobs/src/view.ts b/packages/jobs/jobs/src/view.ts index b20e1c693b..ac38add1b5 100644 --- a/packages/jobs/jobs/src/view.ts +++ b/packages/jobs/jobs/src/view.ts @@ -46,6 +46,8 @@ export interface JobChunk { readonly channel?: JobChannel /** Bytes immediately before this chunk were lost, at the producer or to retention. */ readonly gapBefore?: true + /** Host path of a file holding the complete stream before the gap, when the producer keeps one; present only with `gapBefore`. */ + readonly spillPath?: string } /** diff --git a/packages/jobs/tool-jobs/src/render.ts b/packages/jobs/tool-jobs/src/render.ts index 74b0255cbd..fa379b6a81 100644 --- a/packages/jobs/tool-jobs/src/render.ts +++ b/packages/jobs/tool-jobs/src/render.ts @@ -63,18 +63,23 @@ export function statusLine(snapshot: Pick chunk.channel === undefined || chunk.channel === 'stdout').map(chunk => chunk.text).join('') - const err = chunks.filter(chunk => chunk.channel === 'stderr').map(chunk => chunk.text).join('') + const visible = chunks.filter(chunk => chunk.channel !== 'log') + const out = visible.filter(chunk => chunk.channel !== 'stderr').map(chunk => chunk.text).join('') + const err = visible.filter(chunk => chunk.channel === 'stderr').map(chunk => chunk.text).join('') const separator = out.length > 0 && !out.endsWith('\n') ? '\n' : '' const body = out + (err.length > 0 ? `${separator}[stderr]\n${err}` : '') - if (!lossy) return body - return `${body}${body.length > 0 && !body.endsWith('\n') ? '\n' : ''}[some output was dropped from memory before this read]` + const gaps = visible.filter(chunk => chunk.gapBefore === true) + if (!lossy && gaps.length === 0) return body + const paths = [...new Set(gaps.flatMap(chunk => chunk.spillPath === undefined ? [] : [chunk.spillPath]))] + const notice = `[some output was dropped from memory; full output: ${paths.length > 0 ? paths.join(', ') : '(unavailable)'}]` + return `${body}${body.length > 0 && !body.endsWith('\n') ? '\n' : ''}${notice}` } diff --git a/packages/jobs/tool-jobs/tests/tool-jobs.spec.ts b/packages/jobs/tool-jobs/tests/tool-jobs.spec.ts index ef1814d57e..e98a60554b 100644 --- a/packages/jobs/tool-jobs/tests/tool-jobs.spec.ts +++ b/packages/jobs/tool-jobs/tests/tool-jobs.spec.ts @@ -213,7 +213,7 @@ describe('tool-jobs setup', () => { it('renders a model delta as stdout, one stderr section, and a dropped-output notice', () => { expect(renderModelDelta([], false)).toBe('') - expect(renderModelDelta([], true)).toBe('[some output was dropped from memory before this read]') + expect(renderModelDelta([], true)).toBe('[some output was dropped from memory; full output: (unavailable)]') expect(renderModelDelta([ { at: 0, text: 'a' }, { at: 1, text: 'e1\n', channel: 'stderr' }, @@ -221,8 +221,24 @@ describe('tool-jobs setup', () => { { at: 13, text: 'b', channel: 'stdout' }, { at: 14, text: 'e2', channel: 'stderr' }, ], false)).toBe('ab\n[stderr]\ne1\ne2') - expect(renderModelDelta([{ at: 0, text: 'tail' }], true)).toBe('tail\n[some output was dropped from memory before this read]') - expect(renderModelDelta([{ at: 0, text: 'line\n' }], true)).toBe('line\n[some output was dropped from memory before this read]') + expect(renderModelDelta([{ at: 0, text: 'tail' }], true)).toBe('tail\n[some output was dropped from memory; full output: (unavailable)]') + expect(renderModelDelta([{ at: 0, text: 'line\n' }], true)).toBe('line\n[some output was dropped from memory; full output: (unavailable)]') + }) + + it('reports a producer-side gap as dropped output and names the spill files the gap chunks point at', () => { + // The ring itself lost nothing (`lossy: false`); the source did, between two pumps. + expect(renderModelDelta([{ at: 0, text: 'head' }, { at: 4, text: 'tail', gapBefore: true, spillPath: '/spill/out.log' }], false)) + .toBe('headtail\n[some output was dropped from memory; full output: /spill/out.log]') + // Distinct spill files list once each, in offset order; a gap without a file yields the generic notice. + expect(renderModelDelta([ + { at: 0, text: 'o', gapBefore: true, spillPath: '/spill/out.log' }, + { at: 1, text: 'e', channel: 'stderr', gapBefore: true, spillPath: '/spill/err.log' }, + { at: 2, text: 'o2', gapBefore: true, spillPath: '/spill/out.log' }, + ], false)).toBe('oo2\n[stderr]\ne\n[some output was dropped from memory; full output: /spill/out.log, /spill/err.log]') + expect(renderModelDelta([{ at: 0, text: 'tail', gapBefore: true }], false)) + .toBe('tail\n[some output was dropped from memory; full output: (unavailable)]') + // A gap on observer-only narration never reaches the model. + expect(renderModelDelta([{ at: 0, text: 'phase', channel: 'log', gapBefore: true, spillPath: '/spill/log' }], false)).toBe('') }) it('applies the built-in wait bounds when apply() receives a bare config', async () => { @@ -288,7 +304,7 @@ describe('job_output', () => { ctx.jobs.start(p.spec) p.append('x'.repeat(40)) const first = text(await call(ctx, 'job_output', { job_id: 'bash-1' })) - expect(first).toContain('[some output was dropped from memory before this read]') + expect(first).toContain('[some output was dropped from memory; full output: (unavailable)]') expect(first).toContain('[status: running]') expect(first).not.toContain('x'.repeat(17)) // The cursor now sits at the ring's end: the next read is clean. diff --git a/packages/shell/tool-bash/tests/background.spec.ts b/packages/shell/tool-bash/tests/background.spec.ts index 5862887118..535326cd77 100644 --- a/packages/shell/tool-bash/tests/background.spec.ts +++ b/packages/shell/tool-bash/tests/background.spec.ts @@ -144,6 +144,16 @@ describe('processSources', () => { // A missing stderr reader is an absent stream, not an error. expect(stderr!.read(0)).toEqual({ text: '', nextOffset: 0, lossy: false }) }) + + it("passes a lossy read's spill file through, so the model's notice can name it", () => { + const proc = { + observed: { + stderr: { readFrom: (from: number) => ({ text: 'tail', nextOffset: from + 4, lossy: true, spillPath: '/spill/err.log' }) }, + }, + } as unknown as ShellProcess + const [, stderr] = processSources(() => proc) + expect(stderr!.read(0)).toEqual({ text: 'tail', nextOffset: 4, lossy: true, spillPath: '/spill/err.log' }) + }) }) describe('owned background output', () => {