Merge branch 'worktree/process-plugin-streaming-72f9b3' into worktree/job-kill-delivery

# Conflicts:
#	packages/api/job-controller/README.i18n.yaml
#	packages/api/job-controller/README.md
#	packages/api/job-controller/README.zh.md

.agents/notes/implemented/architecture/2026-09-03-jobs-seam-consolidation.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 .agents/notes/implemented/architecture/2026-09-03-jobs-seam-consolidation.md
-2026-09-03-jobs-seam-consolidation.md: 2b7f9c5cab7bba20c50dded9d361d8701e0ad792
-2026-09-03-jobs-seam-consolidation.zh.md: 169bb52a2fe512e7e94911e42e0c7981c5fa7273
+2026-09-03-jobs-seam-consolidation.md: d2bf22493fce39d6dfffc4127e5575569b2599c1
+2026-09-03-jobs-seam-consolidation.zh.md: 90fc35ddb19805fbf5d4a7ca9e0fd83064f1a5bd

.agents/notes/implemented/architecture/2026-09-03-jobs-seam-consolidation.md

@@ -32,4 +32,4 @@ The job seam had grown by accretion. Producers declared their output twice — a
 
 The model-visible text is unchanged: status lines, tool descriptions and schemas, and notice wording are byte-identical, and the only new model-visible fact is the kill reason a `killed` status line now carries. A producer that needs to stream output picks pull sources or `append`; it never formats a consuming delta. The Web client renders every live job as expandable, so a job that never writes shows an empty running panel instead of a static row. `JobKindMap` merges are host-side, so a browser program types `JobView.kind` as the kinds it knows and treats the rest as opaque strings, which is all the roster does with it.
 
-Review follow-ups after the master merge tightened four of these edges. A job's first event is always `registered`: the announcement precedes the pump, whose first drain runs synchronously and may announce output. A settled ring keeps every byte the model cursor has not consumed until the first terminal model read, so a job that finishes before its first `job_output` loses nothing the live cap retained. `dsh-tool-jobs` counts wait claims per call, so a timed-out wait withdraws only its own claim while a concurrent wait keeps the settlement covered. The browser records an observation failure that precedes the anchor as an errored view, so the panel shows the notice instead of staying blank. The `@deepseek-ai/dsh-jobs/invariant` companion checks the announced event protocol per job (registered first, one settlement, removal last) and each announced projection against the registry's own read, which is the cross-observation the [companion rule](../simplification/2026-08-28-omit-unneeded-invariant-companions.md) requires; the earlier single-view field checks are gone. A lossy source read now carries the source's spill file onto its gap chunk, and the model's read renders the shell tools' dropped-output notice for a producer-side gap as well as for ring eviction, naming those files — the `before this read` variant this change had introduced is gone. An observation generation that sees its job removed mid-stream (the owner's teardown) closes with the removed job's terminal projection instead of a failed read. A rejected background spawn produces no process output, so the shell providers now serve the `spawn failed: …` note as the whole observed stderr stream (the consuming read still folds it in once), and `ShellProcess.observed` with both readers is required: the model's first `job_output` after such a failure reads exactly what master's consuming read rendered.
+Review follow-ups after the master merge tightened four of these edges. A job's first event is always `registered`: the announcement precedes the pump, whose first drain runs synchronously and may announce output. A settled ring keeps every byte the model cursor has not consumed until the first terminal model read, so a job that finishes before its first `job_output` loses nothing the live cap retained. `dsh-tool-jobs` holds one claim per wait call with a `live` flag that the caller's abort clears synchronously from the abort event, so a settlement announced before the registry's rejection reaches the tool still delivers its notice, and a timed-out wait drops only its own claim while a concurrent wait keeps the settlement covered. The browser records an observation failure that precedes the anchor as an errored view, so the panel shows the notice instead of staying blank. The `@deepseek-ai/dsh-jobs/invariant` companion checks the announced event protocol per job (registered first, one settlement, removal last) and each announced projection against the registry's own read, which is the cross-observation the [companion rule](../simplification/2026-08-28-omit-unneeded-invariant-companions.md) requires; the earlier single-view field checks are gone. The spill file a pull source keeps is job metadata (`JobView.output.spillPaths`, reported by every pump read and withdrawn once a read stops naming it), not chunk provenance, and the model's read renders the shell tools' dropped-output notice for a producer-side gap as well as for ring eviction, naming those files even after the gap chunk itself was evicted — the `before this read` variant this change had introduced is gone. An observation generation that sees its job removed mid-stream (the owner's teardown) closes with the removed job's terminal projection instead of a failed read. A rejected background spawn produces no process output, so the shell providers now serve the `spawn failed: …` note as the whole observed stderr stream (the consuming read still folds it in once), and `ShellProcess.observed` with both readers is required: the model's first `job_output` after such a failure reads exactly what master's consuming read rendered.

.agents/notes/implemented/architecture/2026-09-03-jobs-seam-consolidation.zh.md

@@ -32,4 +32,4 @@ job seam 是逐步堆出来的。生产者要把输出声明两次——模型
 
 模型可见文本保持不变：状态行、工具描述与 schema、播报措辞逐字节相同，唯一新增的模型可见事实是 `killed` 状态行现在携带的 kill 原因。需要流式输出的生产者选择拉取源或 `append`，从不格式化消耗式增量。Web 客户端把每个进行中的 job 渲染为可展开，因此从不写输出的 job 显示为空的运行中面板而非静态行。`JobKindMap` 的合并发生在宿主侧，浏览器程序把 `JobView.kind` 视为它认识的 kind 集合，其余当作不透明字符串——名册对它也只做这些。
 
-合并 master 之后的评审跟进收紧了其中四条边。job 的第一个事件永远是 `registered`：通告先于泵，而泵的首轮 drain 同步执行且可能通告输出。已结算的环在模型的首次终态读取之前保留模型游标尚未消费的全部字节，因此在首次 `job_output` 之前结束的 job 不会丢失运行期上限保留的任何内容。`dsh-tool-jobs` 按调用计数等待认领，超时的等待只撤回自己的认领，并发的另一个等待仍覆盖该结算。浏览器把先于锚点发生的观察失败记录为带错误的视图，面板显示提示而不是留白。`@deepseek-ai/dsh-jobs/invariant` 伴生插件校验每个 job 的事件协议（先 `registered`、恰一次结算、最后 `removed`）以及每个通告的投影与注册表自身读取的一致性，这正是[伴生插件规则](../simplification/2026-08-28-omit-unneeded-invariant-companions.zh.md)要求的交叉观察；先前只针对单个视图的字段校验已删除。lossy 的源读取现在把源的 spill 文件带到它的缺口块上，模型读取对生产者侧缺口与环淘汰都渲染 shell 工具的丢失输出提示并列出这些文件——本次变更曾引入的 `before this read` 变体已删除。观察生成过程若在流中看到其 job 被移除（拥有者 teardown），以被移除 job 的终态投影收尾，而不是一次失败的读取。被拒绝的后台 spawn 不产生任何进程输出，因此 shell provider 现在把 `spawn failed: …` 提示作为 observed stderr 的整条流提供（消耗式读取仍只折入一次），且 `ShellProcess.observed` 及其两个读取器改为必填：这类失败之后模型的首次 `job_output` 读到的正是 master 的消耗式读取所渲染的内容。
+合并 master 之后的评审跟进收紧了其中四条边。job 的第一个事件永远是 `registered`：通告先于泵，而泵的首轮 drain 同步执行且可能通告输出。已结算的环在模型的首次终态读取之前保留模型游标尚未消费的全部字节，因此在首次 `job_output` 之前结束的 job 不会丢失运行期上限保留的任何内容。`dsh-tool-jobs` 为每次等待调用保存一条带 `live` 标记的认领，调用方 abort 时在 abort 事件里同步清掉该标记，因此注册表的拒绝尚未传回工具之前宣布的结算仍会投递通知；超时的等待只丢弃自己的认领，并发的另一个等待仍覆盖该结算。浏览器把先于锚点发生的观察失败记录为带错误的视图，面板显示提示而不是留白。`@deepseek-ai/dsh-jobs/invariant` 伴生插件校验每个 job 的事件协议（先 `registered`、恰一次结算、最后 `removed`）以及每个通告的投影与注册表自身读取的一致性，这正是[伴生插件规则](../simplification/2026-08-28-omit-unneeded-invariant-companions.zh.md)要求的交叉观察；先前只针对单个视图的字段校验已删除。拉取源保留的 spill 文件是 job 元数据（`JobView.output.spillPaths`，每次泵读取都会上报，读取不再点名时即撤回），不是 chunk 的来源信息；模型读取对生产者侧缺口与环淘汰都渲染 shell 工具的丢失输出提示并列出这些文件，即使缺口块本身已被淘汰——本次变更曾引入的 `before this read` 变体已删除。观察生成过程若在流中看到其 job 被移除（拥有者 teardown），以被移除 job 的终态投影收尾，而不是一次失败的读取。被拒绝的后台 spawn 不产生任何进程输出，因此 shell provider 现在把 `spawn failed: …` 提示作为 observed stderr 的整条流提供（消耗式读取仍只折入一次），且 `ShellProcess.observed` 及其两个读取器改为必填：这类失败之后模型的首次 `job_output` 读到的正是 master 的消耗式读取所渲染的内容。

docs/config-catalog.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/config-catalog.md
-config-catalog.md: 6eaf8bd3fa26d6ba83ee72c8868aafdb4eae2934
+config-catalog.md: 6054bb583bd31d3266a93ae3edc6fb8921375465
 config-catalog.zh.md: ad2b630272cb7ac056b24505468b879b18dcbcfc

docs/config-catalog.md

@@ -2778,7 +2778,7 @@ export interface Config {
 export type CompletionDelivery = 'quiet' | 'wakeup'
 ```
 
-Source: [`packages/jobs/tool-jobs/src/index.ts:38`](../packages/jobs/tool-jobs/src/index.ts)
+Source: [`packages/jobs/tool-jobs/src/index.ts:46`](../packages/jobs/tool-jobs/src/index.ts)
 
 <a id="deepseek-aidsh-tool-lsp"></a>
 

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: 51e42429d3200f90a64ef185377c01f04d86dce1
-jobs.zh.md: ea536cd89b223569b9afe39f2b2bcc7c51e7681f
+jobs.md: 39a183acaadd256befd94a9f3da28189eaf6c2d6
+jobs.zh.md: c7594d385af3afd5894b784a3621a3e0f2f9c1c5

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, the lossy flag, and the spill path when one exists.
+   * @returns the delta text, the next offset, the lossy flag, and the spill path the source currently keeps.
    */
   read(fromByte: number): JobSourceRead
 }
@@ -165,14 +165,18 @@ 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. */
+  /**
+   * Host path of a file holding the complete stream, when the source
+   * currently keeps an intact one. Reported on every read, so the registry
+   * tracks it as source metadata: a later read without it withdraws the file.
+   */
   spillPath?: string
 }
 ```
 
 ## The output ring
 
-Every job owns one bounded ring. Pull sources are pumped into it and `JobHandle.append` pushes land whole; the model consumes the ring through a registry-kept cursor (`CallerJobs.read`), any number of observers read it at absolute byte offsets (`CallerJobs.readAt`), and neither disturbs the other. `JobChannel` labels `stdout`, `stderr`, and `log`; `log` is producer narration that reaches observers only, never the model's consuming read. Settlement ends the stream and trims retention to the settled cap — the ring has no separate lifecycle. Browsers reach the roster and the ring through `job.rows` and `job.observe`, the Remote streams of [`dsh-api-job-controller`](../../packages/api/job-controller/README.md), whose frames are listed under its Cordis API section below.
+Every job owns one bounded ring. Pull sources are pumped into it and `JobHandle.append` pushes land whole; the model consumes the ring through a registry-kept cursor (`CallerJobs.read`), any number of observers read it at absolute byte offsets (`CallerJobs.readAt`), and neither disturbs the other. `JobChannel` labels `stdout`, `stderr`, and `log`; `log` is producer narration that reaches observers only, never the model's consuming read. Settlement ends the stream and trims retention to the settled cap — the ring has no separate lifecycle. The spill file a pull source keeps is job metadata (`JobView.output.spillPaths`, refreshed by every pump read), not chunk provenance, so the model's dropped-output notice names it after the ring evicted the bytes and even after the gap chunk itself is gone. Browsers reach the roster and the ring through `job.rows` and `job.observe`, the Remote streams of [`dsh-api-job-controller`](../../packages/api/job-controller/README.md), whose frames are listed under its Cordis API section below.
 
 ```ts type-equiv
 /** One chunk of a job's output ring: absolute offset, text, and its provenance. */
@@ -185,8 +189,6 @@ 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
 }
 ```
 
@@ -243,11 +245,15 @@ interface JobView {
   /** Epoch ms when the job settled; absent while live. */
   readonly finishedAt?: number
   /**
-   * The output ring's absolute coordinates. `total` is the offset the next
-   * chunk starts at (0 while nothing was written); `earliest` is the oldest
-   * retained byte, greater than zero exactly when retention dropped the head.
+   * The output ring's absolute coordinates and the complete-stream files
+   * behind it. `total` is the offset the next chunk starts at (0 while
+   * nothing was written); `earliest` is the oldest retained byte, greater
+   * than zero exactly when retention dropped the head. `spillPaths` lists the
+   * spill files the job's pull sources currently keep, in source order and
+   * deduplicated, and is absent while no source keeps one: it outlives any
+   * chunk, so a reader below `earliest` can still name where the bytes went.
    */
-  readonly output: { readonly total: number; readonly earliest: number }
+  readonly output: { readonly total: number; readonly earliest: number; readonly spillPaths?: readonly string[] }
 }
 ```
 

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, the lossy flag, and the spill path when one exists.
+   * @returns the delta text, the next offset, the lossy flag, and the spill path the source currently keeps.
    */
   read(fromByte: number): JobSourceRead
 }
@@ -165,14 +165,18 @@ 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. */
+  /**
+   * Host path of a file holding the complete stream, when the source
+   * currently keeps an intact one. Reported on every read, so the registry
+   * tracks it as source metadata: a later read without it withdraws the file.
+   */
   spillPath?: string
 }
 ```
 
 ## 输出环
 
-每个 job 拥有一个有界的环。拉取源被泵入其中，`JobHandle.append` 的推送整块落地；模型通过注册表保管的游标（`CallerJobs.read`）消耗该环，任意数量的观察者按绝对字节偏移读取它（`CallerJobs.readAt`），二者互不干扰。`JobChannel` 标记 `stdout`、`stderr` 与 `log`；`log` 是只到达观察者、从不进入模型消耗式读取的生产方叙述。结算即封流并把保留量裁剪到结算上限——环没有独立的生命周期。浏览器通过 [`dsh-api-job-controller`](../../packages/api/job-controller/README.zh.md) 的 Remote 流 `job.rows` 与 `job.observe` 触达名册与环，其帧列于下文的 Cordis API 一节。
+每个 job 拥有一个有界的环。拉取源被泵入其中，`JobHandle.append` 的推送整块落地；模型通过注册表保管的游标（`CallerJobs.read`）消耗该环，任意数量的观察者按绝对字节偏移读取它（`CallerJobs.readAt`），二者互不干扰。`JobChannel` 标记 `stdout`、`stderr` 与 `log`；`log` 是只到达观察者、从不进入模型消耗式读取的生产方叙述。结算即封流并把保留量裁剪到结算上限——环没有独立的生命周期。拉取源保留的 spill 文件是 job 元数据（`JobView.output.spillPaths`，每次泵读取都会刷新），不是 chunk 的来源信息，所以环淘汰了字节之后、甚至缺口 chunk 本身也被淘汰之后，模型的丢失输出提示仍能点名该文件。浏览器通过 [`dsh-api-job-controller`](../../packages/api/job-controller/README.zh.md) 的 Remote 流 `job.rows` 与 `job.observe` 触达名册与环，其帧列于下文的 Cordis API 一节。
 
 ```ts type-equiv
 /** One chunk of a job's output ring: absolute offset, text, and its provenance. */
@@ -185,8 +189,6 @@ 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
 }
 ```
 
@@ -243,11 +245,15 @@ interface JobView {
   /** Epoch ms when the job settled; absent while live. */
   readonly finishedAt?: number
   /**
-   * The output ring's absolute coordinates. `total` is the offset the next
-   * chunk starts at (0 while nothing was written); `earliest` is the oldest
-   * retained byte, greater than zero exactly when retention dropped the head.
+   * The output ring's absolute coordinates and the complete-stream files
+   * behind it. `total` is the offset the next chunk starts at (0 while
+   * nothing was written); `earliest` is the oldest retained byte, greater
+   * than zero exactly when retention dropped the head. `spillPaths` lists the
+   * spill files the job's pull sources currently keep, in source order and
+   * deduplicated, and is absent while no source keeps one: it outlives any
+   * chunk, so a reader below `earliest` can still name where the bytes went.
    */
-  readonly output: { readonly total: number; readonly earliest: number }
+  readonly output: { readonly total: number; readonly earliest: number; readonly spillPaths?: readonly string[] }
 }
 ```
 

packages/api/job-controller/README.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 packages/api/job-controller/README.md
-README.md: bfe93770ea86067fe4e2ee46d168ed91b7c266d7
-README.zh.md: 1ac5a6b19500aa073d812b35326fcf48c41f7e53
+README.md: 80ecae6518412aa628fb105125d4aca87cae0392
+README.zh.md: a0d4060fd41853678838daa6eea0e02c34fb9679

packages/api/job-controller/README.md

@@ -26,7 +26,7 @@ The Host controller requires the live Agent registry and the job registry (`dsh-
 
 `job.kill({ sessionId, jobId })` cancels one job the session can see with the reason `cancelled by the user`, which the registry merges into the killed job's detail. It claims nothing in the model's notice ledger — that ledger lives in `dsh-tool-jobs`, where only the model's own `job_kill` and waits claim — so the owning agent's completion notice stays due, reason included. It answers `{ outcome: 'requested' }` or `{ outcome: 'already-finished' }`, rejects an id the session cannot see as `job/not-found`, and applies the subagent ownership fence exactly as `session.cancel` does.
 
-The Client entry provides `ClientJobs` (`ctx.jobs`) over `ClientJobsModel`. `kill(sessionId, jobId)` forwards to `job.kill` and returns the Remote result for the caller's admission verdict. `watchRows(sessionId)` keeps one roster stream open per watched session however many viewers hold it and drops the rows after the last release; a reconnect's first frame is already the whole truth. `observe(sessionId, jobId)` opens one Gateway stream per job however many viewers expand it, keeps a bounded render tail per job with `gapBefore` marking eviction or resume gaps, closes the view on the terminal frame or records a stream failure on it, and drops the view after the last viewer releases. The plugin resolves the Gateway stream factory and the `job` namespace while its own context is current, because stream (re)opens run on caller stacks whose dynamic context has not declared `remote.job`.
+The Client entry installs `ctx.jobs` (`IJobs`), backed by the package-internal `ClientJobsModel`. `kill(sessionId, jobId)` forwards to `job.kill` and returns the Remote result for the caller's admission verdict. `watchRows(sessionId)` keeps one roster stream open per watched session however many viewers hold it and drops the rows after the last release; a reconnect's first frame is already the whole truth. `observe(sessionId, jobId)` opens one Gateway stream per job however many viewers expand it, keeps a bounded render tail per job with `gapBefore` marking eviction or resume gaps, closes the view on the terminal frame or records a stream failure on it, and drops the view after the last viewer releases. The plugin resolves the Gateway stream factory and the `job` namespace while its own context is current, because stream (re)opens run on caller stacks whose dynamic context has not declared `remote.job`.
 
 ### Config
 

packages/api/job-controller/README.zh.md

@@ -26,7 +26,7 @@ Host 控制器要求活体 Agent 注册表与 job 注册表（已发布组合里
 
 `job.kill({ sessionId, jobId })` 以 `cancelled by the user` 为原因取消一个该会话看得到的 job，注册表把该原因合并进被杀 job 的 detail。它不在模型的播报台账里认领任何东西——台账在 `dsh-tool-jobs` 里，只有模型自己的 `job_kill` 与等待会认领——因此拥有者 agent 的完成通知照常送达，并带上原因。它回答 `{ outcome: 'requested' }` 或 `{ outcome: 'already-finished' }`，对该会话看不到的 id 以 `job/not-found` 拒绝，并与 `session.cancel` 一样施加 subagent 所有权围栏。
 
-Client 入口在 `ClientJobsModel` 之上提供 `ClientJobs`（`ctx.jobs`）。`kill(sessionId, jobId)` 转发到 `job.kill` 并把 Remote 结果交给调用方判定准入。`watchRows(sessionId)` 不论多少查看器持有都只为每个被关注的会话开一条名册流，最后一个释放后丢弃行；重连后的首帧已经是完整事实。`observe(sessionId, jobId)` 不论多少查看器展开同一 job 都只开一条 Gateway 流，按 job 保留有界的渲染尾部并用 `gapBefore` 标记淘汰或续读缺口，在终态帧上关闭视图或把流失败记到视图上，最后一个查看器释放后丢弃视图。插件在自己的上下文仍是当前上下文时解析 Gateway 流工厂与 `job` namespace，因为流的（重）开启跑在未声明 `remote.job` 的调用栈上。
+Client 入口安装 `ctx.jobs`（`IJobs`），由包内部的 `ClientJobsModel` 支撑。`kill(sessionId, jobId)` 转发到 `job.kill` 并把 Remote 结果交给调用方判定准入。`watchRows(sessionId)` 不论多少查看器持有都只为每个被关注的会话开一条名册流，最后一个释放后丢弃行；重连后的首帧已经是完整事实。`observe(sessionId, jobId)` 不论多少查看器展开同一 job 都只开一条 Gateway 流，按 job 保留有界的渲染尾部并用 `gapBefore` 标记淘汰或续读缺口，在终态帧上关闭视图或把流失败记到视图上，最后一个查看器释放后丢弃视图。插件在自己的上下文仍是当前上下文时解析 Gateway 流工厂与 `job` namespace，因为流的（重）开启跑在未声明 `remote.job` 的调用栈上。
 
 ### 配置
 

packages/api/job-controller/src/client/index.ts

@@ -12,10 +12,11 @@ import { ClientJobsModel } from './model.ts'
 import { ClientJobs } from './service.ts'
 import type { JobsRemote } from './service.ts'
 
-export { ClientJobsModel } from './model.ts'
-export type { JobsSnapshot, JobsSource, ObservedJob } from './model.ts'
-export { ClientJobs } from './service.ts'
-export type { IJobs, JobRemote, JobsRemote } from './service.ts'
+export type { JobsSnapshot, ObservedJob } from './model.ts'
+// The `ctx.jobs` contract. Its module also carries the Context augmentation
+// that declares `ctx.jobs`, which reaches consumers only through this export:
+// declaration emit drops the value import above.
+export type { IJobs } from './service.ts'
 export type {
   JobChunk, JobKillRequest, JobKillValue, JobObserveFrame, JobObserveRequest, JobRowsFrame, JobRowsRequest, JobView,
 } from '../types.ts'

packages/extensions/tool-cordis/src/api-catalog.ts

@@ -4303,7 +4303,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
   },
   {
     name: 'JobAppendOptions',
-    declaration: 'export interface JobAppendOptions {\n    channel?: JobChannel;\n    gapBefore?: true;\n    spillPath?: string;\n}',
+    declaration: 'export interface JobAppendOptions {\n    channel?: JobChannel;\n    gapBefore?: true;\n}',
   },
   {
     name: 'JobChannel',
@@ -4311,7 +4311,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
   },
   {
     name: 'JobChunk',
-    declaration: 'export interface JobChunk {\n    readonly at: number;\n    readonly text: string;\n    readonly channel?: JobChannel;\n    readonly gapBefore?: true;\n    readonly spillPath?: string;\n}',
+    declaration: 'export interface JobChunk {\n    readonly at: number;\n    readonly text: string;\n    readonly channel?: JobChannel;\n    readonly gapBefore?: true;\n}',
   },
   {
     name: 'JobEvent',
@@ -4411,7 +4411,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
   },
   {
     name: 'JobView',
-    declaration: 'export interface JobView {\n    readonly id: JobId;\n    readonly kind: string;\n    readonly label: string;\n    readonly owner?: SessionId;\n    readonly outputLimitBytes?: number;\n    readonly status: JobStatus;\n    readonly progress?: string;\n    readonly detail?: string;\n    readonly startedAt: number;\n    readonly finishedAt?: number;\n    readonly output: {\n        readonly total: number;\n        readonly earliest: number;\n    };\n}',
+    declaration: 'export interface JobView {\n    readonly id: JobId;\n    readonly kind: string;\n    readonly label: string;\n    readonly owner?: SessionId;\n    readonly outputLimitBytes?: number;\n    readonly status: JobStatus;\n    readonly progress?: string;\n    readonly detail?: string;\n    readonly startedAt: number;\n    readonly finishedAt?: number;\n    readonly output: {\n        readonly total: number;\n        readonly earliest: number;\n        readonly spillPaths?: readonly string[];\n    };\n}',
   },
   {
     name: 'JsonSchemaNode',

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

@@ -106,6 +106,13 @@ interface TrackedJob {
   waitResolvers: Set<() => void>
   /** The registry-owned pump over the spec's pull sources, when it named any. */
   pump: PumpHandle | undefined
+  /**
+   * The spill file each pull source reported on its latest read, by source
+   * index; an entry is undefined while that source keeps none. Source
+   * metadata rather than chunk provenance, so it survives ring eviction and
+   * follows a source that withdraws its file.
+   */
+  spillPaths: (string | undefined)[]
 }
 
 /** True for the three terminal {@link JobStatus} values. */
@@ -255,6 +262,7 @@ export class LocalJobRegistry extends JobRegistry {
       markSettled,
       waitResolvers: new Set(),
       pump: undefined,
+      spillPaths: [],
     }
     // Binding the shared producer state is the commit: writes staged inside
     // the starter are already in the ring and `state`, and every later handle
@@ -280,7 +288,10 @@ export class LocalJobRegistry extends JobRegistry {
     if (spec.output !== undefined && spec.output.length > 0) {
       job.pump = startPump(
         spec.output.map(source => this.guardSource(job, source)),
-        (text, options) => { this.appendRing(state, ring, text, options, 'pump') },
+        {
+          append: (text, options) => { this.appendRing(state, ring, text, options, 'pump') },
+          spill: (index, path) => { job.spillPaths[index] = path },
+        },
         this.pumpPollMs,
         Promise.race([producerDone, settled]),
       )
@@ -382,6 +393,7 @@ export class LocalJobRegistry extends JobRegistry {
   /** Project a fresh read-only view from the mutable record. */
   private view(job: TrackedJob): JobView {
     const owner = job.owner?.id
+    const spillPaths = [...new Set(job.spillPaths.filter((path): path is string => path !== undefined))]
     return {
       id: job.id,
       kind: job.kind,
@@ -393,7 +405,11 @@ export class LocalJobRegistry extends JobRegistry {
       ...job.detail !== undefined ? { detail: job.detail } : {},
       startedAt: job.startedAt,
       ...job.finishedAt !== undefined ? { finishedAt: job.finishedAt } : {},
-      output: { total: job.ring.total, earliest: job.ring.earliest },
+      output: {
+        total: job.ring.total,
+        earliest: job.ring.earliest,
+        ...spillPaths.length > 0 ? { spillPaths } : {},
+      },
     }
   }
 

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

@@ -2,9 +2,10 @@
  * 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. 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.
+ * closes it. A lossy source read lands as a gap chunk; the spill file a source
+ * keeps is reported to the sink on every read, because that reference
+ * outlives any chunk. Pure utility — no cordis, no timers retained past
+ * settlement.
  * @module @deepseek-ai/dsh-jobs-local/pump
  */
 
@@ -15,6 +16,23 @@ export interface PumpHandle {
   done: Promise<void>
 }
 
+/** Where a pump delivers what it reads. */
+export interface PumpSink {
+  /**
+   * Append copied text to the ring.
+   * @param text - the chunk text, exactly as read.
+   * @param options - stream label and gap marker.
+   */
+  append(text: string, options?: JobAppendOptions): void
+  /**
+   * Record the spill file source `index` advertised on its latest read; called
+   * on every read, with `undefined` when the source keeps none or withdrew it.
+   * @param index - the source's position in the pump's source array.
+   * @param path - the host path the source reported, if any.
+   */
+  spill(index: number, path: string | undefined): void
+}
+
 /**
  * Drain every source in array order, sleep `pollMs` or until `until`
  * settles, repeat, then drain one final time. A lossy source read appends
@@ -27,14 +45,14 @@ export interface PumpHandle {
  * The wait holds constant resources however long the job runs: one
  * subscription on `until` for the whole run and one pending timer at a time.
  * @param sources - producer streams, each pumped at its own offset.
- * @param append - the ring append; receives each copied chunk.
+ * @param sink - receives each copied chunk and each source's current spill file.
  * @param pollMs - poll interval in milliseconds; a positive finite number.
  * @param until - settles (or rejects, which counts as settlement) when the producer finished.
  * @returns the handle whose `done` resolves after the final drain.
  */
 export function startPump(
   sources: readonly JobOutputSource[],
-  append: (text: string, options?: JobAppendOptions) => void,
+  sink: PumpSink,
   pollMs: number,
   until: Promise<unknown>,
 ): PumpHandle {
@@ -43,18 +61,18 @@ export function startPump(
   }
   const states = sources.map(source => ({ source, cursor: 0 }))
   const drain = (): void => {
-    for (const state of states) {
+    for (const [index, state] of states.entries()) {
       const { source } = state
       const read = source.read(state.cursor)
       state.cursor = read.nextOffset
+      sink.spill(index, read.spillPath)
       if (read.text.length === 0) continue
       if (source.channel === undefined && !read.lossy) {
-        append(read.text)
+        sink.append(read.text)
       } else {
-        append(read.text, {
+        sink.append(read.text, {
           ...source.channel !== undefined ? { channel: source.channel } : {},
           ...read.lossy ? { gapBefore: true as const } : {},
-          ...read.lossy && read.spillPath !== undefined ? { spillPath: read.spillPath } : {},
         })
       }
     }

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

@@ -14,7 +14,6 @@ interface RingChunk {
   bytes: number
   channel?: JobChannel
   gapBefore?: true
-  spillPath?: string
 }
 
 /**
@@ -62,7 +61,6 @@ 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
@@ -108,7 +106,6 @@ 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 }

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

@@ -1194,6 +1194,35 @@ describe('LocalJobRegistry output ring', () => {
     expect(() => jobsOf(ctx).readAt(JobId('bash-99'), 0)).toThrow(/unknown job/)
   })
 
+  it('keeps the spill file each source currently reports as job metadata, beyond any chunk', async () => {
+    const ctx = await harness({ retainBytes: 4, pumpPollMs: 5 })
+    let withdrawn = false
+    const out: JobOutputSource = {
+      channel: 'stdout',
+      read: from => from === 0
+        ? { text: 'abcdefgh', nextOffset: 8, lossy: false, spillPath: '/spill/out.log' }
+        : { text: '', nextOffset: from, lossy: false, ...withdrawn ? {} : { spillPath: '/spill/out.log' } },
+    }
+    const err: JobOutputSource = {
+      channel: 'stderr',
+      read: from => ({ text: '', nextOffset: from, lossy: false, spillPath: '/spill/err.log' }),
+    }
+    const p = producer({ output: [out, err] })
+    const id = ctx.jobs.start(p.spec)
+    // The first drain ran inside start(): the live cap already evicted the head, yet the
+    // projection still names the files — no chunk carried them.
+    const view = jobsOf(ctx).get(id)
+    expect(view.output).toEqual({ total: 8, earliest: 4, spillPaths: ['/spill/out.log', '/spill/err.log'] })
+    expect(jobsOf(ctx).readAt(id, 0).chunks).toEqual([{ at: 4, text: 'efgh', channel: 'stdout', gapBefore: true }])
+
+    // A source that stops naming its file (a failed seal) withdraws it on the next drain.
+    withdrawn = true
+    await new Promise<void>(resolve => setTimeout(resolve, 20))
+    expect(jobsOf(ctx).get(id).output.spillPaths).toEqual(['/spill/err.log'])
+    p.settle({ status: 'completed' })
+    await tick()
+  })
+
   it('announces registration before the pump drains a source that is readable at once', async () => {
     const ctx = await harness({ pumpPollMs: 5 })
     const events = collect(ctx)

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

@@ -1,14 +1,17 @@
 import { describe, expect, it, vi } from 'vitest'
 import type { JobAppendOptions, JobOutputSource } from '@deepseek-ai/dsh-jobs'
 import { startPump } from '../src/pump.ts'
+import type { PumpSink } from '../src/pump.ts'
 
-/** Record every append so pump behavior is observable without a registry. */
-function sink() {
+/** Record every append and spill report so pump behavior is observable without a registry. */
+function recorder() {
   const appends: { text: string; options?: JobAppendOptions }[] = []
-  return {
-    appends,
-    append: (text: string, options?: JobAppendOptions) => { appends.push({ text, ...options !== undefined ? { options } : {} }) },
+  const spills: [number, string | undefined][] = []
+  const sink: PumpSink = {
+    append: (text, options) => { appends.push({ text, ...options !== undefined ? { options } : {} }) },
+    spill: (index, path) => { spills.push([index, path]) },
   }
+  return { appends, spills, sink }
 }
 
 /** A scripted source: each read() shifts the next scripted result. */
@@ -39,12 +42,12 @@ describe('startPump', () => {
   it('copies labeled deltas at the poll cadence and resumes each source at its own offset', async () => {
     vi.useFakeTimers()
     try {
-      const { appends, append } = sink()
+      const { appends, sink } = recorder()
       let settle!: () => void
       const until = new Promise<void>((resolve) => { settle = resolve })
       const out = scriptedSource([{ text: 'a' }, { text: 'bc' }], 'stdout')
       const err = scriptedSource([{ text: '' }, { text: 'E' }], 'stderr')
-      const pump = startPump([out.source, err.source], append, 50, until)
+      const pump = startPump([out.source, err.source], sink, 50, until)
 
       await vi.advanceTimersByTimeAsync(50)
       settle()
@@ -65,36 +68,57 @@ describe('startPump', () => {
   })
 
   it('marks a lossy source read as a gap so observers see the discontinuity', async () => {
-    const { appends, append } = sink()
+    const { appends, sink } = recorder()
     const { source } = scriptedSource([{ text: 'tail', lossy: true }])
-    await startPump([source], append, 1, Promise.resolve()).done
+    await startPump([source], sink, 1, Promise.resolve()).done
     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("reports each source's spill file on every read and withdraws it once a read stops naming it", async () => {
+    vi.useFakeTimers()
+    try {
+      const { appends, spills, sink } = recorder()
+      let settle!: () => void
+      const until = new Promise<void>((resolve) => { settle = resolve })
+      const out = scriptedSource([
+        { text: 'tail', lossy: true, spillPath: '/spill/out.log' },
+        { text: 'more', spillPath: '/spill/out.log' },
+        { text: '' },
+      ], 'stdout')
+      const err = scriptedSource([{ text: '' }, { text: 'E', spillPath: '/spill/err.log' }], 'stderr')
+      const pump = startPump([out.source, err.source], sink, 50, until)
+
+      await vi.advanceTimersByTimeAsync(50)
+      settle()
+      await pump.done
+      // Chunks carry the gap marker only; the file reference is source metadata.
+      expect(appends).toEqual([
+        { text: 'tail', options: { channel: 'stdout', gapBefore: true } },
+        { text: 'more', options: { channel: 'stdout' } },
+        { text: 'E', options: { channel: 'stderr' } },
+      ])
+      // Every read reports, lossy or clean, empty or not: the third stdout read withdraws the file.
+      expect(spills).toEqual([
+        [0, '/spill/out.log'], [1, undefined],
+        [0, '/spill/out.log'], [1, '/spill/err.log'],
+        [0, undefined], [1, undefined],
+      ])
+    } finally {
+      vi.useRealTimers()
+    }
   })
 
   it('treats a rejected settlement as settlement and still drains the final bytes', async () => {
-    const { appends, append } = sink()
+    const { appends, sink } = recorder()
     const { source } = scriptedSource([{ text: 'last' }])
-    await startPump([source], append, 1, Promise.reject(new Error('producer broke'))).done
+    await startPump([source], sink, 1, Promise.reject(new Error('producer broke'))).done
     expect(appends).toEqual([{ text: 'last' }])
   })
 
   it('subscribes to the settlement once and holds one poll timer however long the job runs', async () => {
     vi.useFakeTimers()
     try {
-      const { append } = sink()
+      const { sink } = recorder()
       let subscriptions = 0
       let settle!: () => void
       const settled = new Promise<void>((resolve) => { settle = resolve })
@@ -105,7 +129,7 @@ describe('startPump', () => {
         },
       }) as unknown as Promise<unknown>
       const { source, offsets } = scriptedSource([])
-      const pump = startPump([source], append, 50, counting(settled))
+      const pump = startPump([source], sink, 50, counting(settled))
 
       const rounds = 10_000
       await vi.advanceTimersByTimeAsync(50 * rounds)
@@ -123,7 +147,7 @@ describe('startPump', () => {
   })
 
   it('rejects a non-positive poll interval before touching any source', () => {
-    const { append } = sink()
-    expect(() => startPump([], append, 0, Promise.resolve())).toThrow(/invalid pump pollMs/)
+    const { sink } = recorder()
+    expect(() => startPump([], sink, 0, Promise.resolve())).toThrow(/invalid pump pollMs/)
   })
 })

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, spillPath: '/spill/err.log' }, 1024)).toBe(true)
+    expect(ring.append('wörld', { channel: 'stderr', gapBefore: true }, 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, spillPath: '/spill/err.log' },
+        { at: 6, text: 'wörld', channel: 'stderr', gapBefore: true },
       ],
       next: 12,
       lossy: false,

packages/jobs/jobs/src/types.ts

@@ -38,7 +38,11 @@ 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. */
+  /**
+   * Host path of a file holding the complete stream, when the source
+   * currently keeps an intact one. Reported on every read, so the registry
+   * tracks it as source metadata: a later read without it withdraws the file.
+   */
   spillPath?: string
 }
 
@@ -54,7 +58,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, the lossy flag, and the spill path when one exists.
+   * @returns the delta text, the next offset, the lossy flag, and the spill path the source currently keeps.
    */
   read(fromByte: number): JobSourceRead
 }
@@ -69,8 +73,6 @@ 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
 }
 
 /**

packages/jobs/jobs/src/view.ts

@@ -46,8 +46,6 @@ 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
 }
 
 /**
@@ -81,9 +79,13 @@ export interface JobView {
   /** Epoch ms when the job settled; absent while live. */
   readonly finishedAt?: number
   /**
-   * The output ring's absolute coordinates. `total` is the offset the next
-   * chunk starts at (0 while nothing was written); `earliest` is the oldest
-   * retained byte, greater than zero exactly when retention dropped the head.
+   * The output ring's absolute coordinates and the complete-stream files
+   * behind it. `total` is the offset the next chunk starts at (0 while
+   * nothing was written); `earliest` is the oldest retained byte, greater
+   * than zero exactly when retention dropped the head. `spillPaths` lists the
+   * spill files the job's pull sources currently keep, in source order and
+   * deduplicated, and is absent while no source keeps one: it outlives any
+   * chunk, so a reader below `earliest` can still name where the bytes went.
    */
-  readonly output: { readonly total: number; readonly earliest: number }
+  readonly output: { readonly total: number; readonly earliest: number; readonly spillPaths?: readonly string[] }
 }

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

@@ -21,9 +21,6 @@ import type {} from '@deepseek-ai/dsh-agent'
 import { 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']
 
@@ -34,6 +31,17 @@ export const inject = ['tools', 'jobs', 'systemPrompt']
  */
 export type CompletionDelivery = 'quiet' | 'wakeup'
 
+/**
+ * One model touch of a live job's settlement: a wait in flight or a kill the
+ * model requested. `live` says whether the tool result that made the claim
+ * will still carry the settlement; a wait's claim dies the moment its caller
+ * aborts, before the registry's rejection reaches the tool.
+ */
+interface Claim {
+  readonly id: JobId
+  live: boolean
+}
+
 /** Configures bounded `job_output` waits and completion-notice delivery. */
 export interface Config {
   /** Wait duration applied when `job_output` sets `wait` without `timeout_ms` (default 30s). */
@@ -182,7 +190,7 @@ function isTerminal(job: JobView): boolean {
 /** The consuming read as the model sees it: the delta, then the result once, then the status line. */
 function readBody(jobs: CallerJobs, id: JobId): { text: string; job: PublicJobSnapshot } {
   const read = jobs.read(id)
-  const delta = renderModelDelta(read.chunks, read.lossy)
+  const delta = renderModelDelta(read.chunks, read.lossy, read.job.output.spillPaths ?? [])
   const text = read.result === undefined
     ? delta
     : `${delta}${delta.length > 0 && !delta.endsWith('\n') ? '\n' : ''}${read.result}`
@@ -257,23 +265,25 @@ export function apply(ctx: Context, config: Config): void {
   // 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. Claims are counted per call, so a timed-out wait
-  // withdraws only its own claim while a concurrent wait on the same job
-  // keeps the settlement covered. The settlement event drops the entry, so
-  // the ledger holds only live jobs the model already touched.
-  const claims = new Map<JobId, number>()
-  const claim = (id: JobId): (() => void) => {
-    claims.set(id, (claims.get(id) ?? 0) + 1)
-    return () => {
-      // A withdraw runs only for a wait that returned or rejected while the
-      // job was live, so its own claim is still counted here; the fallback
-      // only discharges the Map's optional read.
-      /* v8 ignore next -- settlement hands every waiter the terminal view before dropping the entry; such waits never withdraw. */
-      const remaining = (claims.get(id) ?? 1) - 1
-      if (remaining > 0) claims.set(id, remaining)
-      else claims.delete(id)
+  // continuations run after the event. Each call holds its own claim, so a
+  // timed-out wait drops only its own while a concurrent wait on the same job
+  // keeps the settlement covered. The settlement event drops every claim on
+  // the job, so the ledger holds only live jobs the model already touched.
+  const claims = new Set<Claim>()
+  const claim = (id: JobId): Claim => {
+    const entry: Claim = { id, live: true }
+    claims.add(entry)
+    return entry
+  }
+  // Drop every claim on the job; true when one of them still covered the settlement.
+  const settleClaims = (id: JobId): boolean => {
+    let covered = false
+    for (const entry of claims) {
+      if (entry.id !== id) continue
+      claims.delete(entry)
+      if (entry.live) covered = true
     }
+    return covered
   }
 
   // A busy owner is injected: the notice waits in its next-step inbox, which
@@ -287,11 +297,11 @@ export function apply(ctx: Context, config: Config): void {
   // this listener owns delivery, not the choice of whom to deliver to.
   ctx.jobs.events.subscribe({ owners: 'scope' }, (event) => {
     if (event.type === 'removed') {
-      claims.delete(event.job.id)
+      settleClaims(event.job.id)
       return
     }
     if (event.type !== 'settled') return
-    const claimed = claims.delete(event.job.id)
+    const claimed = settleClaims(event.job.id)
     if (claimed || event.cause === 'teardown' || event.job.owner === undefined) return
     // The destination is the agent registered for the owner session now. An
     // owned job needed the agent registry to start, so the registry is only
@@ -352,13 +362,24 @@ export function apply(ctx: Context, config: Config): void {
       const jobs = ctx.jobs.forCaller(exec.agent?.id)
       if (args.wait === true) {
         const timeout = Math.min(args.timeout_ms ?? waitDefault, waitCap)
-        const withdraw = claim(id)
+        const entry = claim(id)
+        // The registry rejects an aborted wait on a later microtask, and a
+        // settlement announced in between must not find this claim live: the
+        // tool result carries the abort, so the notice is the model's only
+        // completion record. ToolRuntime never dispatches an aborted call, so
+        // the listener covers every abort. A timeout needs no such guard —
+        // the registry resolves it from a timer callback, and the wait's
+        // continuation runs before anything else can settle the job.
+        const abandon = (): void => { entry.live = false }
+        exec.signal.addEventListener('abort', abandon, { once: true })
         try {
           const view = await jobs.wait(id, timeout, exec.signal)
-          if (!isTerminal(view)) withdraw()
+          if (!isTerminal(view)) claims.delete(entry)
         } catch (error: unknown) {
-          withdraw()
+          claims.delete(entry)
           throw error
+        } finally {
+          exec.signal.removeEventListener('abort', abandon)
         }
       }
       return readBody(jobs, id)

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

@@ -66,20 +66,19 @@ export function statusLine(snapshot: Pick<PublicJobSnapshot, 'status' | 'detail'
  * narration for observers and never reach the model. Lost bytes — the cursor
  * fell behind the ring's retention, or a model-visible chunk carries a
  * producer-side gap — end the read with the shell tools' dropped-output
- * notice, naming the spill files the gap chunks point at.
+ * notice, naming the spill files the job's sources keep.
  * @param chunks - the chunks since the model cursor, in offset order.
  * @param lossy - whether bytes before `chunks` were evicted unread.
+ * @param spillPaths - the complete-stream files the job currently advertises (`JobView.output.spillPaths`).
  * @returns the delta text, possibly empty.
  */
-export function renderModelDelta(chunks: readonly JobChunk[], lossy: boolean): string {
+export function renderModelDelta(chunks: readonly JobChunk[], lossy: boolean, spillPaths: readonly string[]): string {
   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}` : '')
-  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)'}]`
+  if (!lossy && !visible.some(chunk => chunk.gapBefore === true)) return body
+  const notice = `[some output was dropped from memory; full output: ${spillPaths.length > 0 ? spillPaths.join(', ') : '(unavailable)'}]`
   return `${body}${body.length > 0 && !body.endsWith('\n') ? '\n' : ''}${notice}`
 }

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

@@ -10,9 +10,9 @@ 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 { JobAppendOptions, JobHandle, JobHooks, JobOutcome, JobSpec, JobView } from '@deepseek-ai/dsh-jobs'
+import type { JobAppendOptions, JobHandle, JobHooks, JobOutcome, JobOutputSource, JobSpec, JobView } from '@deepseek-ai/dsh-jobs'
 import * as ToolTasks from '@deepseek-ai/dsh-tool-jobs'
-import { publicJob, renderModelDelta, statusLine } from '@deepseek-ai/dsh-tool-jobs'
+import { publicJob, renderModelDelta, statusLine } from '../src/render.ts'
 
 const testToolSignal = new AbortController().signal
 
@@ -212,33 +212,60 @@ 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; full output: (unavailable)]')
+    expect(renderModelDelta([], false, [])).toBe('')
+    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' },
       { 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; full output: (unavailable)]')
-    expect(renderModelDelta([{ at: 0, text: 'line\n' }], true)).toBe('line\n[some output was dropped from memory; full output: (unavailable)]')
+    ], false, [])).toBe('ab\n[stderr]\ne1\ne2')
+    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)]')
+    // Ring eviction names the files the job's sources keep, whether or not a retained chunk carries a gap.
+    expect(renderModelDelta([{ at: 8, text: 'tail' }], true, ['/spill/out.log', '/spill/err.log']))
+      .toBe('tail\n[some output was dropped from memory; full output: /spill/out.log, /spill/err.log]')
   })
 
-  it('reports a producer-side gap as dropped output and names the spill files the gap chunks point at', () => {
+  it('reports a producer-side gap as dropped output and names the spill files the job advertises', () => {
     // 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))
+    expect(renderModelDelta([{ at: 0, text: 'head' }, { at: 4, text: 'tail', gapBefore: true }], false, ['/spill/out.log']))
       .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))
+      { at: 0, text: 'o', gapBefore: true },
+      { at: 1, text: 'e', channel: 'stderr', gapBefore: true },
+    ], false, ['/spill/out.log', '/spill/err.log'])).toBe('o\n[stderr]\ne\n[some output was dropped from memory; full output: /spill/out.log, /spill/err.log]')
+    // A gap while no source keeps a file yields the generic notice.
+    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('')
+    expect(renderModelDelta([{ at: 0, text: 'phase', channel: 'log', gapBefore: true }], false, ['/spill/log'])).toBe('')
+  })
+
+  it('names the spill file when the ring evicted output the model never read', async () => {
+    const { ctx } = await setup({}, { retainBytes: 8 })
+    const owner = fakeAgent(ctx, 'sess-1')
+    let settle!: (outcome: JobOutcome) => void
+    // The source never reads lossy: the pump keeps up, and only the ring's live cap drops bytes.
+    const source: JobOutputSource = {
+      channel: 'stdout',
+      read: from => from === 0
+        ? { text: 'x'.repeat(32), nextOffset: 32, lossy: false, spillPath: '/spill/out.log' }
+        : { text: '', nextOffset: from, lossy: false, spillPath: '/spill/out.log' },
+    }
+    ctx.jobs.start({
+      kind: 'bash',
+      label: 'noisy',
+      owner: owner.id,
+      output: [source],
+      run: () => ({ cancel() {}, done: new Promise<JobOutcome>((resolve) => { settle = resolve }) }),
+    })
+
+    expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }, owner)))
+      .toBe(`${'x'.repeat(8)}\n[some output was dropped from memory; full output: /spill/out.log]\n[status: running]`)
+    settle({ status: 'completed' })
+    await tick()
   })
 
   it('applies the built-in wait bounds when apply() receives a bare config', async () => {
@@ -1035,6 +1062,53 @@ describe('completion notices', () => {
     expect(inject).toHaveBeenCalledTimes(1)
   })
 
+  it('delivers the notice when the wait aborts and the job settles before the rejection lands', async () => {
+    const { ctx } = await setup()
+    const inject = vi.fn()
+    const owner = fakeAgent(ctx, 'sess-1', { inject })
+    const p = producer({ owner: owner.id, kind: 'subagent' })
+    ctx.jobs.start(p.spec)
+
+    const aborter = new AbortController()
+    const pending = ctx.tools.execute({
+      signal: aborter.signal,
+      callId: ToolCallId('call-abort-then-settle'),
+      name: 'job_output',
+      arguments: { job_id: 'subagent-1', wait: true },
+      agent: owner,
+    })
+    await tick()
+    // One synchronous span: the registry rejects the aborted wait on a later
+    // microtask, and a push producer's settlement lands before that rejection
+    // reaches the tool. The tool result carries the abort, not the outcome, so
+    // the settlement notice is the model's only completion record.
+    aborter.abort()
+    p.settle({ status: 'completed', result: 'answer' })
+    const result = await pending
+    expect(result.isError).toBe(true)
+    expect(text(result)).toContain('wait aborted')
+    await tick()
+    expect(inject).toHaveBeenCalledTimes(1)
+  })
+
+  it('keeps a claim on one job while another job settles', async () => {
+    const { ctx } = await setup()
+    const inject = vi.fn()
+    const owner = fakeAgent(ctx, 'sess-1', { inject })
+    const killed = producer({ owner: owner.id })
+    const other = producer({ owner: owner.id })
+    ctx.jobs.start(killed.spec)
+    ctx.jobs.start(other.spec)
+
+    await call(ctx, 'job_kill', { job_id: 'bash-1' }, owner)
+    other.settle({ status: 'completed' })
+    await tick()
+    expect(inject).toHaveBeenCalledTimes(1)
+    killed.settle({ status: 'killed' })
+    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.