From 10d76fbd03ee00cd478ef86fbcc3c75c2686ab0a Mon Sep 17 00:00:00 2001 From: Yichen Jiang Date: Thu, 3 Sep 2026 16:37:35 +0800 Subject: [PATCH] refactor(producers): subagent and terminal jobs use session owners, results, and pull sources - tool-subagent registers the parent session id as the owner and settleRun reports the child's final text as the job result, which the first read after settlement carries once - tool-terminal hands the registry one pull source over the backend's consuming send reader (sendSource in background.ts) instead of a readOutput hook; the truncation marker stays in the delivered text - the model's kill reason now shows in a killed job's status line, as the registry records it as the terminal detail - test call sites move from jobs.list(owner) to visibleTo(owner).list() --- .../tests/profiles/headless/tests/ptc.e2e.ts | 10 +++--- .../subagent/subagent/src/run-settlement.ts | 2 +- .../subagent/tests/run-settlement.spec.ts | 4 +-- packages/subagent/tool-subagent/src/index.ts | 4 +-- .../tool-subagent/tests/tool-subagent.spec.ts | 15 ++++----- .../terminal/tool-terminal/src/background.ts | 31 +++++++++++++++++++ packages/terminal/tool-terminal/src/index.ts | 17 +++++----- .../tool-terminal/tests/tools.spec.ts | 22 +++++++++++-- 8 files changed, 79 insertions(+), 26 deletions(-) create mode 100644 packages/terminal/tool-terminal/src/background.ts diff --git a/apps/cli/tests/profiles/headless/tests/ptc.e2e.ts b/apps/cli/tests/profiles/headless/tests/ptc.e2e.ts index 8ee3affb8a..25a7292e45 100644 --- a/apps/cli/tests/profiles/headless/tests/ptc.e2e.ts +++ b/apps/cli/tests/profiles/headless/tests/ptc.e2e.ts @@ -222,7 +222,7 @@ describe('PTC mode typed values: keyless real-worker contracts', () => { return await tools.bash({ command: 'sleep 10', description: 'Must never start', run_in_background: true }); `, pre.signal) expect(preResult.isError).toBe(true) - expect(ctx.jobs.list()).toEqual([]) + expect(ctx.jobs.visibleTo().list()).toEqual([]) const afterPublication = new AbortController() const running = runCode(ctx, ` @@ -230,14 +230,14 @@ describe('PTC mode typed values: keyless real-worker contracts', () => { console.log(started.jobId); await new Promise(() => {}); `, afterPublication.signal) - for (let attempt = 0; attempt < 100 && ctx.jobs.list().length === 0; attempt++) { + for (let attempt = 0; attempt < 100 && ctx.jobs.visibleTo().list().length === 0; attempt++) { await new Promise(resolve => setTimeout(resolve, 10)) } - const job = ctx.jobs.list()[0] + const job = ctx.jobs.visibleTo().list()[0] expect(job).toMatchObject({ id: 'bash-1', status: 'running' }) afterPublication.abort('outer-call-cancelled') expect((await running).isError).toBe(true) - expect(ctx.jobs.list()[0]).toMatchObject({ id: job!.id, status: 'running' }) + expect(ctx.jobs.visibleTo().list()[0]).toMatchObject({ id: job!.id, status: 'running' }) const killed = completion(await runCode(ctx, ` return await tools.job_kill({ job_id: ${JSON.stringify(job!.id)}, reason: 'test owns cancellation' }); @@ -261,7 +261,7 @@ describe('PTC mode typed values: keyless real-worker contracts', () => { const result = await pending expect(result.isError).toBe(true) expect(Date.now() - startedAt).toBeLessThan(5_000) - expect(ctx.jobs.list()).toEqual([]) + expect(ctx.jobs.visibleTo().list()).toEqual([]) }, 15_000) it('uses versioned Cordis DTO ids directly for running and pending Plugins, then confirms removal', async () => { diff --git a/packages/subagent/subagent/src/run-settlement.ts b/packages/subagent/subagent/src/run-settlement.ts index 02f61a7baf..34baacbcd4 100644 --- a/packages/subagent/subagent/src/run-settlement.ts +++ b/packages/subagent/subagent/src/run-settlement.ts @@ -37,7 +37,7 @@ function failureDetail(result: SubagentResult): string { function runOutcome(result: SubagentResult): JobOutcome { switch (result.stopReason) { case 'completed': - return { status: 'completed', output: finalText(result.output) } + return { status: 'completed', result: finalText(result.output) } case 'aborted': return result.diagnostic === undefined ? { status: 'killed' } diff --git a/packages/subagent/subagent/tests/run-settlement.spec.ts b/packages/subagent/subagent/tests/run-settlement.spec.ts index 146181d43a..20fc514504 100644 --- a/packages/subagent/subagent/tests/run-settlement.spec.ts +++ b/packages/subagent/subagent/tests/run-settlement.spec.ts @@ -9,7 +9,7 @@ const MAX_SUBAGENT_DIAGNOSTIC_BYTES = 4_096 describe('outcome mapping helpers', () => { it.each([ - ['completed', { status: 'completed', output: 'partial' }], + ['completed', { status: 'completed', result: 'partial' }], ['aborted', { status: 'killed' }], ['error', { status: 'failed', detail: 'error' }], ['max-tokens', { status: 'failed', detail: 'max-tokens' }], @@ -34,7 +34,7 @@ describe('outcome mapping helpers', () => { dispose() { order.push('dispose'); return Promise.resolve() }, }) order.push('reported') - expect(completed).toEqual({ status: 'completed', output: 'ok' }) + expect(completed).toEqual({ status: 'completed', result: 'ok' }) expect(order).toEqual(['dispose', 'reported']) // An infrastructure rejection still disposes and reports failed. diff --git a/packages/subagent/tool-subagent/src/index.ts b/packages/subagent/tool-subagent/src/index.ts index 03b56a3c9d..2c87bbf213 100644 --- a/packages/subagent/tool-subagent/src/index.ts +++ b/packages/subagent/tool-subagent/src/index.ts @@ -537,7 +537,7 @@ export function apply(ctx: Context, config: Config): void { const id = jobs.start({ kind: 'subagent', label: args.description, - owner: parent, + owner: parent.id, run: () => { const controller = new AbortController() const start = runtimeCtx.subagents.start(config.provider, { ...request, signal: controller.signal }) @@ -546,7 +546,7 @@ export function apply(ctx: Context, config: Config): void { controller.abort(reason ?? 'background subagent task killed') }, done: settleStart(start, controller.signal), - // No readOutput: the child session owns intermediate detail. + // No output sources: the child session owns intermediate detail. } }, }) diff --git a/packages/subagent/tool-subagent/tests/tool-subagent.spec.ts b/packages/subagent/tool-subagent/tests/tool-subagent.spec.ts index e5f9502a43..c1b2046182 100644 --- a/packages/subagent/tool-subagent/tests/tool-subagent.spec.ts +++ b/packages/subagent/tool-subagent/tests/tool-subagent.spec.ts @@ -883,7 +883,7 @@ describe('dsh-tool-subagent background mode', () => { }) expect(text(collected)).toBe('background answer\n[status: completed]') - // Final-output reads are idempotent (not consumed). + // The result rides the first read after settlement; a later read carries only the status. const again = await ctx.tools.execute({ signal: testToolSignal, callId: ToolCallId('collect-2'), @@ -891,7 +891,7 @@ describe('dsh-tool-subagent background mode', () => { arguments: { job_id: 'subagent-1' }, agent: parent, }) - expect(text(again)).toBe('background answer\n[status: completed]') + expect(text(again)).toBe('(no new output)\n[status: completed]') }) it('preserves provider diagnostics in one-shot background failure detail', async () => { @@ -972,7 +972,7 @@ describe('dsh-tool-subagent background mode', () => { const result = await resultPromise expect(result.isError).toBe(true) - expect(ctx.jobs.list(parent)).toEqual([]) + expect(ctx.jobs.visibleTo(parent.id).list()).toEqual([]) }) it('rejects startup when the provider changes during asynchronous route preflight', async () => { @@ -1082,7 +1082,8 @@ describe('dsh-tool-subagent background mode', () => { arguments: { job_id: 'subagent-1', wait: true }, agent: parent, }) - expect(text(output)).toBe('(no new output)\n[status: killed]') + // The registry records the model's kill reason as the terminal detail. + expect(text(output)).toBe('(no new output)\n[status: killed, no longer needed]') }) it('reports startup rollback failure after cancellation as a failed job', async () => { @@ -1169,7 +1170,7 @@ describe('dsh-tool-subagent background mode', () => { // The aborted children settle as killed tasks. const killed = await ctx.tools.execute({ signal: testToolSignal, callId: ToolCallId('w1'), name: 'job_output', arguments: { job_id: 'subagent-1', wait: true }, agent: parent }) - expect(text(killed)).toBe('(no new output)\n[status: killed]') + expect(text(killed)).toBe('(no new output)\n[status: killed, superseded]') }) }) @@ -1240,7 +1241,7 @@ describe('dsh-tool-subagent continuable background mode', () => { expect(match).not.toBeNull() const [, childId] = match! // No Task was created for the continuable child. - expect(ctx.jobs.list(parent)).toEqual([]) + expect(ctx.jobs.visibleTo(parent.id).list()).toEqual([]) await vi.waitFor(() => { expect(ctx.agents.get(SessionId(childId!))).toBeUndefined() @@ -1271,7 +1272,7 @@ describe('dsh-tool-subagent continuable background mode', () => { if (result.isError) throw new Error('expected foreground subagent success') expect(result.value).toMatchObject({ kind: 'foreground' }) expect(text(result)).toBe('continuable answer') - expect(ctx.jobs.list(parent)).toEqual([]) + expect(ctx.jobs.visibleTo(parent.id).list()).toEqual([]) }) it('isolates a cancelled continuable preparation from a concurrent sibling', async () => { diff --git a/packages/terminal/tool-terminal/src/background.ts b/packages/terminal/tool-terminal/src/background.ts new file mode 100644 index 0000000000..47494f7cff --- /dev/null +++ b/packages/terminal/tool-terminal/src/background.ts @@ -0,0 +1,31 @@ +/** + * Generic-job adaptation for background terminal sends: the registry pull + * source over the backend's consuming send reader. + * + * @module @deepseek-ai/dsh-tool-terminal/background + */ + +import type { JobOutputSource } from '@deepseek-ai/dsh-jobs' +import type { TerminalSendOperation } from '@deepseek-ai/dsh-terminal' +import { renderSendRead } from './render.ts' + +/** + * Adapt the backend's consuming send reader to a registry pull source. The + * reader has no offsets of its own — each call hands over what arrived since + * the previous one — so the source counts the bytes it delivered to keep the + * registry's cursor monotonic. It binds lazily because the send starts inside + * the job starter, after the registry admitted the job; a read before that + * point delivers nothing. + * @param operation - the live send, once the starter has begun it. + * @returns one unlabeled source over the send's bounded delta. + */ +export function sendSource(operation: () => TerminalSendOperation | undefined): JobOutputSource { + return { + read: (fromByte) => { + const live = operation() + if (live === undefined) return { text: '', nextOffset: fromByte, lossy: false } + const text = renderSendRead(live.readOutput()) + return { text, nextOffset: fromByte + Buffer.byteLength(text), lossy: false } + }, + } +} diff --git a/packages/terminal/tool-terminal/src/index.ts b/packages/terminal/tool-terminal/src/index.ts index 880cfcd31f..417c1537c4 100644 --- a/packages/terminal/tool-terminal/src/index.ts +++ b/packages/terminal/tool-terminal/src/index.ts @@ -9,11 +9,12 @@ import z from '@deepseek-ai/schemastery' import type { Agent } from '@deepseek-ai/dsh-agent' import type { ContentBlock } from '@deepseek-ai/dsh-llm' import { TerminalSessionId } from '@deepseek-ai/dsh-terminal' -import type { TerminalSendResult, TerminalSessionId as TerminalSessionIdType, TerminalSignal } from '@deepseek-ai/dsh-terminal' +import type { TerminalSendOperation, TerminalSendResult, TerminalSessionId as TerminalSessionIdType, TerminalSignal } from '@deepseek-ai/dsh-terminal' import type {} from '@deepseek-ai/dsh-jobs' import { defineTool } from '@deepseek-ai/dsh-tools' import type { ToolDefinition } from '@deepseek-ai/dsh-tools' -import { boundTerminalText, renderList, renderRead, renderSend, renderSendRead, renderSpawn } from './render.ts' +import { sendSource } from './background.ts' +import { boundTerminalText, renderList, renderRead, renderSend, renderSpawn } from './render.ts' declare module '@deepseek-ai/dsh-jobs' { interface JobKindMap { @@ -252,23 +253,25 @@ export function apply(ctx: Context, config: Config = {}): void { const jobs = ctx.get('jobs') if (jobs === undefined) throw new Error('background terminal sends require @deepseek-ai/dsh-jobs and @deepseek-ai/dsh-tool-jobs') let cancelRequested = false + let operation: TerminalSendOperation | undefined const jobId = jobs.start({ kind: 'pty-send', label: `${id}: ${args.text || '(input)'}`, - owner, + owner: owner.id, outputLimitBytes: maxResultBytes, + output: [sendSource(() => operation)], run: () => { - const operation = ctx.terminals.startSend(owner, id, request) + const started = ctx.terminals.startSend(owner, id, request) + operation = started return { cancel: () => { cancelRequested = true - operation.cancel() + started.cancel() }, - done: operation.done.then( + done: started.done.then( result => ({ status: cancelRequested ? 'killed' as const : 'completed' as const, detail: sendDetail(result) }), (error: unknown) => ({ status: 'failed' as const, detail: String(error) }), ), - readOutput: () => renderSendRead(operation.readOutput()), } }, }) diff --git a/packages/terminal/tool-terminal/tests/tools.spec.ts b/packages/terminal/tool-terminal/tests/tools.spec.ts index 15151f7bd2..eb80741bf8 100644 --- a/packages/terminal/tool-terminal/tests/tools.spec.ts +++ b/packages/terminal/tool-terminal/tests/tools.spec.ts @@ -8,10 +8,11 @@ import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRuntime, { renderToolsSdk } from '@deepseek-ai/dsh-tools' import type { ToolSdkSchema } from '@deepseek-ai/dsh-tools/src/ts-types.ts' import TerminalSessionService, { TerminalSessionId } from '@deepseek-ai/dsh-terminal' -import type { TerminalBackend, TerminalBackendSession, TerminalSendOperation, TerminalSendRequest, TerminalSessionStatus, TerminalSignal } from '@deepseek-ai/dsh-terminal' +import type { TerminalBackend, TerminalBackendSession, TerminalSendOperation, TerminalSendRead, TerminalSendRequest, TerminalSessionStatus, TerminalSignal } from '@deepseek-ai/dsh-terminal' import LocalJobRegistry from '@deepseek-ai/dsh-jobs-local' import * as ToolTasks from '@deepseek-ai/dsh-tool-jobs' import * as ToolPty from '@deepseek-ai/dsh-tool-terminal' +import { sendSource } from '../src/background.ts' function fakeAgent(ctx: Context, rawId: string): Agent { const scope = ctx.plugin(() => {}) @@ -54,7 +55,11 @@ class StubSession implements TerminalBackendSession { })) const operation: TerminalSendOperation = { done, - readOutput: () => ({ delta: this.delta, truncated: this.deltaTruncated }), + readOutput: () => { + const delta = this.delta + this.delta = '' + return { delta, truncated: this.deltaTruncated } + }, cancel: () => { if (cancelled) return false cancelled = true @@ -402,6 +407,19 @@ describe('tool-terminal foreground API', () => { }) }) +describe('sendSource', () => { + it('reads nothing before the send starts and counts delivered bytes afterwards', () => { + const reads: TerminalSendRead[] = [{ delta: '界', truncated: false }, { delta: '', truncated: true }] + const send: { operation?: TerminalSendOperation } = {} + const source = sendSource(() => send.operation) + expect(source.read(0)).toEqual({ text: '', nextOffset: 0, lossy: false }) + send.operation = { readOutput: () => reads.shift()!, cancel: () => false, done: new Promise(() => {}) } + expect(source.read(0)).toEqual({ text: '界', nextOffset: 3, lossy: false }) + // A truncated delta carries the same marker the foreground read shows. + expect(source.read(3)).toEqual({ text: '[output truncated]', nextOffset: 21, lossy: false }) + }) +}) + describe('tool-terminal task integration', () => { it('registers a generic task and exposes incremental output', async () => { const { ctx, agent } = await setup(true)