fix(jobs): constant pump waits, quiescent client disposal, record flag on the roster

Review corrections on the observation record:

- `pumpJobOutput` subscribes to `done` once and keeps one pending timer;
  racing the same pending promise every poll round retained a reaction
  per round for the life of the job.
- The `ctx.jobOutput` effect disposer is async and awaits every open
  carrier's `dispose()`, so the fiber reports unloaded only once the old
  iterators are quiescent.
- `SessionJob.record: true` replaces `outputTotal`: `append` never commits
  a roster change, so the mirrored count was stale while output flowed;
  the live offsets ride the observation stream's `opened` anchor.
- The record is documented as a best-effort live preview whose
  cross-stream reordering is bounded by the poll interval.
- The web e2e scenarios hold their background job on a test-owned barrier
  file instead of a wall-clock sleep, and each is one test rather than a
  chain sharing a job id.

.agents/notes/implemented/architecture/2026-09-01-jobs-absorb-activity-record.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-01-jobs-absorb-activity-record.md
-2026-09-01-jobs-absorb-activity-record.md: aa147778d06133b61f79defae6f9f6807dd97f41
-2026-09-01-jobs-absorb-activity-record.zh.md: 12e8f4bbd909a480f1461433c58d7937a8d7c857
+2026-09-01-jobs-absorb-activity-record.md: c1a668f8a008948957e62b5acf1d9db141441885
+2026-09-01-jobs-absorb-activity-record.zh.md: a23f3cda4c8c53663a6aee99469297d6878539d1

.agents/notes/implemented/architecture/2026-09-01-jobs-absorb-activity-record.md

@@ -18,8 +18,8 @@ The one activity without a job was the foreground workflow mirror. Everything el
 - **There is no `end`.** Job settlement — the producer outcome, a kill, or teardown — is the record's only close: it trims retention to the settled cap and fires the final `onOutput` signal. The dual-settlement pairing (`ActivityHandle.end` first-wins against teardown force-ends) is deleted, not reimplemented.
 - **`readRecord(id, from, caller)`** is the non-consuming multi-reader view (absolute UTF-8 offsets, `lossy` below the retained window), fenced like every other job read; it never marks the job `reported`. The model-facing `readOutput` cursor is untouched — the two projections serve different readers and stay separate by design.
 - **`jobs-local`** absorbs the chunk ring (`retainBytes` 256 KiB live, `settledRetainBytes` 16 KiB settled), and `pumpJobOutput` replaces `pumpActivityOutput` beside the seam. Producers fold the pump's final drain into `hooks.done` so the record holds its last bytes before settlement closes it.
-- **The wire splits by role.** `SessionJob.outputTotal` (present exactly for record jobs) marks a row observable on the session control stream; `job.observe({sessionId?, jobId, from?})` on the `job` namespace of `api-job-controller` streams anchor/output/status frames with the fenced read resolved from the request's session. `api-activity-controller` is deleted; its roster stream is redundant (the jobs frames are the roster) and its observe machinery lives in `api-job-controller` (`observe.ts`, client `ctx.jobOutput`), which resolves its Remote faces in its own apply.
-- **`ui-activity` returns to its upstream name `ui-jobs`** and renders one roster: `jobsBySession` rows, expandable exactly when `outputTotal` is present. The two-roster join is deleted.
+- **The wire splits by role.** `SessionJob.record` (present exactly for record jobs) marks a row observable on the session control stream; `job.observe({sessionId?, jobId, from?})` on the `job` namespace of `api-job-controller` streams anchor/output/status frames with the fenced read resolved from the request's session. `api-activity-controller` is deleted; its roster stream is redundant (the jobs frames are the roster) and its observe machinery lives in `api-job-controller` (`observe.ts`, client `ctx.jobOutput`), which resolves its Remote faces in its own apply.
+- **`ui-activity` returns to its upstream name `ui-jobs`** and renders one roster: `jobsBySession` rows, expandable exactly when `record` is present. The two-roster join is deleted.
 - **Foreground workflow loses its live panel deliberately.** `tool-workflow`'s activity mirror is removed; a foreground run surfaces through its recorded run/member lifecycle events only, and the per-line `workflow/phase` / `workflow/log` narration has no observer until workflow gains `run_in_background` and registers a record job. Jobs stay a pure background registry — no `foreground` mode bit, no model-invisible rows, no run-less rows.
 
 ## Alternatives considered
@@ -30,6 +30,16 @@ The one activity without a job was the foreground workflow mirror. Everything el
 
 The [prior seam note](../feature/2026-08-24-activity-observation-seam.md)'s argument against durable session events and control-stream framing for live output still holds and carries over unchanged: the record is process-local observation state, never a session event, so "model-visible ⟺ logged" is untouched.
 
+## Review corrections
+
+Review of the merged design settled five points that the sections above leave implicit:
+
+- **The pump's wait holds constant resources.** `pumpJobOutput` subscribes to the producer's `done` once and keeps one pending timer; settlement clears the timer and wakes the current wait. Racing the same pending promise every poll round retained a reaction per round until the job ended — a day-long job at the default cadence accumulated over a million closures.
+- **Client disposal waits for carrier quiescence.** The `ctx.jobOutput` effect disposer is async and awaits every open `RemoteStream.dispose()`, because Cordis awaits async disposers and a fiber that reported unloaded while an old iterator was still closing could overlap the next plugin instance under HMR.
+- **The roster flags a record, it does not count bytes.** `SessionJob.record: true` replaced `outputTotal`: `append` commits no roster change (only lifecycle commits fire `onJobsChanged`), so a mirrored count went stale the moment output arrived, and broadcasting the roster per append was rejected as traffic for a number no row needed. The live offsets ride the observation stream's `opened` anchor.
+- **The record is a best-effort live preview.** Producers copy stdout and stderr per poll round, so writes inside one window land stdout first; pipe capture cannot recover the true interleaving (only a PTY or `2>&1` can), and the client concatenates chunks regardless of `channel`. `channel` stays on the wire because it is what makes that limit visible and what a distinct stderr rendering would need.
+- **Web e2e keep-alive is a barrier, not a clock.** The background command holds on a file the test owns in the job's cwd and the scenario kills it explicitly, so no CI stall can settle the job first; each scenario is one test, never a chain of tests sharing a job id.
+
 ## Consequences
 
 Every observable row is a killable job; a pure-observation surface with no lifecycle would need a new home (harness diagnostics belong to the inspector plane, not here). A future remote job provider must implement lifecycle and record together. The record's retention config lives on `jobs-local`; the observation wire's cadence config (`observeFlushMs`, `observeMaxFrameBytes`) lives on `api-job-controller`.

.agents/notes/implemented/architecture/2026-09-01-jobs-absorb-activity-record.zh.md

@@ -18,8 +18,8 @@ Status: implemented
 - **没有 `end`。** job 结算——生产者 outcome、kill 或 teardown——是 record 唯一的关闭：裁剪到结算保留量并发出最后一个 `onOutput` 信号。双结算配对（`ActivityHandle.end` 与 teardown 强结的 first-wins 对账）被删除，而非重新实现。
 - **`readRecord(id, from, caller)`** 是非消耗多读者视图（绝对 UTF-8 偏移，低于保留窗口为 `lossy`），围栏与其他 job 读取一致；永不置 `reported`。模型面的 `readOutput` 游标原样保留——两个投影服务不同读者，刻意保持分离。
 - **`jobs-local`** 吸收块环（`retainBytes` 运行期 256 KiB，`settledRetainBytes` 结算后 16 KiB），`pumpJobOutput` 在 seam 旁替换 `pumpActivityOutput`。生产者把 pump 的末次排空折进 `hooks.done`，让 record 在结算关闭前握有最后的字节。
-- **wire 按角色拆分。** `SessionJob.outputTotal`（恰在 record job 上存在）在会话控制流上标记行可观测；`api-job-controller` 的 `job` 命名空间上的 `job.observe({sessionId?, jobId, from?})` 流式发送 anchor/output/status 帧，围栏读取者由请求的 session 解析。`api-activity-controller` 删除；其名册流是冗余的（jobs 帧就是名册），observe 机制位于 `api-job-controller`（`observe.ts`，客户端 `ctx.jobOutput`），它在自己的 apply 里解析所依赖的 Remote 面。
-- **`ui-activity` 改回上游名字 `ui-jobs`**，渲染单一名册：`jobsBySession` 行，恰在 `outputTotal` 存在时可展开。双名册 join 被删除。
+- **wire 按角色拆分。** `SessionJob.record`（恰在 record job 上存在）在会话控制流上标记行可观测；`api-job-controller` 的 `job` 命名空间上的 `job.observe({sessionId?, jobId, from?})` 流式发送 anchor/output/status 帧，围栏读取者由请求的 session 解析。`api-activity-controller` 删除；其名册流是冗余的（jobs 帧就是名册），observe 机制位于 `api-job-controller`（`observe.ts`，客户端 `ctx.jobOutput`），它在自己的 apply 里解析所依赖的 Remote 面。
+- **`ui-activity` 改回上游名字 `ui-jobs`**，渲染单一名册：`jobsBySession` 行，恰在 `record` 存在时可展开。双名册 join 被删除。
 - **前台 workflow 有意失去实时面板。** `tool-workflow` 的 activity 镜像被移除；前台 run 只通过已记录的 run/member 生命周期事件呈现，逐行的 `workflow/phase` / `workflow/log` 叙述在 workflow 获得 `run_in_background` 并登记 record job 之前没有观察者。jobs 保持纯后台注册表——没有 `foreground` 模式位、没有模型不可见行、没有无 run 行。
 
 ## 曾考虑的替代方案
@@ -30,6 +30,16 @@ Status: implemented
 
 [先前 seam note](../feature/2026-08-24-activity-observation-seam.zh.md) 反对用持久会话事件或 control 流承载实时输出的论证仍然成立并原样沿用：record 是进程本地观测状态，从不是会话事件，「模型可见 ⟺ 已记录」不受影响。
 
+## 评审修正
+
+对已合并设计的评审确定了上文各节未言明的五点：
+
+- **泵的等待资源恒定。** `pumpJobOutput` 对生产者的 `done` 只订阅一次，同一时刻只保留一个待触发的 timer；结算清掉 timer 并唤醒当前等待。此前每个轮询轮次都对同一个 pending promise 赛跑，每轮留下一个 reaction 直到任务结束——默认节奏下跑一天的任务会累积上百万个闭包。
+- **客户端卸载等待载体静止。** `ctx.jobOutput` 的 effect disposer 是异步的，await 每条打开的 `RemoteStream.dispose()`：Cordis 会 await 异步 disposer，而一个在旧 iterator 仍在关闭时就宣告卸载完成的 fiber，在 HMR 下可能与下一个插件实例重叠。
+- **名册标记 record，不计字节。** `SessionJob.record: true` 取代了 `outputTotal`：`append` 不提交名册变更（只有生命周期提交触发 `onJobsChanged`），镜像的计数在输出到达的瞬间就陈旧了；而为一个没有任何行需要的数字按 append 广播整份名册被否决。实时偏移随观测流的 `opened` 锚帧到达。
+- **record 是尽力而为的实时预览。** 生产者按轮询轮次复制 stdout 与 stderr，同一窗口内的写入先 stdout 落地；pipe 捕获无法还原真实交错（只有 PTY 或 `2>&1` 可以），客户端拼接 chunk 时也不区分 `channel`。`channel` 留在 wire 上，因为它正是让这一限制可见、也是将来区分渲染 stderr 所需要的东西。
+- **Web e2e 的保活是 barrier 而非时钟。** 后台命令守在测试拥有的、位于 job cwd 的文件上，由场景显式 kill，因此 CI 卡顿不可能先把任务结算掉；每个场景是一个测试，绝不是共享 job id 的测试链。
+
 ## 后果
 
 每个可观测行都是可 kill 的 job；没有生命周期的纯观测面需要新的归属（harness 诊断属于 inspector 平面，不在这里）。未来的远端 job provider 必须同时实现生命周期与 record。record 的保留配置在 `jobs-local`；观测 wire 的节奏配置（`observeFlushMs`、`observeMaxFrameBytes`）在 `api-job-controller`。

apps/web/tests/background-job-list.e2e.ts

@@ -1,6 +1,6 @@
 // Session-header background jobs driven by a real `ctx.jobs` entry. No model
 // call is involved.
-import { readFile } from 'node:fs/promises'
+import { readFile, writeFile } from 'node:fs/promises'
 import { fileURLToPath } from 'node:url'
 import { join } from 'node:path'
 import type { Browser, Page } from 'playwright'
@@ -22,9 +22,12 @@ const RUNNING_EXPECTED = join(SNAPSHOT_DIR, 'running.expected.md')
 const SETTLED_EXPECTED = join(SNAPSHOT_DIR, 'settled.expected.md')
 const MODE = webSnapshotMode()
 const SEED_ID = 'background-job-list-web-e2e'
-// Long enough that the running assertions never race the process exiting on
-// their own; the test kills it explicitly to reach the settled state.
-const COMMAND = 'sleep 45'
+// A hold on a barrier file the test owns in the job's cwd: the process never
+// exits on its own, so no CI stall can settle it before the scenario kills it.
+// The bounded loop only caps an orphan's lifetime if the runner dies before
+// `afterAll` releases the barrier.
+const RELEASE = '.background-job-list.release'
+const COMMAND = `for _ in $(seq 1 3000); do [ -e ${RELEASE} ] && break; sleep 0.2; done`
 
 /**
  * Wait for opening a session to publish its live Agent.
@@ -48,7 +51,6 @@ describe.skipIf(MODE === 'record')('web e2e: background job list', () => {
   let page: Page
   let tripwire: ReturnType<typeof watchConsole>
   let agent: Agent
-  let jobId: JobId
 
   beforeAll(async () => {
     scaffold = await launchWebScaffold({})
@@ -73,12 +75,14 @@ describe.skipIf(MODE === 'record')('web e2e: background job list', () => {
   }, 120_000)
 
   afterAll(async () => {
+    if (scaffold !== undefined) await writeFile(join(scaffold.workspaceCwd, RELEASE), '')
     await browser?.close()
     await scaffold?.close()
   })
 
-  it('shows a running background job in the session header without a refresh', async () => {
-    onTestFailed(() => saveFailureShot(page, 'web-e2e-background-job-running'))
+  it('shows a running background job in the header without a refresh, then flips it to cancelled on settlement', async () => {
+    let phase = 'running'
+    onTestFailed(() => saveFailureShot(page, `web-e2e-background-job-${phase}`))
     // Polling for zero would pass at t=0 before delivery and prove nothing.
     const trigger = page.getByRole('button', { name: '1 background job running' })
     expect(await trigger.count()).toBe(0)
@@ -93,7 +97,7 @@ describe.skipIf(MODE === 'record')('web e2e: background job list', () => {
     const reported = started.content.map(block => block.type === 'text' ? block.text : '').join('')
     const matched = /\bbash-\d+\b/.exec(reported)
     if (matched === null) throw new Error(`background bash reported no job id: ${reported}`)
-    jobId = JobId(matched[0])
+    const jobId = JobId(matched[0])
 
     await trigger.waitFor({ timeout: 15_000 })
     await trigger.click()
@@ -101,24 +105,22 @@ describe.skipIf(MODE === 'record')('web e2e: background job list', () => {
     await row.waitFor({ timeout: 10_000 })
     await expect.poll(() => row.textContent()).toContain(COMMAND)
 
-    const snapshot = await captureStableAria(page, '[class*="menu"]', scaffold.workspaceCwd)
-    await compareOrRefreshGolden(RUNNING_EXPECTED, snapshot, MODE)
+    const running = await captureStableAria(page, '[class*="menu"]', scaffold.workspaceCwd)
+    await compareOrRefreshGolden(RUNNING_EXPECTED, running, MODE)
     expect(tripwire.pageErrors).toEqual([])
     expect(tripwire.warnings).toEqual([])
-  }, 60_000)
 
-  it('flips the open list to the cancelled outcome when the registry settles it', async () => {
-    onTestFailed(() => saveFailureShot(page, 'web-e2e-background-job-settled'))
+    phase = 'settled'
     expect(scaffold.ctx.jobs.kill(jobId, agent, 'web e2e cancellation')).toBe('requested')
 
     const idle = page.getByRole('button', { name: '1 background job', exact: true })
     await idle.waitFor({ timeout: 20_000 })
 
-    const snapshot = await captureStableAria(page, '[class*="menu"]', scaffold.workspaceCwd)
-    await compareOrRefreshGolden(SETTLED_EXPECTED, snapshot, MODE)
+    const settled = await captureStableAria(page, '[class*="menu"]', scaffold.workspaceCwd)
+    await compareOrRefreshGolden(SETTLED_EXPECTED, settled, MODE)
     expect(tripwire.pageErrors).toEqual([])
     expect(tripwire.warnings).toEqual([])
-  }, 60_000)
+  }, 90_000)
 
   it('keeps its snapshot inventory closed', async () => {
     await assertFixtureInventory(SNAPSHOT_DIR, ['running.expected.md', 'settled.expected.md'])

apps/web/tests/live-job-stream.e2e.ts

@@ -2,7 +2,7 @@
 // arrives over the session control stream, expanding it opens the record
 // observation stream, and the panel shows the process's real output while it
 // is still running. No model call is involved.
-import { readFile } from 'node:fs/promises'
+import { readFile, writeFile } from 'node:fs/promises'
 import { fileURLToPath } from 'node:url'
 import { join } from 'node:path'
 import type { Browser, Page } from 'playwright'
@@ -24,9 +24,12 @@ const STREAMING_EXPECTED = join(SNAPSHOT_DIR, 'streaming.expected.md')
 const SETTLED_EXPECTED = join(SNAPSHOT_DIR, 'settled.expected.md')
 const MODE = webSnapshotMode()
 const SEED_ID = 'live-job-stream-web-e2e'
-// One fixed line of early output, then silence long enough that the streaming
-// assertions never race the process exiting; the test kills it explicitly.
-const COMMAND = "printf 'streamed-%s\\n' marker-line; sleep 45"
+// One fixed line of early output, then a hold on a barrier file the test owns
+// in the job's cwd: the process never exits on its own, so no CI stall can
+// settle it before the scenario kills it. The bounded loop only caps an
+// orphan's lifetime if the runner dies before `afterAll` releases the barrier.
+const RELEASE = '.live-job-stream.release'
+const COMMAND = `printf 'streamed-%s\\n' marker-line; for _ in $(seq 1 3000); do [ -e ${RELEASE} ] && break; sleep 0.2; done`
 
 /**
  * Wait for opening a session to publish its live Agent.
@@ -50,7 +53,6 @@ describe.skipIf(MODE === 'record')('web e2e: live job stream', () => {
   let page: Page
   let tripwire: ReturnType<typeof watchConsole>
   let agent: Agent
-  let jobId: JobId
 
   beforeAll(async () => {
     scaffold = await launchWebScaffold({})
@@ -74,12 +76,14 @@ describe.skipIf(MODE === 'record')('web e2e: live job stream', () => {
   }, 120_000)
 
   afterAll(async () => {
+    if (scaffold !== undefined) await writeFile(join(scaffold.workspaceCwd, RELEASE), '')
     await browser?.close()
     await scaffold?.close()
   })
 
-  it('streams a running background command\'s real output into the expanded panel', async () => {
-    onTestFailed(() => saveFailureShot(page, 'web-e2e-live-job-streaming'))
+  it('streams a running job\'s real output into the expanded panel, then settles it in place on kill', async () => {
+    let phase = 'streaming'
+    onTestFailed(() => saveFailureShot(page, `web-e2e-live-job-${phase}`))
     const trigger = page.getByRole('button', { name: '1 background job running' })
     expect(await trigger.count()).toBe(0)
 
@@ -93,7 +97,7 @@ describe.skipIf(MODE === 'record')('web e2e: live job stream', () => {
     const reported = started.content.map(block => block.type === 'text' ? block.text : '').join('')
     const matched = /\bbash-\d+\b/.exec(reported)
     if (matched === null) throw new Error(`background bash reported no job id: ${reported}`)
-    jobId = JobId(matched[0])
+    const jobId = JobId(matched[0])
 
     // The job row reaches the header over the session control stream.
     await trigger.waitFor({ timeout: 15_000 })
@@ -106,14 +110,12 @@ describe.skipIf(MODE === 'record')('web e2e: live job stream', () => {
     await expand.click()
     await page.getByText('streamed-marker-line').waitFor({ timeout: 15_000 })
 
-    const snapshot = await captureStableAria(page, '[class*="menu"]', scaffold.workspaceCwd)
-    await compareOrRefreshGolden(STREAMING_EXPECTED, snapshot, MODE)
+    const streaming = await captureStableAria(page, '[class*="menu"]', scaffold.workspaceCwd)
+    await compareOrRefreshGolden(STREAMING_EXPECTED, streaming, MODE)
     expect(tripwire.pageErrors).toEqual([])
     expect(tripwire.warnings).toEqual([])
-  }, 60_000)
 
-  it('settles the open panel when the job is killed, keeping the streamed tail', async () => {
-    onTestFailed(() => saveFailureShot(page, 'web-e2e-live-job-settled'))
+    phase = 'settled'
     expect(scaffold.ctx.jobs.kill(jobId, agent, 'web e2e cancellation')).toBe('requested')
 
     // Settlement arrives on the observation stream itself, so the open panel
@@ -122,11 +124,11 @@ describe.skipIf(MODE === 'record')('web e2e: live job stream', () => {
     await idle.waitFor({ timeout: 20_000 })
     await page.getByText('streamed-marker-line').waitFor({ timeout: 10_000 })
 
-    const snapshot = await captureStableAria(page, '[class*="menu"]', scaffold.workspaceCwd)
-    await compareOrRefreshGolden(SETTLED_EXPECTED, snapshot, MODE)
+    const settled = await captureStableAria(page, '[class*="menu"]', scaffold.workspaceCwd)
+    await compareOrRefreshGolden(SETTLED_EXPECTED, settled, MODE)
     expect(tripwire.pageErrors).toEqual([])
     expect(tripwire.warnings).toEqual([])
-  }, 60_000)
+  }, 90_000)
 
   it('keeps its snapshot inventory closed', async () => {
     await assertFixtureInventory(SNAPSHOT_DIR, ['settled.expected.md', 'streaming.expected.md'])

apps/web/tests/snapshots/live-job-stream/settled.expected.md

@@ -1,8 +1,8 @@
 - list "Background jobs":
   - listitem:
-    - button "Hide live output of printf 'streamed-%s\\n' marker-line; sleep 45" [expanded]:
-      - text: "bash printf 'streamed-%s\\n' marker-line; sleep 45 signal: SIGTERM {{duration}}"
+    - button "Hide live output of printf 'streamed-%s\\n' marker-line; for _ in $(seq 1 3000); do [ -e .live-job-stream.release ] && break; sleep 0.2; done" [expanded]:
+      - text: "bash printf 'streamed-%s\\n' marker-line; for _ in $(seq 1 3000); do [ -e .live-job-stream.release ] && break; sleep 0.2; done signal: SIGTERM {{duration}}"
       - img
-    - text: $ printf 'streamed-%s\n' marker-line; sleep 45
+    - text: $ printf 'streamed-%s\n' marker-line; for _ in $(seq 1 3000); do [ -e .live-job-stream.release ] && break; sleep 0.2; done
     - button "Copy"
     - text: streamed-marker-line

apps/web/tests/snapshots/live-job-stream/streaming.expected.md

@@ -1,8 +1,8 @@
 - list "Background jobs":
   - listitem:
-    - button "Hide live output of printf 'streamed-%s\\n' marker-line; sleep 45" [expanded]:
-      - text: printf 'streamed-%s\n' marker-line; sleep 45 {{duration}} bash running
+    - button "Hide live output of printf 'streamed-%s\\n' marker-line; for _ in $(seq 1 3000); do [ -e .live-job-stream.release ] && break; sleep 0.2; done" [expanded]:
+      - text: printf 'streamed-%s\n' marker-line; for _ in $(seq 1 3000); do [ -e .live-job-stream.release ] && break; sleep 0.2; done {{duration}} bash running
       - img
-    - text: $ printf 'streamed-%s\n' marker-line; sleep 45
+    - text: $ printf 'streamed-%s\n' marker-line; for _ in $(seq 1 3000); do [ -e .live-job-stream.release ] && break; sleep 0.2; done
     - button "Copy"
     - text: streamed-marker-line

docs/event-producer-consumer.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/event-producer-consumer.md
-event-producer-consumer.md: 930ffba24789ee51db110310cb17a2231135cf1d
-event-producer-consumer.zh.md: 6583b1d9f333ad934db2123a2bc5c40b12f1ae36
+event-producer-consumer.md: e0b4709875723bd96f9910b1bf518600612f0a12
+event-producer-consumer.zh.md: 0f0fa1317eb5a017daf435648f5c3e36208626e0

docs/event-producer-consumer.md

@@ -21,11 +21,11 @@ This matrix shows which packages dispatch each harness-owned event and which pac
 | `agent/session-start` | `emit` | [`packages/core/agent/src/runtime-types.ts:224`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | `agent-team`, [`goal`](../packages/goal/goal), [`goal-round-driver`](../packages/goal/goal-round-driver), [`hooks-claude-code`](../packages/hooks/hooks-claude-code), [`hooks-codex`](../packages/hooks/hooks-codex) |
 | `agent/status` | `emit` | [`packages/core/agent/src/runtime-types.ts:185`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`agent`](../packages/core/agent), `agent-team`, [`compaction-basic`](../packages/compaction/compaction-basic), [`goal-round-driver`](../packages/goal/goal-round-driver), [`schedule`](../packages/schedule/schedule), `server`, `session-controller` |
 | `agent/turn-stopping` | `serial` | [`packages/core/agent/src/runtime-types.ts:285`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`hooks-claude-code`](../packages/hooks/hooks-claude-code), [`hooks-codex`](../packages/hooks/hooks-codex) |
-| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:550`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
-| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:530`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
-| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:557`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
-| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:536`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
-| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:543`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
+| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:551`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
+| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:531`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
+| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:558`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
+| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:537`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
+| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:544`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
 | `approval/request` | `waterfall` | [`packages/interaction/user-approval/src/types.ts:85`](../packages/interaction/user-approval/src/types.ts) | [`user-approval`](../packages/interaction/user-approval) (`waterfall`) | [`acp`](../packages/acp/acp), `remotes` |
 | `authorization/settled` | `emit` | [`packages/credentials/authorization/src/index.ts:57`](../packages/credentials/authorization/src/index.ts) | [`authorization`](../packages/credentials/authorization) (`events.dispatch`) | [`authorization`](../packages/credentials/authorization) |
 | `commands/change` | `emit` | [`packages/interaction/commands/src/types.ts:81`](../packages/interaction/commands/src/types.ts) | [`commands`](../packages/interaction/commands) (`events.dispatch`) | `remotes` |

docs/event-producer-consumer.zh.md

@@ -23,11 +23,11 @@
 | `agent/session-start` | `emit` | [`packages/core/agent/src/runtime-types.ts:224`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | `agent-team`, [`goal`](../packages/goal/goal), [`goal-round-driver`](../packages/goal/goal-round-driver), [`hooks-claude-code`](../packages/hooks/hooks-claude-code), [`hooks-codex`](../packages/hooks/hooks-codex) |
 | `agent/status` | `emit` | [`packages/core/agent/src/runtime-types.ts:185`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`agent`](../packages/core/agent), `agent-team`, [`compaction-basic`](../packages/compaction/compaction-basic), [`goal-round-driver`](../packages/goal/goal-round-driver), [`schedule`](../packages/schedule/schedule), `server`, `session-controller` |
 | `agent/turn-stopping` | `serial` | [`packages/core/agent/src/runtime-types.ts:285`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`hooks-claude-code`](../packages/hooks/hooks-claude-code), [`hooks-codex`](../packages/hooks/hooks-codex) |
-| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:550`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
-| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:530`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
-| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:557`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
-| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:536`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
-| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:543`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
+| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:551`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
+| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:531`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
+| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:558`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
+| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:537`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
+| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:544`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
 | `approval/request` | `waterfall` | [`packages/interaction/user-approval/src/types.ts:85`](../packages/interaction/user-approval/src/types.ts) | [`user-approval`](../packages/interaction/user-approval) (`waterfall`) | [`acp`](../packages/acp/acp), `remotes` |
 | `authorization/settled` | `emit` | [`packages/credentials/authorization/src/index.ts:57`](../packages/credentials/authorization/src/index.ts) | [`authorization`](../packages/credentials/authorization) (`events.dispatch`) | [`authorization`](../packages/credentials/authorization) |
 | `commands/change` | `emit` | [`packages/interaction/commands/src/types.ts:81`](../packages/interaction/commands/src/types.ts) | [`commands`](../packages/interaction/commands) (`events.dispatch`) | `remotes` |

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: 1af9cac6574b36fb9ee0fe2bd4ff4a529848c2fe
-README.zh.md: a232d8e8655fea6ccddedb60e222837466705549
+README.md: 6aaab1bff1f13cc21672a039170f811b570230db
+README.zh.md: 47e68802346fbc5e41e74171a42afa17289ee156

packages/api/job-controller/README.md

@@ -8,7 +8,7 @@ English | [中文](README.zh.md)
 
 ## Summary
 
-`@deepseek-ai/dsh-api-job-controller` owns the Host `ctx.jobController` service and the generated Client `ctx.remote.job` namespace. Its one Remote stream, `job.observe`, delivers a record-declaring job's retained output from an absolute byte offset; the Client half installs `ctx.jobOutput`, the reference-counted per-job observation service whose accumulated views the session-header job list renders. The roster itself is not here: `SessionJob` rows still ride the session control stream owned by [`dsh-api-session-controller`](../session-controller/README.md), and a row's `outputTotal` says whether this controller has anything to stream for it.
+`@deepseek-ai/dsh-api-job-controller` owns the Host `ctx.jobController` service and the generated Client `ctx.remote.job` namespace. Its one Remote stream, `job.observe`, delivers a record-declaring job's retained output from an absolute byte offset; the Client half installs `ctx.jobOutput`, the reference-counted per-job observation service whose accumulated views the session-header job list renders. The roster itself is not here: `SessionJob` rows still ride the session control stream owned by [`dsh-api-session-controller`](../session-controller/README.md), and a row's `record` says whether this controller has anything to stream for it.
 
 ## Table of Contents
 
@@ -50,6 +50,7 @@ No direct effect; observation reads never touch model requests.
 
 <a id="known-limitations-and-deferred-work"></a>
 
+- The record is a best-effort live preview, not a terminal transcript: producers copy stdout and stderr per poll round, so the two streams' writes inside one poll window reach the record stdout first, and the client concatenates chunks regardless of `channel`.
 - The stream is process-local: a Host restart loses every record, and a resumed observation then anchors on an empty registry.
 - Per-session fencing is enforced by the registry read; the Remote layer itself serves any connected browser, matching the session control stream's `jobsBySession` broadcast.
 

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

@@ -8,7 +8,7 @@ kind: "package-reference"
 
 ## 概述
 
-`@deepseek-ai/dsh-api-job-controller` 拥有 Host 的 `ctx.jobController` 服务与生成的 Client `ctx.remote.job` namespace。它唯一的 Remote 流 `job.observe` 从绝对字节偏移发送声明了 record 的 job 的保留输出；Client 半侧安装 `ctx.jobOutput`——按 job 引用计数的观测服务，会话头部任务列表渲染其累积视图。名册不在这里：`SessionJob` 行仍随 [`dsh-api-session-controller`](../session-controller/README.zh.md) 拥有的会话控制流到达，行上的 `outputTotal` 说明本控制器是否有东西可流。
+`@deepseek-ai/dsh-api-job-controller` 拥有 Host 的 `ctx.jobController` 服务与生成的 Client `ctx.remote.job` namespace。它唯一的 Remote 流 `job.observe` 从绝对字节偏移发送声明了 record 的 job 的保留输出；Client 半侧安装 `ctx.jobOutput`——按 job 引用计数的观测服务，会话头部任务列表渲染其累积视图。名册不在这里：`SessionJob` 行仍随 [`dsh-api-session-controller`](../session-controller/README.zh.md) 拥有的会话控制流到达，行上的 `record` 说明本控制器是否有东西可流。
 
 ## 目录
 
@@ -50,6 +50,7 @@ Client 入口在 `ClientJobOutputModel` 之上提供 `ClientJobOutput`（`ctx.jo
 
 <a id="known-limitations-and-deferred-work"></a>
 
+- record 是尽力而为的实时预览，不是终端转录：生产者按轮询轮次复制 stdout 与 stderr，同一轮询窗口内两条流的写入先 stdout 后到达 record，客户端拼接 chunk 时也不区分 `channel`。
 - 流是进程本地的：Host 重启丢失全部 record，续读的观测随后锚定在空注册表上。
 - 按 session 的围栏由注册表读取强制；Remote 层本身服务任何已连接浏览器，与会话控制流对 `jobsBySession` 的广播一致。
 

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

@@ -74,13 +74,15 @@ export class ClientJobOutput extends Service implements IJobOutput {
   ) {
     super(ctx, 'jobOutput')
     this.state = model
-    ctx.effect(() => () => {
+    ctx.effect(() => async () => {
       const open = [...this.entries.values()]
       this.entries.clear()
-      for (const entry of open) {
-        entry.stopped = true
-        void entry.dispose()
-      }
+      for (const entry of open) entry.stopped = true
+      // Cordis awaits an async disposer, so the fiber stays unloading until
+      // every carrier iterator has closed and a successor plugin instance
+      // cannot overlap one. A carrier whose teardown fails is stopped all the
+      // same; its failure has no consumer here.
+      await Promise.allSettled(open.map(entry => entry.dispose()))
     }, 'job-controller.client.observations')
   }
 

packages/api/job-controller/tests/model.client.spec.ts

@@ -391,6 +391,30 @@ describe('ClientJobOutput observation streams', () => {
     expect(model.getSnapshot().observed[String(ID)]?.error).toBeUndefined()
   })
 
+  it('holds service disposal until every open stream reports quiescence', async () => {
+    const { ctx, output, streams } = bench()
+    output.observe(undefined, ID)
+    output.observe(undefined, 'bash-2' as JobId)
+    for (const { stream } of streams) stream.deferDispose = true
+    await tick()
+
+    let settled = false
+    const disposal = ctx.fiber.dispose().then(() => { settled = true })
+    await tick()
+    // Both carriers were told to stop, but neither iterator has closed yet:
+    // the fiber must still be unloading, or a successor plugin instance could
+    // overlap the old generation's iterator.
+    expect(streams.every(entry => entry.stream.disposed)).toBe(true)
+    expect(settled).toBe(false)
+
+    streams[0]!.stream.releaseDispose()
+    await tick()
+    expect(settled).toBe(false)
+    streams[1]!.stream.releaseDispose()
+    await disposal
+    expect(settled).toBe(true)
+  })
+
   it('records a terminal stream failure and service disposal closes live streams', async () => {
     const { ctx, model, output, streams } = bench()
     output.observe(undefined, ID)

packages/api/session-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/session-controller/README.md
-README.md: 4f94607e70a5b963fa50d7a497afbcc97d273cb9
-README.zh.md: 455769f77ce8ad304949624de938e56579024145
+README.md: 4414cd7221b40cc880122415e83b6dc3818df481
+README.zh.md: 8f457d4a94f57c802f67624d9c84ba15c78b1ada

packages/api/session-controller/README.md

@@ -29,7 +29,7 @@ Each endpoint states its activation policy. List, search, attachment, history pa
 
 The Client adapter exposes `SessionEventStream`, a Gateway `RemoteJournalStream` bound to one ordinary or direct-subagent address. It opens follow before the initial page, publishes only contiguous `replace`, `prepend`, and `append` changes, and repairs reconnect or sequence gaps through a tail page. Backwards paging has two verbs: `loadOlder()` pulls one 50-message page, and `loadThrough(seq)` — the turn-jump loader — loops 200-message pages until the window covers the target seq, lowering a shared target on repeated calls, stopping on a page that makes no progress, and reporting busy through the same `loadingOlder` snapshot bit. Ordinary records cover `[event.seq, event.seq]`; packed rows cover `[event.seq, event.seq + memberCount - 1]`. A business, persistence, or unresolved continuity failure terminates the stream, while only physical carrier loss selects automatic resumption. `SessionControlStream` is a Gateway `RemoteSnapshotStream`; every generation opens with a complete process-local baseline, so reconnect replaces queue, jobs, and projection state instead of treating transient values as durable events.
 
-A `SessionJob` row carries `outputTotal` exactly when the job declared an observation record; streaming that record is the job controller's `job.observe` (see [`dsh-api-job-controller`](../job-controller/README.md)), while this package keeps the roster on the control stream.
+A `SessionJob` row carries `record: true` exactly when the job declared an observation record; streaming that record is the job controller's `job.observe` (see [`dsh-api-job-controller`](../job-controller/README.md)), while this package keeps the roster on the control stream.
 
 The Session object also carries local submission echoes: `session.beginSubmission` inserts one into `SessionSnapshot.pendingSubmissions` synchronously, before the caller serializes and prompts, so a conversation UI can show the message on the submit click's own frame. Session derives each echo's `transcript`, `queued`, or `steering` placement from its current running state and the requested delivery mode, then retains that placement while serialization is in flight. The prompt's `requestId` is the correlation identity: the Host echoes it as the durable user source's `rpcId`, and queue occurrences project it as `SessionQueuedItem.rpcId`. An echo retires one animation frame after its durable event or queue occurrence is observed (the delay keeps it renderable until the replacement is ready), immediately when its identified prompt fails or is abandoned, and as failed on disposal; each retirement fires the registered `onRetire` callback exactly once. Echoes are Client memory only; reload and reconnect rebuild the conversation from durable events alone.
 

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

@@ -29,7 +29,7 @@ kind: "package-reference"
 
 Client adapter 提供 `SessionEventStream`，即绑定到一个普通 Session 或 direct subagent address 的 Gateway `RemoteJournalStream`。它在读取首个 page 前打开 follow，只发布连续的 `replace`、`prepend` 和 `append` 变更，并通过 tail page 修复重连或 seq 缺口。向后分页有两个动词：`loadOlder()` 拉一页 50 条 message，而 `loadThrough(seq)`——轮次跳转加载器——按 200 条 message 一页循环拉取直到窗口覆盖目标 seq，重复调用会下调共享目标，遇到无进展的页即停止，忙碌状态复用同一个 `loadingOlder` 快照位。普通 record 覆盖 `[event.seq, event.seq]`，packed row 覆盖 `[event.seq, event.seq + memberCount - 1]`。业务、persistence 或无法恢复的连续性错误会终止 stream，只有物理载体断开才触发自动恢复。`SessionControlStream` 是 Gateway `RemoteSnapshotStream`；每代都以完整的进程本地 baseline 开始，因此重连会替换 queue、jobs 和 projection 状态，而不会把瞬态值当作 durable event。
 
-`SessionJob` 行恰在该 job 声明了观测 record 时携带 `outputTotal`；流式发送该 record 的是 job 控制器的 `job.observe`（见 [`dsh-api-job-controller`](../job-controller/README.zh.md)），本包只在控制流上保留名册。
+`SessionJob` 行恰在该 job 声明了观测 record 时携带 `record: true`；流式发送该 record 的是 job 控制器的 `job.observe`（见 [`dsh-api-job-controller`](../job-controller/README.zh.md)），本包只在控制流上保留名册。
 
 Session 对象还承载本地提交回显：`session.beginSubmission` 在调用方序列化与 prompt 之前，同步把一条回显写入 `SessionSnapshot.pendingSubmissions`，会话 UI 因此能在点击提交的当帧显示消息。Session 根据当前运行状态与请求的投递模式推导每条回显的 `transcript`、`queued` 或 `steering` 位置，并在序列化期间保留该位置。prompt 的 `requestId` 是关联标识：Host 把它回显为 durable user source 的 `rpcId`，queue occurrence 也把它投影为 `SessionQueuedItem.rpcId`。回显在观察到其 durable event 或 queue occurrence 后延迟一个动画帧退休，该延迟保证替代内容就绪前回显仍可渲染；带标识的 prompt 失败或被放弃时立即退休，销毁时按 failed 退休；每次退休恰好触发一次注册的 `onRetire` 回调。回显只存在于 Client 内存；刷新与重连只从 durable event 重建会话。
 

packages/api/session-controller/src/control.ts

@@ -212,6 +212,6 @@ function jobView(job: JobSnapshot): SessionJob {
     ...(job.detail === undefined ? {} : { detail: job.detail }),
     startedAt: job.startedAt,
     ...(job.finishedAt === undefined ? {} : { finishedAt: job.finishedAt }),
-    ...(job.outputTotal === undefined ? {} : { outputTotal: job.outputTotal }),
+    ...(job.outputTotal === undefined ? {} : { record: true as const }),
   }
 }

packages/api/session-controller/src/types.ts

@@ -489,11 +489,12 @@ export interface SessionJob {
   readonly startedAt: number
   readonly finishedAt?: number
   /**
-   * Total UTF-8 bytes the job's observation record holds so far. Present
-   * exactly when the job declared a record — the row is observable through
-   * the job controller's `job.observe` stream.
+   * Present exactly when the job declared an observation record: the row is
+   * observable through the job controller's `job.observe` stream, whose
+   * `opened` anchor carries the live byte offsets. The roster refreshes on
+   * lifecycle commits only, so it mirrors no byte count.
    */
-  readonly outputTotal?: number
+  readonly record?: true
 }
 
 /** Complete live control baseline emitted once per control stream generation. */

packages/api/session-controller/tests/control-jobs.host.spec.ts

@@ -162,7 +162,7 @@ describe('Session control jobs updates', () => {
     ])
   })
 
-  it('carries outputTotal exactly for record-declaring jobs', async () => {
+  it('flags record-declaring jobs instead of mirroring a byte count the roster never refreshes', async () => {
     const { ctx, agent, control } = await harness(true)
     const abort = new AbortController()
     const collected = collectJobs(control.control(abort.signal), 1, abort)
@@ -178,7 +178,8 @@ describe('Session control jobs updates', () => {
     })
 
     const [frame] = await collected
-    expect(frame?.jobs[0]?.outputTotal).toBe(2)
+    expect(frame?.jobs[0]).toMatchObject({ record: true })
+    expect(frame?.jobs[0]).not.toHaveProperty('outputTotal')
   })
 
   it('fans an unowned change out to every attached session', async () => {

packages/client/ui-jobs/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/client/ui-jobs/README.md
-README.md: a0ecc137ab1eb57aeaea9ccdfdbaf898794de2fe
-README.zh.md: 744450c480b4d26c0558d992904817174ca2ed73
+README.md: fe357b5555cdc2f7a94ea0a44e97c499652695d3
+README.zh.md: d23b96becc5b2598197f7720a355a37e5c8bf7cb

packages/client/ui-jobs/README.md

@@ -29,7 +29,7 @@ Load the plugin through the web-app manifest; it renders nothing until the sessi
 
 ### One row per job
 
-The session control stream's `jobsBySession` mirror is the single roster: each `SessionJob` row carries lifecycle, duration, and the model-visible `detail`, and its `outputTotal` field is present exactly when the job declared a record — that presence is what makes a row expandable. There is no second roster to join.
+The session control stream's `jobsBySession` mirror is the single roster: each `SessionJob` row carries lifecycle, duration, and the model-visible `detail`, and its `record` flag is present exactly when the job declared a record — that flag is what makes a row expandable. There is no second roster to join.
 
 ### The expanded panel
 

packages/client/ui-jobs/README.zh.md

@@ -29,7 +29,7 @@ kind: "package-reference"
 
 ### 一 job 一行
 
-session control 流的 `jobsBySession` 镜像是唯一名册：每个 `SessionJob` 行携带生命周期、时长与模型可见的 `detail`，其 `outputTotal` 字段恰在 job 声明了 record 时存在——这个存在性就是行可展开的依据。不存在需要 join 的第二份名册。
+session control 流的 `jobsBySession` 镜像是唯一名册：每个 `SessionJob` 行携带生命周期、时长与模型可见的 `detail`，其 `record` 标志恰在 job 声明了 record 时存在——这个标志就是行可展开的依据。不存在需要 join 的第二份名册。
 
 ### 展开的面板
 

packages/client/ui-jobs/src/client/JobListAction.tsx

@@ -46,7 +46,7 @@ function isLive(job: SessionJob): boolean {
 
 /** True exactly when the job declared an observation record. */
 function isObservable(job: SessionJob): boolean {
-  return job.outputTotal !== undefined
+  return job.record === true
 }
 
 /** Closed-union exhaustiveness fence for the wire status set. */

packages/client/ui-jobs/tests/job-list-action.client.spec.tsx

@@ -30,7 +30,7 @@ function job(over: Partial<SessionJob> = {}): SessionJob {
 
 /** A job that declared an observation record, so its row offers a panel. */
 function recordJob(over: Partial<SessionJob> = {}): SessionJob {
-  return job({ outputTotal: 0, ...over })
+  return job({ record: true, ...over })
 }
 
 function props(
@@ -180,7 +180,7 @@ describe('JobListAction observation', () => {
       streaming: false,
     }
     render(<JobListAction {...props(
-      [recordJob({ status: 'killed', finishedAt: 1_700_000_001_000, outputTotal: 240 })],
+      [recordJob({ status: 'killed', finishedAt: 1_700_000_001_000 })],
       undefined,
       { 'bash-1': view },
     )} />)

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

@@ -4987,7 +4987,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
   },
   {
     name: 'SessionJob',
-    declaration: 'export interface SessionJob {\n    readonly id: JobId;\n    readonly kind: string;\n    readonly label: string;\n    readonly status: \'running\' | \'stopping\' | \'completed\' | \'killed\' | \'failed\';\n    readonly detail?: string;\n    readonly startedAt: number;\n    readonly finishedAt?: number;\n    readonly outputTotal?: number;\n}',
+    declaration: 'export interface SessionJob {\n    readonly id: JobId;\n    readonly kind: string;\n    readonly label: string;\n    readonly status: \'running\' | \'stopping\' | \'completed\' | \'killed\' | \'failed\';\n    readonly detail?: string;\n    readonly startedAt: number;\n    readonly finishedAt?: number;\n    readonly record?: true;\n}',
   },
   {
     name: 'SessionLineageNode',

packages/jobs/jobs/src/pump.ts

@@ -60,6 +60,14 @@ export interface JobPumpOptions {
  * `gapBefore`, so the discontinuity stays visible to observers. Fold the
  * returned promise into the producer's `JobHooks.done` chain so the final
  * drain lands before settlement trims and closes the record.
+ *
+ * Each round drains the sources in array order, so bytes two sources
+ * produced inside one poll window land in that order, not in the order they
+ * were written; the record is a best-effort live view whose cross-source
+ * reordering is bounded by `pollMs`, not a terminal transcript.
+ *
+ * The wait holds constant resources however long the job runs: one
+ * subscription on `done` for the whole run and one pending timer at a time.
  * @param job - the running job's producer face receiving the copied chunks.
  * @param sources - producer streams, each pumped at its own offset.
  * @param options - poll cadence and the producer's settlement promise.
@@ -91,18 +99,24 @@ export async function pumpJobOutput(
     }
   }
   let settled = false
-  const settlement = options.done.then(() => { settled = true }, () => { settled = true })
+  let timer: ReturnType<typeof setTimeout> | undefined
+  let wake: (() => void) | undefined
+  const finish = (): void => {
+    settled = true
+    // Clear the pending poll so a finished pump holds no timer for up to one
+    // interval, and release the current wait so the final drain runs at once.
+    if (timer !== undefined) clearTimeout(timer)
+    wake?.()
+  }
+  void options.done.then(finish, finish)
   while (!settled) {
     drain()
-    await Promise.race([
-      new Promise<void>((resolve) => {
-        const timer = setTimeout(resolve, options.pollMs)
-        // The settlement branch wins the race; clear the loser so a finished
-        // pump holds no pending timer for up to one poll interval.
-        void settlement.then(() => { clearTimeout(timer); resolve() })
-      }),
-      settlement,
-    ])
+    await new Promise<void>((resolve) => {
+      wake = resolve
+      timer = setTimeout(resolve, options.pollMs)
+    })
+    wake = undefined
+    timer = undefined
   }
   drain()
 }

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

@@ -173,6 +173,42 @@ describe('pumpJobOutput', () => {
     }
   })
 
+  it('subscribes to done once and holds one poll timer however long the job runs', async () => {
+    vi.useFakeTimers()
+    try {
+      const { job } = recordingJob()
+      // A thenable stands in for the producer's done promise, and whatever the
+      // pump derives from it counts its own reactions too: a pending promise
+      // retains every reaction until it settles, so a day-long job must not
+      // gain one per poll round.
+      let subscriptions = 0
+      let settle!: () => void
+      const settled = new Promise<void>((resolve) => { settle = resolve })
+      const counting = (promise: Promise<unknown>): Promise<unknown> => ({
+        then(onFulfilled?: (value: unknown) => unknown, onRejected?: (reason: unknown) => unknown) {
+          subscriptions += 1
+          return counting(promise.then(onFulfilled, onRejected))
+        },
+      }) as unknown as Promise<unknown>
+      const done = counting(settled)
+      const { source, offsets } = scriptedSource([])
+      const pump = pumpJobOutput(job, [source], { pollMs: 50, done })
+
+      const rounds = 10_000
+      await vi.advanceTimersByTimeAsync(50 * rounds)
+      expect(offsets).toHaveLength(rounds + 1)
+      expect(subscriptions).toBeLessThanOrEqual(2)
+      expect(vi.getTimerCount()).toBe(1)
+
+      settle()
+      await pump
+      expect(offsets).toHaveLength(rounds + 2)
+      expect(vi.getTimerCount()).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('marks a lossy source read as a gap so observers see the discontinuity', async () => {
     const { job, appends } = recordingJob()
     const { source } = scriptedSource([{ text: 'tail', lossy: true }])

packages/shell/tool-bash/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/shell/tool-bash/README.md
-README.md: 24e13266a7ee24b39f524f165401076860f494b1
-README.zh.md: 3d2c9f22033d711ce33382054a372e2c26aa2d46
+README.md: 50047a5fe0d69da1feb38f75ada27595b88fa5c5
+README.zh.md: c74a45561298defb4baa5b34be7cebc1cdfbc744

packages/shell/tool-bash/README.md

@@ -55,7 +55,7 @@ The tool executes `bash -c <command>` and returns the combined output. Commands
 
 ### Running long commands in the background
 
-Passing `run_in_background: true` returns a job id immediately and no timeout applies; the command keeps running while the agent works on something else. The agent reads its output with `job_output` (non-blocking unless `wait: true`), lists jobs with `job_list`, and stops it with `job_kill`; a finished job notifies the owning agent in-session. Background support needs the generic job runtime (`dsh-jobs-local`) and its control tools (`dsh-tool-jobs`) mounted. The background job declares an observation record: the run's non-consuming `observed` readers are pumped into it at `recordPollMs` (default 150), so the Web client streams live output without touching the job's consuming cursor; a pump failure is logged and swallowed while the job runs on.
+Passing `run_in_background: true` returns a job id immediately and no timeout applies; the command keeps running while the agent works on something else. The agent reads its output with `job_output` (non-blocking unless `wait: true`), lists jobs with `job_list`, and stops it with `job_kill`; a finished job notifies the owning agent in-session. Background support needs the generic job runtime (`dsh-jobs-local`) and its control tools (`dsh-tool-jobs`) mounted. The background job declares an observation record: the run's non-consuming `observed` readers are pumped into it at `recordPollMs` (default 150), so the Web client streams live output without touching the job's consuming cursor; a pump failure is logged and swallowed while the job runs on. The record is a best-effort live preview: stdout and stderr are copied per poll round, so writes the two streams made inside one `recordPollMs` window appear stdout first rather than in write order.
 
 ### Sandboxed execution and escalation
 

packages/shell/tool-bash/README.zh.md

@@ -55,7 +55,7 @@ kind: "package-reference"
 
 ### 后台运行长时间命令
 
-传入 `run_in_background: true` 会立即返回 job id，不应用超时；命令继续运行，agent 同时处理其他事情。agent 用 `job_output` 读取输出（除非 `wait: true`，否则非阻塞）、用 `job_list` 列出任务、用 `job_kill` 停止任务；完成的任务会在会话内通知拥有它的 agent。后台支持需要挂载通用任务运行时（`dsh-jobs-local`）及其控制工具（`dsh-tool-jobs`）。后台 job 声明观测 record：运行的非消费 `observed` 读取器以 `recordPollMs`（默认 150）的节奏泵入其中，使 Web 客户端在不触碰 job 消耗游标的情况下流式看到实时输出；泵失败会被记录并吞掉，job 继续运行。
+传入 `run_in_background: true` 会立即返回 job id，不应用超时；命令继续运行，agent 同时处理其他事情。agent 用 `job_output` 读取输出（除非 `wait: true`，否则非阻塞）、用 `job_list` 列出任务、用 `job_kill` 停止任务；完成的任务会在会话内通知拥有它的 agent。后台支持需要挂载通用任务运行时（`dsh-jobs-local`）及其控制工具（`dsh-tool-jobs`）。后台 job 声明观测 record：运行的非消费 `observed` 读取器以 `recordPollMs`（默认 150）的节奏泵入其中，使 Web 客户端在不触碰 job 消耗游标的情况下流式看到实时输出；泵失败会被记录并吞掉，job 继续运行。record 是尽力而为的实时预览：stdout 与 stderr 按轮询轮次复制，同一个 `recordPollMs` 窗口内两条流的写入会先 stdout 后 stderr 出现，而不是按写入顺序。
 
 ### 沙箱执行与升权
 

packages/shell/tool-bash/src/observe.ts

@@ -5,6 +5,9 @@
  * `readOutput` cursor. Observation is strictly best-effort: absent backend
  * offset readers degrade to no observation, and a pump failure is logged and
  * swallowed — the job path is unaffected either way.
+ * The record interleaves stdout and stderr per poll round, not per write:
+ * output the two streams produced inside one `pollMs` window lands stdout
+ * first, so the panel is a live preview, not a terminal-exact transcript.
  *
  * @module @deepseek-ai/dsh-tool-bash/observe
  */

packages/shell/tool-pwsh/src/observe.ts

@@ -5,6 +5,9 @@
  * `readOutput` cursor. Observation is strictly best-effort: absent backend
  * offset readers degrade to no observation, and a pump failure is logged and
  * swallowed — the job path is unaffected either way.
+ * The record interleaves stdout and stderr per poll round, not per write:
+ * output the two streams produced inside one `pollMs` window lands stdout
+ * first, so the panel is a live preview, not a terminal-exact transcript.
  *
  * @module @deepseek-ai/dsh-tool-pwsh/observe
  */

snapshots/web/background-job-list/running.expected.md

@@ -1,5 +1,5 @@
 - list "Background jobs":
   - listitem:
-    - button "Show live output of sleep 45":
-      - text: sleep 45 {{duration}} bash running
+    - button "Show live output of for _ in $(seq 1 3000); do [ -e .background-job-list.release ] && break; sleep 0.2; done":
+      - text: for _ in $(seq 1 3000); do [ -e .background-job-list.release ] && break; sleep 0.2; done {{duration}} bash running
       - img

snapshots/web/background-job-list/settled.expected.md

@@ -1,5 +1,5 @@
 - list "Background jobs":
   - listitem:
-    - button "Show live output of sleep 45":
-      - text: "bash sleep 45 signal: SIGTERM {{duration}}"
+    - button "Show live output of for _ in $(seq 1 3000); do [ -e .background-job-list.release ] && break; sleep 0.2; done":
+      - text: "bash for _ in $(seq 1 3000); do [ -e .background-job-list.release ] && break; sleep 0.2; done signal: SIGTERM {{duration}}"
       - img