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', () => {