refactor(tool-jobs): render model reads from the ring and own the notice ledger

- job_output reads the per-job ring through visibleTo(): stdout, then one
  [stderr] section, log narration excluded, a dropped-output notice on a
  lossy read, and the result exactly once after settlement
- completion notices come from the registry event stream subscribed with
  { owners: 'scope' }; a wait claims its job when it starts and withdraws
  on timeout or abort, job_kill claims on 'requested', teardown settlements
  and unowned jobs are skipped, and a removal drops the ledger entry
- the notice destination is the agent registered for the owner session at
  settlement, so tool-jobs now injects the agent registry
- JobView exposes the producer's outputLimitBytes for the tool-result bound
- lint: hub disposer identity comment, unknown-typed matchers in seam tests

packages/jobs/jobs-local/src/events.ts

@@ -62,10 +62,12 @@ export class JobEventHub {
     if ('owners' in filter && filter.owners === 'scope') {
       return this.layers.effect(ctx, layer => layer.scoped.append(subscription), { label: 'jobs.events.subscribe()' })
     }
-    return ctx.effect(() => {
+    const dispose = ctx.effect(() => {
       this.unscoped.add(subscription)
       return () => { this.unscoped.delete(subscription) }
     }, 'jobs.events.subscribe()')
+    // oxlint-disable-next-line typescript/no-misused-promises -- exact synchronous disposer preserves Cordis effect identity
+    return dispose
   }
 
   /**

packages/jobs/jobs-local/src/index.ts

@@ -177,7 +177,7 @@ export class LocalJobRegistry extends JobRegistry {
     this.settledRetainBytes = resolved.settledRetainBytes
     this.pumpPollMs = resolved.pumpPollMs
     this.selfCtx = ctx
-    this.hub = new JobEventHub(this.layers, message => ctx.logger.warn(message))
+    this.hub = new JobEventHub(this.layers, (message) => { ctx.logger.warn(message) })
     ctx.effect(() => () => this.disposeAll(), 'jobs teardown')
   }
 

packages/jobs/jobs-local/tests/jobs.spec.ts

@@ -395,7 +395,7 @@ describe('LocalJobRegistry reads and settlement', () => {
     p.settle({ status: 'completed', detail: 'exit code: 0' })
     await tick()
 
-    expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id, status: 'completed', detail: 'exit code: 0' }) }])
+    expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id, status: 'completed', detail: 'exit code: 0' }) as unknown }])
     expect(warn).toHaveBeenCalledWith(expect.stringContaining('listener boom'))
   })
 
@@ -440,14 +440,14 @@ describe('LocalJobRegistry.kill', () => {
     expect(jobsOf(ctx).kill(id, { reason: 'no longer needed' })).toBe('requested')
     expect(p.cancels).toEqual(['no longer needed'])
     expect(jobsOf(ctx).list()[0]).toMatchObject({ status: 'stopping' })
-    expect(seen).toEqual([{ type: 'stopping', job: expect.objectContaining({ id, status: 'stopping' }) }])
+    expect(seen).toEqual([{ type: 'stopping', job: expect.objectContaining({ id, status: 'stopping' }) as unknown }])
 
     p.settle({ status: 'killed', detail: 'signal: SIGTERM' })
     await tick()
     expect(seen[1]).toEqual({
       type: 'settled',
       cause: 'kill',
-      job: expect.objectContaining({ id, status: 'killed', detail: 'signal: SIGTERM; no longer needed' }),
+      job: expect.objectContaining({ id, status: 'killed', detail: 'signal: SIGTERM; no longer needed' }) as unknown,
     })
   })
 
@@ -495,7 +495,7 @@ describe('LocalJobRegistry.kill', () => {
     expect(seen).toEqual([])
     settle({ status: 'completed' })
     await tick()
-    expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id }) }])
+    expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id }) as unknown }])
 
     broken = false
     expect(jobsOf(ctx).kill(id)).toBe('already-finished')
@@ -587,7 +587,7 @@ describe('LocalJobRegistry.wait', () => {
     p.settle({ status: 'completed', detail: 'exit code: 0' })
     controller.abort()
     await expect(wait).rejects.toThrow('wait aborted')
-    expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id, status: 'completed' }) }])
+    expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id, status: 'completed' }) as unknown }])
   })
 
   it('an abort landing after settlement still delivers the terminal projection it owes', async () => {
@@ -751,7 +751,7 @@ describe('LocalJobRegistry owner cleanup', () => {
     })
 
     await disposeAgentScope(owner)
-    expect(seen).toEqual([{ type: 'settled', cause: 'teardown', job: expect.objectContaining({ status: 'killed' }) }])
+    expect(seen).toEqual([{ type: 'settled', cause: 'teardown', job: expect.objectContaining({ status: 'killed' }) as unknown }])
   })
 
   it('attaches one cleanup per owner and drains all owned jobs with the scope', async () => {
@@ -933,7 +933,7 @@ describe('LocalJobRegistry disposal', () => {
 
     expect(disposedWithoutProducerDone).toBe(true)
     expect(warn).toHaveBeenCalledWith(expect.stringContaining('work may be orphaned'))
-    expect(seen).toEqual([{ type: 'settled', cause: 'teardown', job: expect.objectContaining({ status: 'failed' }) }])
+    expect(seen).toEqual([{ type: 'settled', cause: 'teardown', job: expect.objectContaining({ status: 'failed' }) as unknown }])
   })
 
   it('detaches owner effects from still-live agent scopes when the service unloads', async () => {
@@ -1012,14 +1012,14 @@ describe('LocalJobRegistry events', () => {
     const p = producer({ owner })
     const id = ctx.jobs.start(p.spec)
     // Registration is announced only once the record is readable.
-    expect(seen).toEqual([{ type: 'registered', job: expect.objectContaining({ id, owner: 'alice', status: 'running' }) }])
+    expect(seen).toEqual([{ type: 'registered', job: expect.objectContaining({ id, owner: 'alice', status: 'running' }) as unknown }])
     expect(jobsOf(ctx, owner).list()).toHaveLength(1)
 
     p.job().updateProgress('3/10')
-    expect(seen[1]).toEqual({ type: 'progress', job: expect.objectContaining({ id, progress: '3/10' }) })
+    expect(seen[1]).toEqual({ type: 'progress', job: expect.objectContaining({ id, progress: '3/10' }) as unknown })
 
     expect(jobsOf(ctx, owner).kill(id)).toBe('requested')
-    expect(seen[2]).toEqual({ type: 'stopping', job: expect.objectContaining({ id, status: 'stopping', progress: '3/10' }) })
+    expect(seen[2]).toEqual({ type: 'stopping', job: expect.objectContaining({ id, status: 'stopping', progress: '3/10' }) as unknown })
 
     p.settle({ status: 'killed' })
     await tick()
@@ -1033,7 +1033,7 @@ describe('LocalJobRegistry events', () => {
     const ctx = await harness()
     const seen = collect(ctx, { owner: SessionId('anyone') }, ['registered'])
     ctx.jobs.start(producer().spec)
-    expect(seen).toEqual([{ type: 'registered', job: expect.objectContaining({ id: 'bash-1' }) }])
+    expect(seen).toEqual([{ type: 'registered', job: expect.objectContaining({ id: 'bash-1' }) as unknown }])
     expect((seen[0] as { job: JobView }).job.owner).toBeUndefined()
   })
 

packages/jobs/jobs/src/view.ts

@@ -62,6 +62,8 @@ export interface JobView {
   readonly label: string
   /** Owning session; absent for an unowned job, which every caller can see. */
   readonly owner?: SessionId
+  /** Producer-owned cap for complete model-facing notices and reads, in UTF-8 bytes. */
+  readonly outputLimitBytes?: number
   /** Current lifecycle state. */
   readonly status: JobStatus
   /** The producer's live progress line (`3/10`, the current phase); cleared at settlement. */

packages/jobs/jobs/tests/service.spec.ts

@@ -77,7 +77,7 @@ describe('JobRegistry seam', () => {
     expect(jobs.list()).toHaveLength(1)
     expect(jobs.get(id).status).toBe('running')
     expect(jobs.get(id).owner).toBe(caller)
-    expect(jobs.read(id)).toEqual({ chunks: [], lossy: false, job: expect.objectContaining({ id }) })
+    expect(jobs.read(id)).toEqual({ chunks: [], lossy: false, job: expect.objectContaining({ id }) as unknown })
     expect(jobs.readAt(id, 7)).toEqual({ chunks: [], next: 7, lossy: false })
     expect(jobs.kill(id, { reason: 'seam test' })).toBe('requested')
     await expect(jobs.wait(id, 5)).resolves.toMatchObject({ id })

packages/jobs/tool-jobs/src/index.ts

@@ -1,9 +1,10 @@
 /**
  * Model-facing `job_output`, `job_list`, and `job_kill` tools over
  * `ctx.jobs`. Loading the plugin attaches the controller required by
- * producers. It also delivers unreported completions to the owning agent:
- * injected into a busy owner's next step, or opening a turn on an idle one
- * under the default `wakeup` delivery, bounded per owner.
+ * producers. It also delivers completions the model has not already
+ * collected to the owning agent: injected into a busy owner's next step, or
+ * opening a turn on an idle one under the default `wakeup` delivery, bounded
+ * per owner.
  * @module @deepseek-ai/dsh-tool-jobs
  */
 
@@ -14,14 +15,19 @@ import { TextRetainer } from '@deepseek-ai/dsh-output-retention'
 import { defineTool } from '@deepseek-ai/dsh-tools'
 import type { GenericCallView, ToolDefinition, ToolExecution } from '@deepseek-ai/dsh-tools'
 import { JobId } from '@deepseek-ai/dsh-jobs'
-import type { JobSnapshot } from '@deepseek-ai/dsh-jobs'
+import type { JobView, VisibleJobs } from '@deepseek-ai/dsh-jobs'
 import type { Agent } from '@deepseek-ai/dsh-agent'
+import { jobDetail, publicJob, renderModelDelta, statusLine } from './render.ts'
+import type { PublicJobSnapshot } from './render.ts'
+
+export { publicJob, renderModelDelta, statusLine } from './render.ts'
+export type { PublicJobSnapshot } from './render.ts'
 
 export const name = 'tool-jobs'
-export const inject = ['tools', 'jobs', 'systemPrompt']
+export const inject = ['tools', 'jobs', 'agents', 'systemPrompt']
 
 /**
- * How an unreported completion reaches an owner that is already idle: `wakeup`
+ * How an uncollected completion reaches an owner that is already idle: `wakeup`
  * opens a turn for it, `quiet` leaves it pending until something else wakes the
  * owner. A busy owner is injected either way.
  */
@@ -51,17 +57,6 @@ export const Config: z<Config> = z.object({
   maxConsecutiveWakes: z.number().min(1).default(3),
 })
 
-/** Task state safe for model-authored programs; ownership/bookkeeping fields are omitted. */
-export interface PublicJobSnapshot {
-  id: string
-  kind: string
-  label: string
-  status: JobSnapshot['status']
-  detail?: string
-  startedAt: number
-  finishedAt?: number
-}
-
 /** Shared schema for job-control outputs. */
 const PUBLIC_TASK_SCHEMA = {
   type: 'object',
@@ -81,30 +76,6 @@ const PUBLIC_TASK_SCHEMA = {
   },
 } as const
 
-/** Remove job ownership and notification bookkeeping from a registry snapshot. */
-function publicJob(snapshot: JobSnapshot): PublicJobSnapshot {
-  return {
-    id: snapshot.id,
-    kind: snapshot.kind,
-    label: snapshot.label,
-    status: snapshot.status,
-    ...snapshot.detail !== undefined ? { detail: snapshot.detail } : {},
-    startedAt: snapshot.startedAt,
-    ...snapshot.finishedAt !== undefined ? { finishedAt: snapshot.finishedAt } : {},
-  }
-}
-
-/**
- * Render generic status with optional producer detail.
- * @param snapshot - job state to render.
- * @returns a bracketed status line.
- */
-export function statusLine(snapshot: Pick<JobSnapshot, 'status' | 'detail'>): string {
-  return snapshot.detail !== undefined
-    ? `[status: ${snapshot.status}, ${snapshot.detail}]`
-    : `[status: ${snapshot.status}]`
-}
-
 const encoder = new TextEncoder()
 
 function retainTail(text: string, maxBytes: number): string {
@@ -135,19 +106,19 @@ function fitWithSuffix(
 
 /**
  * One-line account of a settled job for the `notice` form's collapsed row.
- * @param snapshot - the settled job.
+ * @param job - the settled job.
  * @returns its kind, label, and status, bounded like every notice summary.
  */
-function completionSummary(snapshot: JobSnapshot): string {
-  return boundContextSummary(`${snapshot.kind} ${snapshot.label} ${statusLine(snapshot)}`)
+function completionSummary(job: JobView): string {
+  return boundContextSummary(`${job.kind} ${job.label} ${statusLine({ status: job.status, detail: jobDetail(job) })}`)
 }
 
-function fitCompletionNotice(snapshot: JobSnapshot): string {
-  const prefix = `background job ${snapshot.id}`
-  const detail = ` (${snapshot.kind}: ${snapshot.label}) finished ${statusLine(snapshot)}`
+function fitCompletionNotice(job: JobView): string {
+  const prefix = `background job ${job.id}`
+  const detail = ` (${job.kind}: ${job.label}) finished ${statusLine({ status: job.status, detail: jobDetail(job) })}`
   const action = '\nDone; job_output.'
   const complete = `${prefix}${detail}. Read its output with job_output.`
-  const maxBytes = snapshot.outputLimitBytes
+  const maxBytes = job.outputLimitBytes
   if (maxBytes === undefined || encoder.encode(complete).byteLength <= maxBytes) return complete
   const omitted = '\n[notice truncated]'
   const fixed = `${prefix}${omitted}${action}`
@@ -181,11 +152,12 @@ function boundSingleText(content: readonly ContentBlock[], maxBytes: number): Co
   }]
 }
 
+/** The producer's cap for the job a `job_output` or `job_kill` call names, when it is visible to the caller. */
 function visibleOutputLimit(ctx: Context, exec: ToolExecution): number | undefined {
   if (exec.name !== 'job_output' && exec.name !== 'job_kill') return undefined
   const jobId = (exec.arguments as { job_id?: unknown } | null | undefined)?.job_id
   if (typeof jobId !== 'string' || jobId.length === 0) return undefined
-  return ctx.jobs.list(exec.agent).find(snapshot => snapshot.id === jobId)?.outputLimitBytes
+  return ctx.jobs.visibleTo(exec.agent?.id).list().find(job => job.id === jobId)?.outputLimitBytes
 }
 
 /** Validate the non-empty constraint that ParameterSchemaSpec cannot express. */
@@ -201,6 +173,21 @@ function presentTaskCall(title: string, kind: 'read' | 'execute', rawInput?: str
   return { card: 'generic', title, kind, ...rawInput !== undefined ? { rawInput } : {} }
 }
 
+/** True for the three terminal statuses. */
+function isTerminal(job: JobView): boolean {
+  return job.status === 'completed' || job.status === 'killed' || job.status === 'failed'
+}
+
+/** The consuming read as the model sees it: the delta, then the result once, then the status line. */
+function readBody(jobs: VisibleJobs, id: JobId): { text: string; job: PublicJobSnapshot } {
+  const read = jobs.read(id)
+  const delta = renderModelDelta(read.chunks, read.lossy)
+  const text = read.result === undefined
+    ? delta
+    : `${delta}${delta.length > 0 && !delta.endsWith('\n') ? '\n' : ''}${read.result}`
+  return { text, job: publicJob(read.job) }
+}
+
 export function apply(ctx: Context, config: Config): void {
   const waitDefault = config.waitTimeoutMs ?? 30_000
   const waitCap = config.maxWaitTimeoutMs ?? 600_000
@@ -262,31 +249,52 @@ export function apply(ctx: Context, config: Config): void {
   ctx.systemPrompt.section({
     name: 'tool:jobs',
     order: ctx.systemPrompt.getSectionOrder('TOOL_JOBS'),
-    text: 'Track every background job id you start. You are notified in-session when a job finishes — do not busy-poll or sleep on one; keep working on independent steps and do not duplicate a running job\'s work. Before giving a final answer, collect every still-relevant job with job_output (set wait: true only when you are genuinely blocked on it), and job_kill jobs that stopped mattering.',
+    text: 'Track every background job id you start. You are notified in-session when a job finishes — do not busy-poll or sleep on one; keep working on independent steps and do not stop your turn just to wait. Use job_output when you need its output now. Use job_kill to stop a job you no longer need.',
   })
 
-  // Use the exact lifecycle owner; reusable ids could resolve to a replacement.
+  // Jobs whose terminal state this plugin already handed to the model through
+  // a tool result: a terminal `job_output`, a wait that returned it, or a
+  // `job_kill` it requested. A wait claims when it starts, not when it
+  // returns — settlement releases waiters before it announces, but their
+  // continuations run after the event — and withdraws the claim if the job
+  // outlives the wait. The settlement event drops the entry, so the ledger
+  // holds only live jobs the model already touched.
+  const delivered = new Set<JobId>()
+  const claim = (id: JobId): (() => void) => {
+    delivered.add(id)
+    return () => { delivered.delete(id) }
+  }
+
   // A busy owner is injected: the notice waits in its next-step inbox, which
   // the turn cannot close over, so jobs settling together cost one step. An
   // idle owner is woken instead, because an unclaimed notice is a completion
   // the model never learns about. Either way, disposal before the claim
-  // discards it with the owner, and teardown settlements arrive `reported`.
+  // discards it with the owner, and a teardown settlement has no reader left.
   //
-  // The registry routes each settlement to the listeners its owner's scope
-  // chain reaches, so a mount under one preset never sees another preset's
-  // agents; this listener owns delivery, not the choice of whom to deliver to.
-  ctx.jobs.onJobDone((snapshot, owner) => {
-    if (snapshot.reported || owner === undefined) return
+  // The registry routes each settlement to the scope this plugin was mounted
+  // under, so a mount under one preset never sees another preset's agents;
+  // this listener owns delivery, not the choice of whom to deliver to.
+  ctx.jobs.events.subscribe({ owners: 'scope' }, (event) => {
+    if (event.type === 'removed') {
+      delivered.delete(event.job.id)
+      return
+    }
+    if (event.type !== 'settled') return
+    const claimed = delivered.delete(event.job.id)
+    if (claimed || event.cause === 'teardown' || event.job.owner === undefined) return
+    // Use the exact lifecycle owner; a session whose agent left has no inbox.
+    const owner = ctx.agents.get(event.job.owner)
+    if (owner === undefined) return
     const message = createUserMessage({
       content: [{
         type: 'text',
-        text: fitCompletionNotice(snapshot),
+        text: fitCompletionNotice(event.job),
       }],
       source: {
         kind: 'plugin',
         plugin: 'tool-jobs',
         form: 'notice',
-        summary: completionSummary(snapshot),
+        summary: completionSummary(event.job),
       },
     })
     const spent = spentWakes.get(owner) ?? 0
@@ -328,12 +336,19 @@ export function apply(ctx: Context, config: Config): void {
     },
     async execute(args, exec) {
       const id = validateJobId(args.job_id)
+      const jobs = ctx.jobs.visibleTo(exec.agent?.id)
       if (args.wait === true) {
         const timeout = Math.min(args.timeout_ms ?? waitDefault, waitCap)
-        await ctx.jobs.wait(id, timeout, exec.agent, exec.signal)
+        const withdraw = claim(id)
+        try {
+          const view = await jobs.wait(id, timeout, exec.signal)
+          if (!isTerminal(view)) withdraw()
+        } catch (error: unknown) {
+          withdraw()
+          throw error
+        }
       }
-      const read = ctx.jobs.read(id, exec.agent)
-      return { text: read.text, job: publicJob(read.snapshot) }
+      return readBody(jobs, id)
     },
     presentCall: args => presentTaskCall(`Read output from background job ${args.job_id}`, 'read', args.job_id),
   }))
@@ -352,7 +367,7 @@ export function apply(ctx: Context, config: Config): void {
       }],
     },
     execute(_args, exec) {
-      const jobs = ctx.jobs.list(exec.agent)
+      const jobs = ctx.jobs.visibleTo(exec.agent?.id).list()
       return Promise.resolve(jobs.map(publicJob))
     },
     presentCall: () => presentTaskCall('List background jobs', 'read'),
@@ -388,12 +403,16 @@ export function apply(ctx: Context, config: Config): void {
     },
     execute(args, exec) {
       const id = validateJobId(args.job_id)
-      const result = ctx.jobs.kill(id, exec.agent, args.reason)
-      // A snapshot describes current state without consuming pending output.
-      const snapshot = publicJob(ctx.jobs.get(id, exec.agent))
+      const jobs = ctx.jobs.visibleTo(exec.agent?.id)
+      const result = jobs.kill(id, args.reason === undefined ? {} : { reason: args.reason })
+      // The model's own kill is its delivery: the settlement notice would only
+      // repeat what this tool result already said.
+      if (result === 'requested') claim(id)
+      // A projection describes current state without consuming pending output.
+      const job = publicJob(jobs.get(id))
       return Promise.resolve({
         outcome: result === 'already-finished' ? 'already-finished' as const : 'cancellation-requested' as const,
-        job: snapshot,
+        job,
       })
     },
     presentCall: args => presentTaskCall(`Kill background job ${args.job_id}`, 'execute', args.job_id),

packages/jobs/tool-jobs/src/render.ts

@@ -0,0 +1,80 @@
+/**
+ * Model-facing rendering of registry reads: the consuming delta as the
+ * shell tools have always shown it (stdout, then one marked stderr section),
+ * the `[status: …]` line, and the public job projection the tool schemas
+ * expose.
+ * @module @deepseek-ai/dsh-tool-jobs/render
+ */
+
+import type { JobChunk, JobView } from '@deepseek-ai/dsh-jobs'
+
+/** Job state safe for model-authored programs; ownership and offsets are omitted. */
+export interface PublicJobSnapshot {
+  id: string
+  kind: string
+  label: string
+  status: JobView['status']
+  /** The live progress line while running, the terminal reason once settled. */
+  detail?: string
+  startedAt: number
+  finishedAt?: number
+}
+
+/**
+ * The one-line qualifier the model reads beside a status: live progress while
+ * the job runs, the terminal reason once it settled.
+ * @param job - the job projection.
+ * @returns the qualifier, or undefined when the job carries neither.
+ */
+export function jobDetail(job: Pick<JobView, 'progress' | 'detail'>): string | undefined {
+  return job.progress ?? job.detail
+}
+
+/**
+ * Remove ownership, offsets, and limits from a registry projection.
+ * @param job - the registry projection.
+ * @returns the model-safe projection.
+ */
+export function publicJob(job: JobView): PublicJobSnapshot {
+  const detail = jobDetail(job)
+  return {
+    id: job.id,
+    kind: job.kind,
+    label: job.label,
+    status: job.status,
+    ...detail !== undefined ? { detail } : {},
+    startedAt: job.startedAt,
+    ...job.finishedAt !== undefined ? { finishedAt: job.finishedAt } : {},
+  }
+}
+
+/**
+ * Render generic status with optional detail.
+ * @param snapshot - job state to render.
+ * @returns a bracketed status line.
+ */
+export function statusLine(snapshot: Pick<PublicJobSnapshot, 'status' | 'detail'>): string {
+  return snapshot.detail !== undefined
+    ? `[status: ${snapshot.status}, ${snapshot.detail}]`
+    : `[status: ${snapshot.status}]`
+}
+
+/**
+ * Render one consuming read for the model: stdout and unlabeled chunks in
+ * order, then every stderr chunk in one `[stderr]` section, exactly as the
+ * shell tools render a foreground result. `log` chunks are producer
+ * narration for observers and never reach the model. A lossy read — the
+ * cursor fell behind the ring's retention — ends with a dropped-output
+ * notice.
+ * @param chunks - the chunks since the model cursor, in offset order.
+ * @param lossy - whether bytes before `chunks` were evicted unread.
+ * @returns the delta text, possibly empty.
+ */
+export function renderModelDelta(chunks: readonly JobChunk[], lossy: boolean): string {
+  const out = chunks.filter(chunk => 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 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]`
+}

packages/jobs/tool-jobs/tests/tool-jobs.spec.ts

@@ -10,21 +10,26 @@ import { SessionId } from '@deepseek-ai/dsh-session'
 import { bindScopeParent, createScope, scopeOf } from '@deepseek-ai/dsh-scope'
 import { JobId } from '@deepseek-ai/dsh-jobs'
 import LocalJobRegistry from '@deepseek-ai/dsh-jobs-local'
-import type { JobHooks, JobOutcome, JobSnapshot, JobStart } from '@deepseek-ai/dsh-jobs'
+import type { JobAppendOptions, JobHandle, JobHooks, JobOutcome, JobSpec, JobView } from '@deepseek-ai/dsh-jobs'
 import * as ToolTasks from '@deepseek-ai/dsh-tool-jobs'
-import { statusLine } from '@deepseek-ai/dsh-tool-jobs'
+import { publicJob, renderModelDelta, statusLine } from '@deepseek-ai/dsh-tool-jobs'
 
 const testToolSignal = new AbortController().signal
 
 const agentRegistryDisposers = new WeakMap<Agent, () => void>()
 const agentScopeFibers = new WeakMap<Agent, { dispose: () => Promise<void> }>()
 
-async function setup(config: ToolTasks.Config = {}) {
+/** Registry knobs a test may narrow; every producer here pushes, so the pump stays idle. */
+interface RegistryConfig {
+  retainBytes?: number
+}
+
+async function setup(config: ToolTasks.Config = {}, registry: RegistryConfig = {}) {
   const ctx = new Context()
   await ctx.plugin(SystemPrompt)
   await ctx.plugin(ToolRuntime)
   const agentsFiber = await ctx.plugin(AgentRegistry)
-  await ctx.plugin(LocalJobRegistry)
+  await ctx.plugin(LocalJobRegistry, registry)
   const toolsFiber = await ctx.plugin(ToolTasks, config)
   return { ctx, agentsFiber, toolsFiber }
 }
@@ -70,9 +75,13 @@ async function disposeAgentScope(agent: Agent): Promise<void> {
   await fiber.dispose()
 }
 
-/** A controllable producer start-spec (settle `done` on demand, record cancels). */
-function producer(overrides: Partial<Omit<JobStart, 'run'> & JobHooks> = {}) {
+/**
+ * A controllable producer spec: pushes output and progress through the handle
+ * the registry hands its starter, settles `done` on demand, records cancels.
+ */
+function producer(overrides: Partial<Omit<JobSpec, 'run' | 'output'> & JobHooks> = {}) {
   let settle!: (outcome: JobOutcome) => void
+  let handle: JobHandle | undefined
   const cancels: (string | undefined)[] = []
   const { kind = 'bash', label = 'sleep 60', owner, outputLimitBytes, ...hookOverrides } = overrides
   const hooks: JobHooks = {
@@ -80,14 +89,24 @@ function producer(overrides: Partial<Omit<JobStart, 'run'> & JobHooks> = {}) {
     done: new Promise<JobOutcome>((res) => { settle = res }),
     ...hookOverrides,
   }
-  const spec: JobStart = {
+  const spec: JobSpec = {
     kind,
     label,
     ...owner !== undefined ? { owner } : {},
     ...outputLimitBytes !== undefined ? { outputLimitBytes } : {},
-    run: () => hooks,
+    run: (job) => { handle = job; return hooks },
+  }
+  const started = (): JobHandle => {
+    if (handle === undefined) throw new Error('producer not started')
+    return handle
+  }
+  return {
+    spec,
+    settle,
+    cancels,
+    append: (text: string, options?: JobAppendOptions) => { started().append(text, options) },
+    progress: (line: string) => { started().updateProgress(line) },
   }
-  return { spec, settle, cancels }
 }
 
 let callCounter = 0
@@ -104,7 +123,7 @@ const tick = () => new Promise<void>(r => setTimeout(r, 0))
 /** Start and settle `count` owned jobs one at a time, letting each notice land. */
 async function settleTasks(ctx: Context, owner: Agent, count: number): Promise<void> {
   for (let i = 0; i < count; i += 1) {
-    const p = producer({ owner })
+    const p = producer({ owner: owner.id })
     ctx.jobs.start(p.spec)
     p.settle({ status: 'completed' })
     await tick()
@@ -119,10 +138,18 @@ describe('tool-jobs setup', () => {
     expect(() => ctx.jobs.start(producer().spec)).toThrow('no job controller serves this agent')
   })
 
+  it('leaves with the agent registry it resolves owners through', async () => {
+    const { ctx, agentsFiber } = await setup()
+    await agentsFiber.dispose()
+    expect(ctx.tools.get('job_output')).toBeUndefined()
+    expect(() => ctx.jobs.start(producer().spec)).toThrow('no job controller serves this agent')
+  })
+
   it('rejects a config whose default wait exceeds the cap', async () => {
     const ctx = new Context()
     await ctx.plugin(SystemPrompt)
     await ctx.plugin(ToolRuntime)
+    await ctx.plugin(AgentRegistry)
     await ctx.plugin(LocalJobRegistry)
     await expect(ctx.plugin(ToolTasks, { waitTimeoutMs: 100, maxWaitTimeoutMs: 50 }))
       .rejects.toThrow('waitTimeoutMs (100) exceeds maxWaitTimeoutMs (50)')
@@ -141,6 +168,7 @@ describe('tool-jobs setup', () => {
       const ctx = new Context()
       await ctx.plugin(SystemPrompt)
       await ctx.plugin(ToolRuntime)
+      await ctx.plugin(AgentRegistry)
       await ctx.plugin(LocalJobRegistry)
       try {
         await ctx.plugin(ToolTasks, { maxConsecutiveWakes })
@@ -158,9 +186,35 @@ describe('tool-jobs setup', () => {
   })
 
   it('renders status lines with and without producer detail', () => {
-    const base = { id: 'bash-1', kind: 'bash', label: 'x', startedAt: 0, reported: false } as unknown as JobSnapshot
-    expect(statusLine({ ...base, status: 'running' })).toBe('[status: running]')
-    expect(statusLine({ ...base, status: 'completed', detail: 'exit code: 0' })).toBe('[status: completed, exit code: 0]')
+    expect(statusLine({ status: 'running' })).toBe('[status: running]')
+    expect(statusLine({ status: 'completed', detail: 'exit code: 0' })).toBe('[status: completed, exit code: 0]')
+  })
+
+  it('projects the public job: live progress or terminal detail, never ownership or offsets', () => {
+    const live: JobView = {
+      id: JobId('bash-1'), kind: 'bash', label: 'x', owner: SessionId('sess-1'), outputLimitBytes: 8,
+      status: 'running', progress: '3/10 files', startedAt: 5, output: { total: 12, earliest: 0 },
+    }
+    expect(publicJob(live)).toEqual({ id: 'bash-1', kind: 'bash', label: 'x', status: 'running', detail: '3/10 files', startedAt: 5 })
+    const settled: JobView = {
+      id: JobId('bash-1'), kind: 'bash', label: 'x', status: 'completed', detail: 'exit code: 0',
+      startedAt: 5, finishedAt: 9, output: { total: 12, earliest: 0 },
+    }
+    expect(publicJob(settled)).toEqual({ id: 'bash-1', kind: 'bash', label: 'x', status: 'completed', detail: 'exit code: 0', startedAt: 5, finishedAt: 9 })
+  })
+
+  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([
+      { at: 0, text: 'a' },
+      { at: 1, text: 'e1\n', channel: 'stderr' },
+      { at: 4, text: 'narration', channel: 'log' },
+      { 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]')
   })
 
   it('applies the built-in wait bounds when apply() receives a bare config', async () => {
@@ -169,6 +223,7 @@ describe('tool-jobs setup', () => {
     const ctx = new Context()
     await ctx.plugin(SystemPrompt)
     await ctx.plugin(ToolRuntime)
+    await ctx.plugin(AgentRegistry)
     await ctx.plugin(LocalJobRegistry)
     ToolTasks.apply(ctx, {})
     expect(ctx.tools.get('job_output')).toBeDefined()
@@ -179,8 +234,9 @@ describe('tool-jobs setup', () => {
 describe('job_output', () => {
   it('reads a consuming delta with a trailing status line', async () => {
     const { ctx } = await setup()
-    const chunks = ['line one\n', '']
-    ctx.jobs.start(producer({ readOutput: () => chunks.shift() ?? '' }).spec)
+    const p = producer()
+    ctx.jobs.start(p.spec)
+    p.append('line one\n')
 
     // A body already ending in a newline gets no doubled separator.
     const first = await call(ctx, 'job_output', { job_id: 'bash-1' })
@@ -190,29 +246,75 @@ describe('job_output', () => {
       text: 'line one\n',
       job: { id: 'bash-1', kind: 'bash', label: 'sleep 60', status: 'running' },
     })
-    expect(firstValue.job).not.toHaveProperty('ownerSession')
-    expect(firstValue.job).not.toHaveProperty('reported')
+    expect(firstValue.job).not.toHaveProperty('owner')
+    expect(firstValue.job).not.toHaveProperty('output')
     expect(text(first)).toBe('line one\n[status: running]')
     expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('(no new output)\n[status: running]')
   })
 
-  it('returns the final output of a settled final-output job', async () => {
+  it('renders stderr in a trailing section and never shows log narration', async () => {
+    const { ctx } = await setup()
+    const p = producer()
+    ctx.jobs.start(p.spec)
+    p.append('out\n')
+    p.append('warn\n', { channel: 'stderr' })
+    p.append('observer-only narration', { channel: 'log' })
+    expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('out\n[stderr]\nwarn\n[status: running]')
+    expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('(no new output)\n[status: running]')
+  })
+
+  it('shows live progress beside the status until settlement replaces it with the terminal detail', async () => {
+    const { ctx } = await setup()
+    const p = producer()
+    ctx.jobs.start(p.spec)
+    p.progress('3/10 files')
+    expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('(no new output)\n[status: running, 3/10 files]')
+    p.settle({ status: 'completed', detail: 'exit code: 0' })
+    await tick()
+    expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('(no new output)\n[status: completed, exit code: 0]')
+  })
+
+  it('flags a read whose cursor fell behind the ring', async () => {
+    const { ctx } = await setup({}, { retainBytes: 16 })
+    const p = producer()
+    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('[status: running]')
+    expect(first).not.toContain('x'.repeat(17))
+    // The cursor now sits at the ring's end: the next read is clean.
+    expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('(no new output)\n[status: running]')
+  })
+
+  it('returns the result of a settled job exactly once, after the pending delta', async () => {
     const { ctx } = await setup()
     const p = producer({ kind: 'subagent', label: 'research' })
     ctx.jobs.start(p.spec)
     expect(text(await call(ctx, 'job_output', { job_id: 'subagent-1' }))).toBe('(no new output)\n[status: running]')
 
-    p.settle({ status: 'completed', detail: 'completed', output: 'the answer' })
+    p.append('partial')
+    p.settle({ status: 'completed', detail: 'completed', result: 'the answer' })
     await tick()
-    expect(text(await call(ctx, 'job_output', { job_id: 'subagent-1' }))).toBe('the answer\n[status: completed, completed]')
+    expect(text(await call(ctx, 'job_output', { job_id: 'subagent-1' }))).toBe('partial\nthe answer\n[status: completed, completed]')
+    expect(text(await call(ctx, 'job_output', { job_id: 'subagent-1' }))).toBe('(no new output)\n[status: completed, completed]')
+  })
+
+  it('keeps a newline-terminated delta and the result on separate lines', async () => {
+    const { ctx } = await setup()
+    const p = producer({ kind: 'subagent', label: 'research' })
+    ctx.jobs.start(p.spec)
+    p.append('progress log\n')
+    p.settle({ status: 'completed', result: 'the answer' })
+    await tick()
+    expect(text(await call(ctx, 'job_output', { job_id: 'subagent-1' }))).toBe('progress log\nthe answer\n[status: completed]')
   })
 
   it('applies a producer limit to the complete body and status result', async () => {
     const { ctx } = await setup()
-    ctx.jobs.start(producer({
-      outputLimitBytes: 48,
-      readOutput: () => '界'.repeat(100),
-    }).spec)
+    const p = producer({ outputLimitBytes: 48 })
+    ctx.jobs.start(p.spec)
+    p.append('界'.repeat(100))
 
     const output = text(await call(ctx, 'job_output', { job_id: 'bash-1' }))
     expect(Buffer.byteLength(output)).toBeLessThanOrEqual(48)
@@ -221,24 +323,21 @@ describe('job_output', () => {
 
   it('preserves empty and newline-terminated output under a producer limit', async () => {
     const { ctx } = await setup()
-    const chunks = ['', 'line\n']
-    ctx.jobs.start(producer({
-      outputLimitBytes: 64,
-      readOutput: () => chunks.shift() ?? '',
-    }).spec)
+    const p = producer({ outputLimitBytes: 64 })
+    ctx.jobs.start(p.spec)
 
     expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' })))
       .toBe('(no new output)\n[status: running]')
+    p.append('line\n')
     expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' })))
       .toBe('line\n[status: running]')
   })
 
   it('bounds post-policy output without restoring the canonical status rendering', async () => {
     const { ctx } = await setup()
-    ctx.jobs.start(producer({
-      outputLimitBytes: 64,
-      readOutput: () => 'canonical output',
-    }).spec)
+    const p = producer({ outputLimitBytes: 64 })
+    ctx.jobs.start(p.spec)
+    p.append('canonical output')
     ctx.on('tools/post-execute', (exec, _result, next) => {
       if (exec.name !== 'job_output') return next()
       return Promise.resolve({ kind: 'accept', content: [{ type: 'text', text: 'p'.repeat(1_000) }] })
@@ -250,17 +349,30 @@ describe('job_output', () => {
     expect(text(result)).not.toContain('[status: running]')
   })
 
-  it('applies a producer limit to a normalized read failure', async () => {
+  it('does not double a truncation marker a policy already appended', async () => {
     const { ctx } = await setup()
-    ctx.jobs.start(producer({
-      outputLimitBytes: 64,
-      readOutput: () => { throw new Error('read failed: '.repeat(100)) },
-    }).spec)
+    ctx.jobs.start(producer({ outputLimitBytes: 64 }).spec)
+    ctx.on('tools/post-execute', (exec, _result, next) => {
+      if (exec.name !== 'job_output') return next()
+      return Promise.resolve({ kind: 'accept', content: [{ type: 'text', text: `${'z'.repeat(1_000)}\n[result truncated]` }] })
+    })
 
     const result = await call(ctx, 'job_output', { job_id: 'bash-1' })
-    expect(result.isError).toBe(true)
     expect(Buffer.byteLength(text(result))).toBeLessThanOrEqual(64)
-    expect(text(result)).toContain('[result truncated]')
+    expect(text(result).match(/\[result truncated\]/g)).toHaveLength(1)
+  })
+
+  it('refuses a job owned by another session as an errored result', async () => {
+    const { ctx } = await setup()
+    const alice = fakeAgent(ctx, 'sess-alice')
+    ctx.jobs.start(producer({ owner: alice.id }).spec)
+    const bob = fakeAgent(ctx, 'sess-bob')
+
+    const result = await call(ctx, 'job_output', { job_id: 'bash-1' }, bob)
+    expect(result.isError).toBe(true)
+    expect(text(result)).toContain('belongs to another session')
+    expect((await call(ctx, 'job_output', { job_id: 'bash-1' })).isError).toBe(true)
+    expect((await call(ctx, 'job_output', { job_id: 'bash-1' }, alice)).isError).toBe(false)
   })
 
   it('bounds pre-, around-, and post-execute policy outcomes and failures', async () => {
@@ -325,7 +437,7 @@ describe('job_output', () => {
     ctx.jobs.start(p.spec)
 
     const pending = call(ctx, 'job_output', { job_id: 'subagent-1', wait: true })
-    p.settle({ status: 'completed', output: 'done deal' })
+    p.settle({ status: 'completed', result: 'done deal' })
     expect(text(await pending)).toBe('done deal\n[status: completed]')
   })
 
@@ -354,9 +466,9 @@ describe('job_list', () => {
     expect(text(await call(ctx, 'job_list', {}))).toBe('(no background jobs)')
 
     const alice = fakeAgent(ctx, 'sess-alice')
-    ctx.jobs.start(producer({ owner: alice, label: 'pnpm test' }).spec)
+    ctx.jobs.start(producer({ owner: alice.id, label: 'pnpm test' }).spec)
     ctx.jobs.start(producer({ kind: 'subagent', label: 'open research' }).spec)
-    const p = producer({ owner: alice, label: 'build' })
+    const p = producer({ owner: alice.id, label: 'build' })
     ctx.jobs.start(p.spec)
     p.settle({ status: 'completed', detail: 'exit code: 0' })
     await tick()
@@ -368,8 +480,8 @@ describe('job_list', () => {
     expect(listedValue[0]).toMatchObject({ id: 'bash-1', kind: 'bash', label: 'pnpm test', status: 'running' })
     expect(listedValue[2]).toMatchObject({ id: 'bash-2', kind: 'bash', label: 'build', status: 'completed', detail: 'exit code: 0' })
     for (const job of listedValue) {
-      expect(job).not.toHaveProperty('ownerSession')
-      expect(job).not.toHaveProperty('reported')
+      expect(job).not.toHaveProperty('owner')
+      expect(job).not.toHaveProperty('output')
     }
     expect(text(listed)).toBe([
       'bash-1 [bash] running — pnpm test',
@@ -395,8 +507,8 @@ describe('job_kill', () => {
       outcome: 'cancellation-requested',
       job: { id: 'bash-1', kind: 'bash', label: 'sleep 60', status: 'stopping' },
     })
-    expect(killValue.job).not.toHaveProperty('ownerSession')
-    expect(killValue.job).not.toHaveProperty('reported')
+    expect(killValue.job).not.toHaveProperty('owner')
+    expect(killValue.job).not.toHaveProperty('output')
     expect(text(result)).toBe('requested cancellation of job bash-1')
     expect(p.cancels).toEqual(['superseded'])
   })
@@ -422,7 +534,7 @@ describe('job_kill', () => {
     expect(result.isError).toBe(true)
     expect(Buffer.byteLength(text(result))).toBeLessThanOrEqual(64)
     expect(text(result)).toContain('[result truncated]')
-    expect(ctx.jobs.get(JobId('bash-1'))).toMatchObject({ status: 'running', reported: false })
+    expect(ctx.jobs.visibleTo().get(JobId('bash-1')).status).toBe('running')
   })
 
   it('bounds single-text post policy while preserving structured policy results', async () => {
@@ -470,9 +582,9 @@ describe('job_kill', () => {
 
   it('reports an already-finished job without consuming its pending delta', async () => {
     const { ctx } = await setup()
-    let delta = 'unread tail'
-    const p = producer({ readOutput: () => { const d = delta; delta = ''; return d } })
+    const p = producer()
     ctx.jobs.start(p.spec)
+    p.append('unread tail')
     p.settle({ status: 'completed', detail: 'exit code: 0' })
     await tick()
 
@@ -483,7 +595,7 @@ describe('job_kill', () => {
       job: { id: 'bash-1', kind: 'bash', label: 'sleep 60', status: 'completed', detail: 'exit code: 0' },
     })
     expect(text(killed)).toBe('job bash-1 had already finished [status: completed, exit code: 0]')
-    // The kill described the job via a non-consuming snapshot: the delta is intact.
+    // The kill described the job via a non-consuming projection: the delta is intact.
     expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('unread tail\n[status: completed, exit code: 0]')
   })
 
@@ -508,9 +620,9 @@ describe('tool-owned UI presentation (presentCall)', () => {
 describe('completion notices across scoped mounts', () => {
   /**
    * Two agent presets mounting `tool-jobs` over ONE host registry: each mount
-   * registers its own `onJobDone` listener on the shared service, and
-   * `settle()` broadcasts one snapshot to every listener with no scope filter.
-   * Only the mount whose scope the owner belongs to may deliver the notice.
+   * subscribes with `{ owners: 'scope' }`, which the registry files into the
+   * mount's own scope layer and reaches along the owner's scope chain. Only
+   * the mount whose scope the owner belongs to sees the settlement.
    */
   it('delivers one notice from the owning scope when two mounts share the registry', async () => {
     const ctx = new Context()
@@ -539,9 +651,7 @@ describe('completion notices across scoped mounts', () => {
     const dispose = ctx.agents.register(owner)
 
     try {
-      // No waiter: `settle()` leaves `reported` false, which is the only path
-      // that reaches the notice listeners at all.
-      const p = producer({ owner, label: 'pnpm test' })
+      const p = producer({ owner: owner.id, label: 'pnpm test' })
       ctx.jobs.start(p.spec)
       p.settle({ status: 'completed', detail: 'exit code: 0' })
       await tick()
@@ -559,7 +669,7 @@ describe('completion notice delivery', () => {
     const inject = vi.fn()
     const followup = vi.fn()
     const owner = fakeAgent(ctx, 'sess-1', { inject, followup, status: 'idle' })
-    const p = producer({ owner, label: 'pnpm test' })
+    const p = producer({ owner: owner.id, label: 'pnpm test' })
     ctx.jobs.start(p.spec)
 
     p.settle({ status: 'completed', detail: 'exit code: 0' })
@@ -573,7 +683,7 @@ describe('completion notice delivery', () => {
     const inject = vi.fn()
     const followup = vi.fn()
     const owner = fakeAgent(ctx, 'sess-1', { inject, followup, status: 'idle' })
-    const p = producer({ owner })
+    const p = producer({ owner: owner.id })
     ctx.jobs.start(p.spec)
 
     p.settle({ status: 'completed' })
@@ -621,15 +731,15 @@ describe('completion notice delivery', () => {
     ctx.jobs.start({
       kind: 'bash',
       label: 'sleep 60',
-      owner,
+      owner: owner.id,
       run: () => ({
         cancel() { settle({ status: 'killed' }) },
         done: new Promise<JobOutcome>((res) => { settle = res }),
       }),
     })
 
-    // Disposal cancels and settles the owned job. Waking here would spend a
-    // model request on an agent the host is destroying, once per tree layer.
+    // Disposal cancels and settles the owned job with a teardown cause. Waking
+    // here would spend a model request on an agent the host is destroying.
     await disposeAgentScope(owner)
     await tick()
     expect(followup).not.toHaveBeenCalled()
@@ -645,17 +755,16 @@ describe('completion notice delivery', () => {
     ctx.jobs.start({
       kind: 'bash',
       label: 'broken producer',
-      owner,
+      owner: owner.id,
       run: () => ({
         cancel() { throw new Error('cancel boom') },
         done: new Promise<JobOutcome>(() => {}),
       }),
     })
 
-    // The registry force-fails the record instead of deadlocking. That path
-    // settles the job too, so it must claim the report as the ordinary
-    // teardown cancel does — otherwise a throwing producer is all it takes to
-    // spend a model request on an owner being destroyed.
+    // The registry force-fails the record instead of deadlocking. That
+    // settlement also announces a teardown cause, so a throwing producer is
+    // not enough to spend a model request on an owner being destroyed.
     await disposeAgentScope(owner)
     await tick()
     expect(warn).toHaveBeenCalledWith(expect.stringContaining('work may be orphaned'))
@@ -682,11 +791,11 @@ describe('completion notice delivery', () => {
 })
 
 describe('completion notices', () => {
-  it('injects a notice into the owning agent when an unreported job settles', async () => {
+  it('injects a notice into the owning agent when an uncollected job settles', async () => {
     const { ctx } = await setup()
     const inject = vi.fn()
     const owner = fakeAgent(ctx, 'sess-1', { inject })
-    const p = producer({ owner, label: 'pnpm test' })
+    const p = producer({ owner: owner.id, label: 'pnpm test' })
     ctx.jobs.start(p.spec)
 
     p.settle({ status: 'completed', detail: 'exit code: 0' })
@@ -710,7 +819,7 @@ describe('completion notices', () => {
     const inject = vi.fn()
     const owner = fakeAgent(ctx, 'sess-1', { inject })
     const first = producer({
-      owner,
+      owner: owner.id,
       kind: 'subagent',
       label: 'x'.repeat(1_000),
       outputLimitBytes: 61,
@@ -737,7 +846,7 @@ describe('completion notices', () => {
     )
 
     const second = producer({
-      owner,
+      owner: owner.id,
       kind: 'subagent',
       label: 'x'.repeat(1_000),
       outputLimitBytes: 80,
@@ -764,7 +873,7 @@ describe('completion notices', () => {
     const inject = vi.fn()
     const owner = fakeAgent(ctx, 'sess-1', { inject })
     const target = producer({
-      owner,
+      owner: owner.id,
       kind: 'pty-send',
       label: 'x'.repeat(1_000),
       outputLimitBytes: 64,
@@ -784,8 +893,8 @@ describe('completion notices', () => {
     const { ctx } = await setup()
     const inject = vi.fn()
     const owner = fakeAgent(ctx, 'sess-1', { inject })
-    const tiny = producer({ owner, kind: 'pty-send', label: 'x'.repeat(100), outputLimitBytes: 8 })
-    const short = producer({ owner, kind: 'pty-send', label: 'x'.repeat(100), outputLimitBytes: 32 })
+    const tiny = producer({ owner: owner.id, kind: 'pty-send', label: 'x'.repeat(100), outputLimitBytes: 8 })
+    const short = producer({ owner: owner.id, kind: 'pty-send', label: 'x'.repeat(100), outputLimitBytes: 32 })
     ctx.jobs.start(tiny.spec)
     ctx.jobs.start(short.spec)
 
@@ -805,7 +914,7 @@ describe('completion notices', () => {
     const { ctx } = await setup()
     const inject = vi.fn()
     const owner = fakeAgent(ctx, 'sess-1', { inject })
-    const p = producer({ owner })
+    const p = producer({ owner: owner.id })
     ctx.jobs.start(p.spec)
 
     await call(ctx, 'job_kill', { job_id: 'bash-1' }, owner)
@@ -818,15 +927,55 @@ describe('completion notices', () => {
     const { ctx } = await setup()
     const inject = vi.fn()
     const owner = fakeAgent(ctx, 'sess-1', { inject })
-    const p = producer({ owner, kind: 'subagent' })
+    const p = producer({ owner: owner.id, kind: 'subagent' })
     ctx.jobs.start(p.spec)
 
     const pending = call(ctx, 'job_output', { job_id: 'subagent-1', wait: true }, owner)
-    p.settle({ status: 'completed', output: 'answer' })
+    p.settle({ status: 'completed', result: 'answer' })
     expect(text(await pending)).toContain('answer')
+    await tick()
     expect(inject).not.toHaveBeenCalled()
   })
 
+  it('withdraws the wait claim when the wait times out, so the later settlement still notifies', async () => {
+    const { ctx } = await setup({ waitTimeoutMs: 10, maxWaitTimeoutMs: 20 })
+    const inject = vi.fn()
+    const owner = fakeAgent(ctx, 'sess-1', { inject })
+    const p = producer({ owner: owner.id })
+    ctx.jobs.start(p.spec)
+
+    expect(text(await call(ctx, 'job_output', { job_id: 'bash-1', wait: true }, owner))).toBe('(no new output)\n[status: running]')
+    p.settle({ status: 'completed' })
+    await tick()
+    expect(inject).toHaveBeenCalledTimes(1)
+  })
+
+  it('withdraws the wait claim when the caller aborts the wait', async () => {
+    const { ctx } = await setup()
+    const inject = vi.fn()
+    const owner = fakeAgent(ctx, 'sess-1', { inject })
+    const p = producer({ owner: owner.id })
+    ctx.jobs.start(p.spec)
+
+    const aborter = new AbortController()
+    const pending = ctx.tools.execute({
+      signal: aborter.signal,
+      callId: ToolCallId('call-aborted-wait'),
+      name: 'job_output',
+      arguments: { job_id: 'bash-1', wait: true },
+      agent: owner,
+    })
+    await tick()
+    aborter.abort()
+    const result = await pending
+    expect(result.isError).toBe(true)
+    expect(text(result)).toContain('wait aborted')
+
+    p.settle({ status: 'completed' })
+    await tick()
+    expect(inject).toHaveBeenCalledTimes(1)
+  })
+
   it('drops the notice for unowned jobs without throwing', async () => {
     const { ctx } = await setup()
     // Unowned: settles with nobody to notify — nothing throws.
@@ -836,31 +985,45 @@ describe('completion notices', () => {
     await tick()
   })
 
-  it('does not route an old owner completion notice to a same-session replacement', async () => {
+  it('delivers to the agent currently registered for the owner session', async () => {
     const { ctx } = await setup()
-    // Delivery into a tearing-down owner is a plain inject: the loop has no
-    // terminal state, so the notice lands in the old owner's (detached)
-    // session instead of throwing or re-routing.
+    // The settlement names the owner session; the notice goes to whichever
+    // agent holds that session when the job settles, which can read the job
+    // through the same session id.
     const oldInject = vi.fn()
     const oldOwner = fakeAgent(ctx, 'shared', { inject: oldInject })
-    const p = producer({ owner: oldOwner })
+    const p = producer({ owner: oldOwner.id })
     ctx.jobs.start(p.spec)
 
     detachAgent(oldOwner)
     const replacementInject = vi.fn()
-    fakeAgent(ctx, 'shared', { inject: replacementInject })
+    const replacement = fakeAgent(ctx, 'shared', { inject: replacementInject })
     p.settle({ status: 'completed' })
     await tick()
 
-    expect(oldInject).toHaveBeenCalledTimes(1)
-    expect(replacementInject).not.toHaveBeenCalled()
+    expect(oldInject).not.toHaveBeenCalled()
+    expect(replacementInject).toHaveBeenCalledTimes(1)
+    expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }, replacement))).toBe('(no new output)\n[status: completed]')
+  })
+
+  it('drops the notice when the owner session has no live agent at settlement', async () => {
+    const { ctx } = await setup()
+    const inject = vi.fn()
+    const owner = fakeAgent(ctx, 'sess-1', { inject })
+    const p = producer({ owner: owner.id })
+    ctx.jobs.start(p.spec)
+
+    detachAgent(owner)
+    p.settle({ status: 'completed' })
+    await tick()
+    expect(inject).not.toHaveBeenCalled()
   })
 
   it('surfaces an inject failure through listener containment (a real bug must be visible)', async () => {
     const { ctx } = await setup()
     const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
     const owner = fakeAgent(ctx, 'sess-1', { inject: () => { throw new Error('unexpected inject bug') } })
-    const p = producer({ owner })
+    const p = producer({ owner: owner.id })
     ctx.jobs.start(p.spec)
     p.settle({ status: 'completed' })
     await tick()
@@ -868,23 +1031,4 @@ describe('completion notices', () => {
     // registry's per-listener containment — visible, not swallowed.
     expect(warn).toHaveBeenCalledWith(expect.stringContaining('unexpected inject bug'))
   })
-
-  it('keeps using the exact owner after the agent registry is gone', async () => {
-    const { ctx, agentsFiber } = await setup()
-    const inject = vi.fn()
-    const owner = fakeAgent(ctx, 'sess-1', { inject })
-
-    // Settlement must not depend on a later registry lookup: the exact owner
-    // supplied at start remains the destination while its own scope is live.
-    const p1 = producer({ owner })
-    ctx.jobs.start(p1.spec)
-    const p2 = producer({ owner })
-    ctx.jobs.start(p2.spec)
-
-    await agentsFiber.dispose()
-    p1.settle({ status: 'completed' })
-    p2.settle({ status: 'failed' })
-    await tick()
-    expect(inject).toHaveBeenCalledTimes(2)
-  })
 })