fix(job-controller): close an observation on the job's removal instead of a failed read

An owner's teardown settles and removes the record within one microtask chain, while the observation generation sleeps its flush window after the wake and then reads an id the registry no longer returns, failing the stream and flashing an error in the panel until the roster frame drops the row. The subscription now records a removed event for the observed id and the generation closes with that terminal projection.

.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: 50dd181d3a36e2a8980670565982588598b269a5
-2026-09-03-jobs-seam-consolidation.zh.md: da15d808ae229034c0266dce7c673254da8b06e8
+2026-09-03-jobs-seam-consolidation.md: 2aa575a124d75437c75b30b96337450704b2bf75
+2026-09-03-jobs-seam-consolidation.zh.md: 304e424d9936c4ab5b751b511dbd89ef71f409ed

.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.
+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.

.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)要求的交叉观察；先前只针对单个视图的字段校验已删除。
+合并 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 的终态投影收尾，而不是一次失败的读取。

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: f4f7a5142e649d0b8335128f2c4012dd957ec676
-README.zh.md: eeb10819fc2ab17f1e2603d5014618874c5efaf7
+README.md: f38d29b40caf57821d66ce6d8d39c6215014c7be
+README.zh.md: 7a0fab218eb67ff6220cbbdff124555bc0b762c8

packages/api/job-controller/README.md

@@ -22,7 +22,7 @@ English | [中文](README.zh.md)
 <a id="use-this-package"></a>
 ## Use this package
 
-The Host controller requires the job registry (`dsh-jobs-local` in the shipped compositions) and fails to load without it. `job.rows({ sessionId })` yields the session's visible set — its own jobs plus every unowned job — once on open and again after each coalesced burst of lifecycle commits (registration, progress, stopping, settlement, removal); output appends never refresh the roster, since a settled projection already carries the final byte count. `job.observe({ sessionId?, jobId, from? })` reads through `visibleTo(sessionId)` — an unowned job needs no session — and yields one `opened` anchor carrying the job projection, coalesced `output` frames, then one terminal `status` once the job has settled and its ring is drained, after which the stream closes normally. Both reads are non-consuming: the model's `job_output` cursor and notice state never observe them.
+The Host controller requires the job registry (`dsh-jobs-local` in the shipped compositions) and fails to load without it. `job.rows({ sessionId })` yields the session's visible set — its own jobs plus every unowned job — once on open and again after each coalesced burst of lifecycle commits (registration, progress, stopping, settlement, removal); output appends never refresh the roster, since a settled projection already carries the final byte count. `job.observe({ sessionId?, jobId, from? })` reads through `visibleTo(sessionId)` — an unowned job needs no session — and yields one `opened` anchor carrying the job projection, coalesced `output` frames, then one terminal `status` once the job has settled and its ring is drained, after which the stream closes normally; a removal announced mid-stream (the owner's teardown) closes it with the removed job's terminal projection. Both reads are non-consuming: the model's `job_output` cursor and notice state never observe them.
 
 The Client entry provides `ClientJobs` (`ctx.jobs`) over `ClientJobsModel`. `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`.
 

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

@@ -22,7 +22,7 @@ kind: "package-reference"
 <a id="use-this-package"></a>
 ## 使用本包
 
-Host 控制器要求 job 注册表（已发布组合里是 `dsh-jobs-local`），缺少则不加载。`job.rows({ sessionId })` 产出该会话可见的集合——自己的 job 加上所有无主 job——打开时一次，之后每一轮聚合过的生命周期提交（注册、进度、停止中、结算、移除）后再一次；输出追加从不刷新名册，因为结算后的投影已经带着最终字节数。`job.observe({ sessionId?, jobId, from? })` 经 `visibleTo(sessionId)` 读取——无主 job 不需要 session——先产出一个携带 job 投影的 `opened` 锚帧，再是聚合的 `output` 帧，job 结算且环排干后产出一个终态 `status`，随后流正常关闭。两种读取都是非消耗的：模型侧 `job_output` 游标与完成播报状态永远观察不到它们。
+Host 控制器要求 job 注册表（已发布组合里是 `dsh-jobs-local`），缺少则不加载。`job.rows({ sessionId })` 产出该会话可见的集合——自己的 job 加上所有无主 job——打开时一次，之后每一轮聚合过的生命周期提交（注册、进度、停止中、结算、移除）后再一次；输出追加从不刷新名册，因为结算后的投影已经带着最终字节数。`job.observe({ sessionId?, jobId, from? })` 经 `visibleTo(sessionId)` 读取——无主 job 不需要 session——先产出一个携带 job 投影的 `opened` 锚帧，再是聚合的 `output` 帧，job 结算且环排干后产出一个终态 `status`，随后流正常关闭；流中若通告了移除（拥有者 teardown），则以被移除 job 的终态投影收尾。两种读取都是非消耗的：模型侧 `job_output` 游标与完成播报状态永远观察不到它们。
 
 Client 入口在 `ClientJobsModel` 之上提供 `ClientJobs`（`ctx.jobs`）。`watchRows(sessionId)` 不论多少查看器持有都只为每个被关注的会话开一条名册流，最后一个释放后丢弃行；重连后的首帧已经是完整事实。`observe(sessionId, jobId)` 不论多少查看器展开同一 job 都只开一条 Gateway 流，按 job 保留有界的渲染尾部并用 `gapBefore` 标记淘汰或续读缺口，在终态帧上关闭视图或把流失败记到视图上，最后一个查看器释放后丢弃视图。插件在自己的上下文仍是当前上下文时解析 Gateway 流工厂与 `job` namespace，因为流的（重）开启跑在未声明 `remote.job` 的调用栈上。
 

packages/api/job-controller/src/observe.ts

@@ -1,7 +1,7 @@
 /** Per-job observation generations: anchor, coalesced output, terminal status. */
 
 import type { JobId } from '@deepseek-ai/dsh-jobs/brand'
-import type { JobChunk, JobRegistry, JobStatus } from '@deepseek-ai/dsh-jobs'
+import type { JobChunk, JobRegistry, JobStatus, JobView } from '@deepseek-ai/dsh-jobs'
 import type { JobObserveFrame, JobObserveRequest } from './types.ts'
 import { OutputWaiter, sleep } from './wake.ts'
 
@@ -25,7 +25,9 @@ function isTerminal(status: JobStatus): boolean {
  * Stream one job's retained output from an absolute offset: one `opened`
  * anchor, coalesced `output` frames as the ring advances, then one terminal
  * `status` after the settled job is drained, after which the generation
- * closes normally. Reads are non-consuming — the model-facing cursor and
+ * closes normally. A removal announced mid-generation (the owner's teardown)
+ * closes it with the removed job's terminal projection instead of a failed
+ * read. Reads are non-consuming — the model-facing cursor and
  * notice state never observe them; reconnecting callers resume by passing
  * the last frame's `next` as `from`. The request's session is the fenced
  * read's caller: the registry rejects a job the session cannot see and an
@@ -50,11 +52,17 @@ export async function* observeJobOutput(
   // than value-importing the registry package's constructor.
   const id = String(request.jobId) as JobId
   const waiter = new OutputWaiter()
+  // A removal announced during the generation: the owner's teardown settled
+  // and dropped the record, so the projection it announced is the last word
+  // and the read after the flush window would only find an unknown id.
+  let removed: JobView | undefined
   // Subscribe before the first read so an append between the anchor read and
   // the wait cannot be missed.
   const unsubscribe = registry.events.subscribe({ owners: 'all' }, (event) => {
     const changed = event.type === 'output' ? event.id : event.job.id
-    if (changed === id) waiter.wake()
+    if (changed !== id) return
+    if (event.type === 'removed') removed = event.job
+    waiter.wake()
   })
   try {
     const visible = registry.visibleTo(request.sessionId)
@@ -62,6 +70,10 @@ export async function* observeJobOutput(
     let cursor = request.from ?? job.output.earliest
     yield { type: 'opened', job, from: cursor }
     while (!signal.aborted) {
+      if (removed !== undefined) {
+        yield { type: 'status', job: removed }
+        return
+      }
       const read = visible.readAt(id, cursor)
       if (read.chunks.length > 0 || read.lossy) {
         yield* outputFrames(read.chunks, read.next, read.lossy, options.maxFrameBytes)

packages/api/job-controller/tests/observe.host.spec.ts

@@ -181,6 +181,22 @@ describe('observeJobOutput', () => {
     await noisy.settle({ status: 'completed' })
   })
 
+  it("closes with the removed job's terminal projection when the owner's teardown drops it mid-observation", async () => {
+    const ctx = await harness()
+    const owner = registerAgent(ctx, 'observer-owner')
+    const job = startJob(ctx, { label: 'torn down', owner: owner.id })
+    job.append('partial')
+    const { frames, done } = observed(ctx, { jobId: job.id, sessionId: owner.id })
+    await wait(20)
+    // Teardown cancels, settles, and removes the record within one microtask
+    // chain; the generation wakes on the removal and closes without a read.
+    await owner.disposeScope()
+    await done
+    expect(frames.map(frame => frame.type)).toEqual(['opened', 'output', 'status'])
+    expect(frames.at(-1)).toMatchObject({ type: 'status', job: { id: job.id, status: 'killed' } })
+    expect(() => ctx.jobs.visibleTo(owner.id).get(job.id)).toThrow(/unknown job/)
+  })
+
   it('stops cleanly on abort while waiting for output', async () => {
     const ctx = await harness()
     const job = startJob(ctx, { label: 'idle' })