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()

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

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' }

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.

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.
                 }
               },
             })

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

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 }
+    },
+  }
+}

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()),
             }
           },
         })

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)