fix(jobs-local): keep unconsumed settled output for the model and announce registration first

Settlement trimmed the ring to settledRetainBytes before releasing waiters, so a job that finished before its first job_output handed the model only the last 16 KiB plus a dropped-output notice where master's consuming read kept the executor's whole tail. The ring now keeps every byte the model cursor has not consumed on top of the settled cap, and the first terminal model read trims to the cap.

The registered event was emitted after startPump, whose first drain runs synchronously and may announce output, so a source readable at once produced an output event for an id no subscriber had seen registered. The announcement now precedes the pump; the events JSDoc states that a job's first event is registered.

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: cdbeb6df0c0f69458bfae89981c8f2b6d3dfc381
-jobs.zh.md: a502fff316a0c2ebccf7888d4eb5aee10de07576
+jobs.md: e8884bae690188d8653fb02c0009be555a49f6f3
+jobs.zh.md: c75fae067e168a6be309febe888b7215e6edb591

docs/subsystems/jobs.md

@@ -424,7 +424,7 @@ Implementations must honor these semantics:
 - Settlement is first-wins: one terminal record, released waiters, then one round of contained event delivery, even against a late producer outcome. The `settled` event follows every released waiter, so a consumer that claims a settlement while waiting always claims before the event.
 - start refuses work while no attached job controller serves the spec's owner, so a producer cannot start work that owner cannot collect or stop. One registry serves every composition in the process, so this question — and event delivery under `{ owners: 'scope' }` — is owner-relative rather than process-wide: registrations made from an unscoped context serve every owner, and registrations made under an agent composition's scope serve exactly the agents composed under it.
 - Every job owns one output ring. Pull sources named by the spec are pumped by the registry and drained once more before settlement; pushed appends land whole. The model's consuming cursor and observers' absolute offsets read the same bytes and never disturb each other.
-- Ring retention is bounded. Appends past the live cap drop the oldest retained bytes; a reader below the retained window gets a lossy read, never an error. Settlement trims retention to the settled cap and ends the stream; the ring has no separate lifecycle.
+- Ring retention is bounded. Appends past the live cap drop the oldest retained bytes; a reader below the retained window gets a lossy read, never an error. Settlement trims retention to the settled cap, but never below the bytes the model cursor has not consumed, and ends the stream; the first terminal model read trims to the cap. The ring has no separate lifecycle.
 
 ```ts cordis-catalog
 /**

docs/subsystems/jobs.zh.md

@@ -424,7 +424,7 @@ Implementations must honor these semantics:
 - Settlement is first-wins: one terminal record, released waiters, then one round of contained event delivery, even against a late producer outcome. The `settled` event follows every released waiter, so a consumer that claims a settlement while waiting always claims before the event.
 - start refuses work while no attached job controller serves the spec's owner, so a producer cannot start work that owner cannot collect or stop. One registry serves every composition in the process, so this question — and event delivery under `{ owners: 'scope' }` — is owner-relative rather than process-wide: registrations made from an unscoped context serve every owner, and registrations made under an agent composition's scope serve exactly the agents composed under it.
 - Every job owns one output ring. Pull sources named by the spec are pumped by the registry and drained once more before settlement; pushed appends land whole. The model's consuming cursor and observers' absolute offsets read the same bytes and never disturb each other.
-- Ring retention is bounded. Appends past the live cap drop the oldest retained bytes; a reader below the retained window gets a lossy read, never an error. Settlement trims retention to the settled cap and ends the stream; the ring has no separate lifecycle.
+- Ring retention is bounded. Appends past the live cap drop the oldest retained bytes; a reader below the retained window gets a lossy read, never an error. Settlement trims retention to the settled cap, but never below the bytes the model cursor has not consumed, and ends the stream; the first terminal model read trims to the cap. The ring has no separate lifecycle.
 
 ```ts cordis-catalog
 /**

packages/jobs/jobs-local/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/jobs/jobs-local/README.md
-README.md: 58a35e813af03a00cf978d001a12f26bbb848ff4
-README.zh.md: 486a9cef3813d4d07c328339ee179e08e97c98b3
+README.md: da48df0917986b7ecc03ee83d2ab31b5b6e8fad7
+README.zh.md: f217341b3f89864b6fc80d9f76c968ac0dc3f72c

packages/jobs/jobs-local/README.md

@@ -43,7 +43,7 @@ Loading the plugin registers `ctx.jobs`; every field is optional.
 |---|---|---|
 | `maxConcurrentJobsPerOwner` | `10` | Maximum `running` plus `stopping` jobs per exact owner, or in the shared unowned bucket |
 | `retainBytes` | `262144` | Live ring retention per job, in UTF-8 bytes |
-| `settledRetainBytes` | `16384` | Ring retention kept after a job settles, in UTF-8 bytes |
+| `settledRetainBytes` | `16384` | Ring retention kept after a job settles, in UTF-8 bytes; bytes the model has not read stay until its first terminal read |
 | `pumpPollMs` | `150` | Poll interval for a job's pull sources, in milliseconds |
 
 The generated [configuration catalog](../../../docs/config-catalog.md#deepseek-aidsh-jobs-local) is the exhaustive source for the accepted fields.
@@ -73,7 +73,7 @@ This section explains the design decisions behind the registry and points at the
 ### Design philosophy
 
 - **In-memory records, fresh projections.** `LocalJobRegistry` keeps one `TrackedJob` per job — lifecycle state, the output ring, and the model cursor — and projects a new read-only view or chunk copy per call; callers never receive live state.
-- **Retention is bounded per ring.** Appends past the live cap drop the oldest retained chunks (a single oversized chunk keeps its UTF-8-safe tail); a reader below the retained window gets a lossy read, never an error, and settlement trims to the settled cap.
+- **Retention is bounded per ring.** Appends past the live cap drop the oldest retained chunks (a single oversized chunk keeps its UTF-8-safe tail); a reader below the retained window gets a lossy read, never an error. Settlement trims to the settled cap but never below the bytes the model cursor has not consumed, so a job that finishes before its first `job_output` hands over everything the live cap retained; that terminal read then trims to the settled cap.
 - **Owner-relative layers, one process-wide registry.** Controllers and `{ owners: 'scope' }` subscriptions are filed into the scope that registered them (`ScopedLayers`), and reads union the global layer with the owner's scope chain — so one preset's job controls never hold `start()` open for an agent whose own composition loads none, and a scoped settlement reaches only the listeners its owner's composition registered.
 - **Preflight before start.** `start()` checks controller service, spec validity, live ownership, and capacity before invoking the producer, so a rejection leaves no job id or execution resource; registration commits without a later failable step.
 - **First-wins settlement, event last.** The earliest terminal outcome records once, waits for the pump's final drain, trims the ring, releases waiters, and then delivers one `settled` event with per-listener containment, followed by the ring's final `output` signal.
@@ -87,7 +87,7 @@ This section explains the design decisions behind the registry and points at the
 | [`src/events.ts`](src/events.ts) | Scope-layered event routing: `{ owner }`, `{ owners: 'all' }`, and `{ owners: 'scope' }` subscriptions |
 | [`src/ring.ts`](src/ring.ts) | The bounded per-job output ring: append, retention trim, offset reads |
 | [`src/pump.ts`](src/pump.ts) | The registry-owned pull pump: one timer per job, final drain before settlement |
-| — | No runtime invariant companion is published; `@deepseek-ai/dsh-jobs/invariant` owns per-snapshot identity, status, timestamp, and owner checks. This provider's admission decision uses private configuration and must fail before a backend starter runs; `LocalJobRegistry.start()` enforces it synchronously for current producers. Repeating an aggregate after publication would expose private configuration solely to this companion and would not verify the fail-closed pre-start guarantee. |
+| — | No runtime invariant companion is published; `@deepseek-ai/dsh-jobs/invariant` owns the event-protocol and event-versus-read checks. This provider's admission decision uses private configuration and must fail before a backend starter runs; `LocalJobRegistry.start()` enforces it synchronously for current producers. Repeating an aggregate after publication would expose private configuration solely to this companion and would not verify the fail-closed pre-start guarantee. |
 
 ### Scope layers
 
@@ -95,7 +95,7 @@ This section explains the design decisions behind the registry and points at the
 
 ### Admission and settlement
 
-`activeTaskCount` counts authoritative records per exact owner or in the shared unowned bucket. `settle` records the terminal outcome once (merging a recorded kill reason into a `killed` detail), clears the progress line, trims the ring to the settled cap, resolves every waiter, then emits `settled` with its cause and the ring's final `output` signal. The cause is `kill` after `VisibleJobs.kill`, `teardown` after an owner or service cancel, and `producer` otherwise; `dsh-tool-jobs` uses it to skip notices nobody can read.
+`activeTaskCount` counts authoritative records per exact owner or in the shared unowned bucket. `settle` records the terminal outcome once (merging a recorded kill reason into a `killed` detail), clears the progress line, trims the ring to the settled cap (keeping every byte the model cursor has not consumed), resolves every waiter, then emits `settled` with its cause and the ring's final `output` signal. The cause is `kill` after `VisibleJobs.kill`, `teardown` after an owner or service cancel, and `producer` otherwise; `dsh-tool-jobs` uses it to skip notices nobody can read.
 
 ### Teardown
 

packages/jobs/jobs-local/README.zh.md

@@ -43,7 +43,7 @@ kind: "package-reference"
 |---|---|---|
 | `maxConcurrentJobsPerOwner` | `10` | 每个精确所有者，或共享的无主桶中，`running` 加 `stopping` 任务的最大数量 |
 | `retainBytes` | `262144` | 每个任务输出环的运行期保留量，UTF-8 字节 |
-| `settledRetainBytes` | `16384` | 任务结算后保留的环容量，UTF-8 字节 |
+| `settledRetainBytes` | `16384` | 任务结算后保留的环容量，UTF-8 字节；模型尚未读取的字节保留到它的首次终态读取 |
 | `pumpPollMs` | `150` | 任务拉取源的轮询间隔，毫秒 |
 
 生成的[配置目录](../../../docs/config-catalog.zh.md#deepseek-aidsh-jobs-local)是每个受支持字段的穷尽式真源。
@@ -73,7 +73,7 @@ kind: "package-reference"
 ### 设计理念
 
 - **内存记录、全新投影。** `LocalJobRegistry` 为每个任务保存一份 `TrackedJob`——生命周期状态、输出环与模型游标——每次调用都投影新的只读视图或块副本；调用方永远拿不到实时状态。
-- **环的保留量有界。** 追加超出运行期上限时丢弃最老的保留块（单个超限块保留其 UTF-8 安全尾部）；低于保留窗口的读取得到 lossy 结果而非错误，结算时裁剪到结算上限。
+- **环的保留量有界。** 追加超出运行期上限时丢弃最老的保留块（单个超限块保留其 UTF-8 安全尾部）；低于保留窗口的读取得到 lossy 结果而非错误。结算时裁剪到结算上限，但绝不低于模型游标尚未消费的字节数，因此在首次 `job_output` 之前结束的任务会交出运行期上限保留的全部输出；那次终态读取之后才裁剪到结算上限。
 - **按所有者分层，一个进程级注册表。** 控制器与 `{ owners: 'scope' }` 订阅归档到注册方所在的 scope（`ScopedLayers`），读取把全局层与所有者的 scope 链求并集——因此某个 preset 的任务控制绝不会为自身组合未加载任何控制器的 agent 保持 `start()` 可用，一次带 scope 的结算也只会抵达其所有者所属组合注册的监听器。
 - **启动前先预检。** `start()` 在调用生产方之前检查控制器服务、spec 有效性、仍存活的所有权与容量，因此拒绝不会留下 job id 或执行资源；注册一旦提交，后续不再有可失败步骤。
 - **结算首次优先，事件最后。** 最早的终止结果只记录一次，等待泵的最后一次排干，裁剪环，释放等待方，然后投递一次带逐监听器隔离的 `settled` 事件，再跟上环的最终 `output` 信号。
@@ -87,7 +87,7 @@ kind: "package-reference"
 | [`src/events.ts`](src/events.ts) | 按 scope 分层的事件路由：`{ owner }`、`{ owners: 'all' }` 与 `{ owners: 'scope' }` 订阅 |
 | [`src/ring.ts`](src/ring.ts) | 每个任务的有界输出环：追加、保留裁剪、按偏移读取 |
 | [`src/pump.ts`](src/pump.ts) | 注册表拥有的拉取泵：每个任务一个定时器，结算前最后一次排干 |
-| — | 不发布运行时不变式伴生入口；快照检查位于 `dsh-jobs/invariant`。 |
+| — | 不发布运行时不变式伴生入口；事件协议与事件对读取的检查位于 `dsh-jobs/invariant`。 |
 
 ### scope 分层
 
@@ -95,7 +95,7 @@ kind: "package-reference"
 
 ### 准入与结算
 
-`activeTaskCount` 按精确所有者或共享无主桶统计权威记录。`settle` 只记录一次终止结果（把记录下来的 kill 原因合并进 `killed` 的 detail），清除进度行，把环裁剪到结算保留量，解析每个等待方，然后发出带原因的 `settled` 与环的最终 `output` 信号。原因在 `VisibleJobs.kill` 之后为 `kill`，在所有者或服务取消之后为 `teardown`，否则为 `producer`；`dsh-tool-jobs` 据此跳过没人能读的通知。
+`activeTaskCount` 按精确所有者或共享无主桶统计权威记录。`settle` 只记录一次终止结果（把记录下来的 kill 原因合并进 `killed` 的 detail），清除进度行，把环裁剪到结算保留量（保留模型游标尚未消费的全部字节），解析每个等待方，然后发出带原因的 `settled` 与环的最终 `output` 信号。原因在 `VisibleJobs.kill` 之后为 `kill`，在所有者或服务取消之后为 `teardown`，否则为 `producer`；`dsh-tool-jobs` 据此跳过没人能读的通知。
 
 ### 销毁
 

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

@@ -50,7 +50,11 @@ export interface Config {
   maxConcurrentJobsPerOwner?: number
   /** Live ring retention per job in UTF-8 bytes; omission defaults to 262144. */
   retainBytes?: number
-  /** Ring retention kept after a job settles, in UTF-8 bytes; omission defaults to 16384. */
+  /**
+   * Ring retention kept after a job settles, in UTF-8 bytes; omission defaults to 16384.
+   * Settlement keeps every byte the model cursor has not consumed on top of
+   * this cap; the first terminal model read then trims to it.
+   */
   settledRetainBytes?: number
   /** Poll interval for a job's pull sources, in milliseconds; omission defaults to 150. */
   pumpPollMs?: number
@@ -257,6 +261,11 @@ export class LocalJobRegistry extends JobRegistry {
     // call reaches the registered record for its terminal checks and signals.
     state.job = job
     this.store.set(id, job)
+    // Registration is complete and cannot fail from here, so the visible set
+    // has genuinely changed. The announcement precedes the pump because the
+    // pump drains its sources once synchronously, and that drain may append
+    // and announce output: a job's first event is always `registered`.
+    this.emit({ type: 'registered', job: this.view(job) }, owner)
 
     // The producer's settlement or a registry-forced one ends the pump; the
     // pump's final drain then lands before this registry trims the ring.
@@ -280,9 +289,6 @@ export class LocalJobRegistry extends JobRegistry {
       if (job.pump !== undefined) await job.pump.done
       this.settle(job, outcome, job.settleCause ?? 'producer')
     })
-    // Registration is complete and cannot fail from here, so the visible set
-    // has genuinely changed.
-    this.emit({ type: 'registered', job: this.view(job) }, owner)
     return id
   }
 
@@ -395,12 +401,17 @@ export class LocalJobRegistry extends JobRegistry {
     this.hub.emit(event, owner)
   }
 
-  /** Consume the ring from the model cursor; the result rides the first read after settlement. */
+  /**
+   * Consume the ring from the model cursor; the result rides the first read
+   * after settlement. A terminal read is the point the settled stream drops
+   * to the settled cap: settlement kept every unconsumed byte for it.
+   */
   private readJob(job: TrackedJob): JobRead {
     const read = job.ring.readFrom(job.modelCursor)
     job.modelCursor = job.ring.total
     const result = isTerminal(job.status) && !job.resultDelivered ? job.result : undefined
     if (result !== undefined) job.resultDelivered = true
+    if (isTerminal(job.status)) job.ring.trim(this.settledRetainBytes)
     return {
       chunks: read.chunks,
       lossy: read.lossy,
@@ -552,8 +563,10 @@ export class LocalJobRegistry extends JobRegistry {
     job.result = outcome.result
     job.finishedAt = Date.now()
     // Settlement ends the stream: trim to the settled cap before any observer
-    // reads the terminal projection.
-    job.ring.trim(this.settledRetainBytes)
+    // reads the terminal projection, but never below the bytes the model
+    // cursor has not consumed. A job that finishes before its first model
+    // read keeps everything the live cap retained until that read.
+    job.ring.trim(Math.max(this.settledRetainBytes, job.ring.total - job.modelCursor))
     const waitResolvers = [...job.waitResolvers]
     job.waitResolvers.clear()
     for (const resolveWait of waitResolvers) resolveWait()

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

@@ -1194,6 +1194,22 @@ describe('LocalJobRegistry output ring', () => {
     expect(() => jobsOf(ctx).readAt(JobId('bash-99'), 0)).toThrow(/unknown job/)
   })
 
+  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)
+    const source: JobOutputSource = {
+      read: from => from === 0 ? { text: 'boot', nextOffset: 4, lossy: false } : { text: '', nextOffset: from, lossy: false },
+    }
+    const p = producer({ output: [source] })
+    const id = ctx.jobs.start(p.spec)
+    // The pump's first drain runs synchronously inside start(); the job's
+    // first event is still its registration.
+    expect(events.slice(0, 2).map(event => event.type)).toEqual(['registered', 'output'])
+    expect(jobsOf(ctx).readAt(id, 0).chunks.map(chunk => chunk.text)).toEqual(['boot'])
+    p.settle({ status: 'completed' })
+    await tick()
+  })
+
   it('evicts whole head chunks past the live cap and flags stale readers lossy', async () => {
     const ctx = await harness({ retainBytes: 8 })
     const p = producer()
@@ -1212,7 +1228,7 @@ describe('LocalJobRegistry output ring', () => {
     expect(jobsOf(ctx).read(id).lossy).toBe(true)
   })
 
-  it('settlement trims to the settled cap, ends the stream, and drops later writes', async () => {
+  it('settlement keeps the unconsumed stream for the model; the terminal read trims to the settled cap, ends the stream, and later writes drop', async () => {
     const ctx = await harness({ retainBytes: 1024, settledRetainBytes: 4 })
     const warn = vi.fn()
     ctx.logger.warn = warn as never
@@ -1223,7 +1239,14 @@ describe('LocalJobRegistry output ring', () => {
     p.settle({ status: 'completed', detail: 'exit code: 0' })
     await tick()
     expect(outputs.map(event => event.type === 'output' ? event.total : -1)).toEqual([8, 8])
-    expect(jobsOf(ctx).get(id)).toMatchObject({ status: 'completed', detail: 'exit code: 0', output: { total: 8, earliest: 4 } })
+    // Nothing the model has not read is trimmed at settlement: a job that
+    // finishes before its first read still hands over everything the live
+    // cap retained, and observers see the same bytes.
+    expect(jobsOf(ctx).get(id)).toMatchObject({ status: 'completed', detail: 'exit code: 0', output: { total: 8, earliest: 0 } })
+    expect(jobsOf(ctx).readAt(id, 0)).toMatchObject({ lossy: false, chunks: [{ at: 0, text: 'abcdefgh' }] })
+    expect(jobsOf(ctx).read(id)).toMatchObject({ lossy: false, chunks: [{ at: 0, text: 'abcdefgh' }] })
+    // The terminal read consumed the stream: retention drops to the settled cap.
+    expect(jobsOf(ctx).get(id).output).toEqual({ total: 8, earliest: 4 })
     const read = jobsOf(ctx).readAt(id, 0)
     expect(read.lossy).toBe(true)
     expect(read.chunks).toEqual([{ at: 4, text: 'efgh', gapBefore: true }])
@@ -1238,6 +1261,19 @@ describe('LocalJobRegistry output ring', () => {
     expect(jobsOf(ctx).get(id).progress).toBeUndefined()
   })
 
+  it('a settlement after the model consumed the stream trims to the settled cap at once', async () => {
+    const ctx = await harness({ retainBytes: 1024, settledRetainBytes: 4 })
+    const p = producer()
+    const id = ctx.jobs.start(p.spec)
+    p.job().append('abcdefgh')
+    expect(jobsOf(ctx).read(id).chunks).toEqual([{ at: 0, text: 'abcdefgh' }])
+    p.settle({ status: 'completed' })
+    await tick()
+    expect(jobsOf(ctx).get(id).output).toEqual({ total: 8, earliest: 4 })
+    // The cursor sits at the end: nothing new, and nothing lost behind it.
+    expect(jobsOf(ctx).read(id)).toMatchObject({ chunks: [], lossy: false })
+  })
+
   it('settlement clears the progress line; the terminal detail is the outcome alone', async () => {
     const ctx = await harness()
     const p = producer()

packages/jobs/jobs/src/types.ts

@@ -296,8 +296,9 @@ export type JobEventListener = (event: JobEvent) => void
 export interface JobEvents {
   /**
    * Register an effect-scoped listener. Events are dispatched synchronously
-   * after the commit they announce; a `settled` event follows the release of
-   * every waiter. No listener runs after service disposal.
+   * after the commit they announce; a job's first event is `registered`, and
+   * a `settled` event follows the release of every waiter. No listener runs
+   * after service disposal.
    * @param filter - which owners' events to deliver.
    * @param listener - receives each matching event.
    * @returns disposer that unregisters the listener.