diff --git a/docs/subsystems/jobs.i18n.yaml b/docs/subsystems/jobs.i18n.yaml index f71f76aa0e..381cd81097 100644 --- a/docs/subsystems/jobs.i18n.yaml +++ b/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 diff --git a/docs/subsystems/jobs.md b/docs/subsystems/jobs.md index 59cb88bdb5..e8884bae69 100644 --- a/docs/subsystems/jobs.md +++ b/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 /** diff --git a/docs/subsystems/jobs.zh.md b/docs/subsystems/jobs.zh.md index fd2f897979..c75fae067e 100644 --- a/docs/subsystems/jobs.zh.md +++ b/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 /** diff --git a/packages/jobs/jobs-local/README.i18n.yaml b/packages/jobs/jobs-local/README.i18n.yaml index 1405d2ee15..27f7aa65ec 100644 --- a/packages/jobs/jobs-local/README.i18n.yaml +++ b/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 diff --git a/packages/jobs/jobs-local/README.md b/packages/jobs/jobs-local/README.md index 58a35e813a..da48df0917 100644 --- a/packages/jobs/jobs-local/README.md +++ b/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 diff --git a/packages/jobs/jobs-local/README.zh.md b/packages/jobs/jobs-local/README.zh.md index 486a9cef38..f217341b3f 100644 --- a/packages/jobs/jobs-local/README.zh.md +++ b/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` 据此跳过没人能读的通知。 ### 销毁 diff --git a/packages/jobs/jobs-local/src/index.ts b/packages/jobs/jobs-local/src/index.ts index 990c03209c..981be78b5a 100644 --- a/packages/jobs/jobs-local/src/index.ts +++ b/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() diff --git a/packages/jobs/jobs-local/tests/jobs.spec.ts b/packages/jobs/jobs-local/tests/jobs.spec.ts index 55e5615a58..c93d764a36 100644 --- a/packages/jobs/jobs-local/tests/jobs.spec.ts +++ b/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() diff --git a/packages/jobs/jobs/src/types.ts b/packages/jobs/jobs/src/types.ts index 0f46ff1715..b3abd3ac68 100644 --- a/packages/jobs/jobs/src/types.ts +++ b/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.