From e680fbc3a0acb2afa5fce49d01a34a35a00ce6d6 Mon Sep 17 00:00:00 2001 From: Yichen Jiang Date: Fri, 4 Sep 2026 12:27:16 +0800 Subject: [PATCH] refactor(jobs): check the announced event protocol in the invariant companion The companion validated fields of single JobView projections, which the invariant rule rejects as a check without independent observations. It now tracks each job's announced protocol (registered first, one settlement, removal last, no progress after settlement), adopting a job first seen late, and compares every announced projection with the registry's own read of the same job: identity, output total, and the settled status and time. An output event announced before registration, which the registry emitted until the previous commit, is one of the cases it rejects. --- .../2026-07-26-job-registry-seam.i18n.yaml | 4 +- .../2026-07-26-job-registry-seam.md | 2 +- .../2026-07-26-job-registry-seam.zh.md | 2 +- packages/api/job-controller/README.i18n.yaml | 4 +- packages/api/job-controller/README.md | 2 +- packages/api/job-controller/README.zh.md | 2 +- packages/jobs/jobs/README.i18n.yaml | 4 +- packages/jobs/jobs/README.md | 2 +- packages/jobs/jobs/README.zh.md | 2 +- packages/jobs/jobs/src/invariant.ts | 125 ++++++++--- packages/jobs/jobs/tests/invariant.spec.ts | 211 ++++++++++++------ 11 files changed, 250 insertions(+), 110 deletions(-) diff --git a/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.i18n.yaml b/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.i18n.yaml index e9122ec8c6..78fba181dc 100644 --- a/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.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-07-26-job-registry-seam.md -2026-07-26-job-registry-seam.md: 53925b5b574432920c2128d63b9202e22caeb0d3 -2026-07-26-job-registry-seam.zh.md: 1e86187e35fd1345ce3a88dc7765b989b5ec2c97 +2026-07-26-job-registry-seam.md: b48a0896b7db202b5d7439500f1fa4e54cee97e3 +2026-07-26-job-registry-seam.zh.md: 0b800430fdf4f77802e6203e948b9defea881d4f diff --git a/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.md b/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.md index 53925b5b57..b48a0896b7 100644 --- a/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.md +++ b/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.md @@ -12,7 +12,7 @@ The [background-job runtime](2026-06-20-generic-long-running-tool-runtime.md) sh `jobs/` is now a three-package capability family in the bash-trio shape: -- **`@deepseek-ai/dsh-jobs` (Service Definition)** — the abstract `JobRegistry extends Service` owning `ctx.jobs`, the contract (`start`, the caller-bound `visibleTo` view with `list`, `get`, `read`, `readAt`, `kill`, and `wait`, the `events` stream, and `attachController`), all vocabulary types (`JobId`, `JobKindMap`, `JobSpec`, `JobHandle`, `JobHooks`, `JobOutcome`, `JobView`, `JobRead`, `JobEvent`), and the projection invariant companion. The class-level JSDoc states the semantics every Service Provider owes: registrations outlive producer and controller fibers, owned access is session-fenced, settlement is first-wins with contained listeners, and `start` refuses work while no attached job controller serves the spec's owner (controllers and event subscriptions are scope-layered, so one process-wide registry answers both questions per owner). +- **`@deepseek-ai/dsh-jobs` (Service Definition)** — the abstract `JobRegistry extends Service` owning `ctx.jobs`, the contract (`start`, the caller-bound `visibleTo` view with `list`, `get`, `read`, `readAt`, `kill`, and `wait`, the `events` stream, and `attachController`), all vocabulary types (`JobId`, `JobKindMap`, `JobSpec`, `JobHandle`, `JobHooks`, `JobOutcome`, `JobView`, `JobRead`, `JobEvent`), and the event-protocol invariant companion. The class-level JSDoc states the semantics every Service Provider owes: registrations outlive producer and controller fibers, owned access is session-fenced, settlement is first-wins with contained listeners, and `start` refuses work while no attached job controller serves the spec's owner (controllers and event subscriptions are scope-layered, so one process-wide registry answers both questions per owner). - **`@deepseek-ai/dsh-jobs-local` (Service Provider)** — `LocalJobRegistry`, the process-local registry: the in-memory store, per-kind id counters, waiter bookkeeping, `TASK_WAIT_TIMEOUT` deadline code, owner-cleanup effects, force-fail teardown, and the default-10 configurable admission policy. Admission derives `running` plus `stopping` capacity from the same records per exact owner, with one unowned bucket; it adds no public count or second state owner. The `dsh-timeout` dependency and Schemastery-owned provider config live here; the Service Definition package has no provider dependencies. - **`@deepseek-ai/dsh-tool-jobs` (Consumer)** — unchanged; it injects `'jobs'` and never imports provider types. diff --git a/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.zh.md b/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.zh.md index 1e86187e35..0b800430fd 100644 --- a/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.zh.md +++ b/.agents/notes/implemented/architecture/2026-07-26-job-registry-seam.zh.md @@ -12,7 +12,7 @@ Status: implemented `jobs/` 如今是一个 bash 三件套形态的三包能力家族： -- **`@deepseek-ai/dsh-jobs`（Service Definition）**——抽象的 `JobRegistry extends Service`，拥有 `ctx.jobs`、约定（`start`、绑定调用方的 `visibleTo` 视图及其 `list`、`get`、`read`、`readAt`、`kill`、`wait`、`events` 事件流，以及 `attachController`）、全部词汇类型（`JobId`、`JobKindMap`、`JobSpec`、`JobHandle`、`JobHooks`、`JobOutcome`、`JobView`、`JobRead`、`JobEvent`），以及投影不变式配套插件。类级 JSDoc 陈述了每个 Service Provider 都必须兑现的语义：注册的存续期长于生产方与控制器的 fiber，有所有者的访问以会话为界，结算遵循首次结果优先且监听器错误被隔离，并且当没有任何已附加的任务控制器服务于 spec 的所有者时 `start` 拒绝启动工作（控制器与事件订阅按 scope 分层，因此一个进程级注册表能逐所有者地回答这两个问题）。 +- **`@deepseek-ai/dsh-jobs`（Service Definition）**——抽象的 `JobRegistry extends Service`，拥有 `ctx.jobs`、约定（`start`、绑定调用方的 `visibleTo` 视图及其 `list`、`get`、`read`、`readAt`、`kill`、`wait`、`events` 事件流，以及 `attachController`）、全部词汇类型（`JobId`、`JobKindMap`、`JobSpec`、`JobHandle`、`JobHooks`、`JobOutcome`、`JobView`、`JobRead`、`JobEvent`），以及事件协议不变式配套插件。类级 JSDoc 陈述了每个 Service Provider 都必须兑现的语义：注册的存续期长于生产方与控制器的 fiber，有所有者的访问以会话为界，结算遵循首次结果优先且监听器错误被隔离，并且当没有任何已附加的任务控制器服务于 spec 的所有者时 `start` 拒绝启动工作（控制器与事件订阅按 scope 分层，因此一个进程级注册表能逐所有者地回答这两个问题）。 - **`@deepseek-ai/dsh-jobs-local`（Service Provider）**——`LocalJobRegistry`，即进程内注册表：内存存储、按 kind 划分的 id 计数器、等待方簿记、`TASK_WAIT_TIMEOUT` deadline 代码、所有者清理 effect、强制失败的拆除，以及默认值为 10 且可配置的准入策略。准入从同一组记录中按确切 owner 派生 `running` 加 `stopping` 容量，并为无 owner 任务使用一个共享桶；它不新增公开计数或第二个状态 owner。`dsh-timeout` 依赖与由 Schemastery 管理的 Service Provider 配置都位于此包；Service Definition 包不含任何提供方依赖。 - **`@deepseek-ai/dsh-tool-jobs`（Consumer）**——保持不变；它注入 `'jobs'`，从不导入提供方类型。 diff --git a/packages/api/job-controller/README.i18n.yaml b/packages/api/job-controller/README.i18n.yaml index 8ad9d12dc8..26f6e38732 100644 --- a/packages/api/job-controller/README.i18n.yaml +++ b/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: b21327eeea33d84d9218f6869fbe9f9b3d8c6052 -README.zh.md: 32e67646d10f49e2626f0dc2a094e1eb303cc211 +README.md: f4f7a5142e649d0b8335128f2c4012dd957ec676 +README.zh.md: eeb10819fc2ab17f1e2603d5014618874c5efaf7 diff --git a/packages/api/job-controller/README.md b/packages/api/job-controller/README.md index b21327eeea..f4f7a5142e 100644 --- a/packages/api/job-controller/README.md +++ b/packages/api/job-controller/README.md @@ -64,4 +64,4 @@ None. -**Runtime invariant:** No companion is published. The controller is a stateless projection of `ctx.jobs` reads; the registry's own `@deepseek-ai/dsh-jobs/invariant` owns the offset relations these streams forward. +**Runtime invariant:** No companion is published. The controller is a stateless projection of `ctx.jobs` reads; the registry's own `@deepseek-ai/dsh-jobs/invariant` owns the event protocol and event-versus-read relations these streams forward. diff --git a/packages/api/job-controller/README.zh.md b/packages/api/job-controller/README.zh.md index 32e67646d1..eeb10819fc 100644 --- a/packages/api/job-controller/README.zh.md +++ b/packages/api/job-controller/README.zh.md @@ -64,4 +64,4 @@ Client 入口在 `ClientJobsModel` 之上提供 `ClientJobs`（`ctx.jobs`）。` -**运行时不变式：** 不发布伴生入口。控制器是 `ctx.jobs` 读取的无状态投影；这些流转发的偏移关系由注册表自己的 `@deepseek-ai/dsh-jobs/invariant` 拥有。 +**运行时不变式：** 不发布伴生入口。控制器是 `ctx.jobs` 读取的无状态投影；这些流转发的事件协议与事件对读取的关系由注册表自己的 `@deepseek-ai/dsh-jobs/invariant` 拥有。 diff --git a/packages/jobs/jobs/README.i18n.yaml b/packages/jobs/jobs/README.i18n.yaml index ca8c2269ef..11407349e0 100644 --- a/packages/jobs/jobs/README.i18n.yaml +++ b/packages/jobs/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/jobs/jobs/README.md -README.md: 4a82606f7988827ce8279c649f0fe1caeb34745e -README.zh.md: 56e4711acd550b7cb1278739cb60415c9e7ad065 +README.md: 491f082d8083e75a1149dbed33f6c697e2f785ac +README.zh.md: a15b4f7b458fde206bd548e02a6e8b91993ea0ba diff --git a/packages/jobs/jobs/README.md b/packages/jobs/jobs/README.md index 4a82606f79..491f082d80 100644 --- a/packages/jobs/jobs/README.md +++ b/packages/jobs/jobs/README.md @@ -80,7 +80,7 @@ This section explains the design decisions behind the contract and points at the | [`src/types.ts`](src/types.ts) | Shared vocabulary: `JobSpec`, `JobHandle`, `JobHooks`, `JobOutcome`, `VisibleJobs`, `JobEvent`, and the read results | | [`src/view.ts`](src/view.ts) | Client-safe leaf: `JobView`, `JobChunk`, `JobStatus`, and the merge-extensible `JobKindMap` | | [`src/brand.ts`](src/brand.ts) | `JobId` branded identifier, importable without the agent dependency | -| [`src/invariant.ts`](src/invariant.ts) | Invariant companion: validates every projection the registry hands out — identity, status, timestamps, owner, and ring offsets | +| [`src/invariant.ts`](src/invariant.ts) | Invariant companion: checks the announced event protocol per job (registered first, one settlement, removal last) and each announced projection against the registry's own read | ### Service operations diff --git a/packages/jobs/jobs/README.zh.md b/packages/jobs/jobs/README.zh.md index 56e4711acd..a15b4f7b45 100644 --- a/packages/jobs/jobs/README.zh.md +++ b/packages/jobs/jobs/README.zh.md @@ -80,7 +80,7 @@ kind: "package-reference" | [`src/types.ts`](src/types.ts) | 共享词汇：`JobSpec`、`JobHandle`、`JobHooks`、`JobOutcome`、`VisibleJobs`、`JobEvent` 与读取结果 | | [`src/view.ts`](src/view.ts) | 客户端安全叶子：`JobView`、`JobChunk`、`JobStatus` 与可合并扩展的 `JobKindMap` | | [`src/brand.ts`](src/brand.ts) | `JobId` 带类型标记的标识符，无需 agent 依赖即可导入 | -| [`src/invariant.ts`](src/invariant.ts) | 不变式伴生插件：校验注册表交出的每个投影——标识、状态、时间戳、所有者与环偏移 | +| [`src/invariant.ts`](src/invariant.ts) | 不变式伴生插件：校验每个 job 的事件协议（先 registered、恰一次结算、最后 removed）以及每个通告的投影与注册表自身读取的一致性 | ### 服务操作 diff --git a/packages/jobs/jobs/src/invariant.ts b/packages/jobs/jobs/src/invariant.ts index 84e3a22fe2..529690804e 100644 --- a/packages/jobs/jobs/src/invariant.ts +++ b/packages/jobs/jobs/src/invariant.ts @@ -1,54 +1,109 @@ -/** Package-owned background-job projection invariants. @module @deepseek-ai/dsh-jobs/invariant */ +/** Package-owned background-job event-protocol invariants. @module @deepseek-ai/dsh-jobs/invariant */ import type { Context } from '@deepseek-ai/cordis' import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants' -import type { JobView } from './types.ts' +import type { JobId } from './brand.ts' +import type { JobEvent, JobView } from './types.ts' const PACKAGE_NAME = '@deepseek-ai/dsh-jobs' const TERMINAL_STATUSES = new Set(['completed', 'killed', 'failed']) +/** + * Where one announced job stands: `live` from its first event until its + * settlement, `settled` until its removal. An id first seen through a later + * event is adopted at that phase, so a companion mounted after the job + * started raises nothing. + */ +type Phase = 'live' | 'settled' + /** Cordis companion plugin name. */ export const name = 'jobs-invariant' /** Service required before the companion can reserve package ownership. */ export const inject = ['invariants'] -/** Validate the cross-field relationships in one registry projection. */ -function validateView(job: JobView, fail: InvariantFailure): void { - const id = String(job.id) - const prefix = `${job.kind}-` - const ordinal = Number(id.slice(prefix.length)) - if (job.kind.length === 0 || !id.startsWith(prefix) - || !Number.isSafeInteger(ordinal) || ordinal < 1) { - fail(`job view id ${JSON.stringify(id)} must be ${JSON.stringify(prefix)} followed by a positive ordinal`) - } - if (job.label.length === 0) fail(`job ${JSON.stringify(id)} label must be non-empty`) - if (!Number.isSafeInteger(job.startedAt) || job.startedAt < 0) { - fail(`job ${JSON.stringify(id)} startedAt must be a non-negative epoch integer`) - } - - const terminal = TERMINAL_STATUSES.has(job.status) - if (terminal !== (job.finishedAt !== undefined)) { - fail(`job ${JSON.stringify(id)} finishedAt must be present exactly for a terminal status`) - } - if (job.finishedAt !== undefined - && (!Number.isSafeInteger(job.finishedAt) || job.finishedAt < job.startedAt)) { - fail(`job ${JSON.stringify(id)} finishedAt must be an epoch integer no earlier than startedAt`) - } - if (terminal && job.progress !== undefined) { - fail(`job ${JSON.stringify(id)} progress must be cleared once settled`) - } - - const { total, earliest } = job.output - if (!Number.isSafeInteger(total) || !Number.isSafeInteger(earliest) || earliest < 0 || earliest > total) { - fail(`job ${JSON.stringify(id)} output offsets must satisfy 0 <= earliest <= total`) +/** The registry's own current projection of `id`, or undefined once the record is gone. */ +function readBack(ctx: Context, id: JobId, owner: JobView['owner']): JobView | undefined { + try { + return ctx.jobs.visibleTo(owner).get(id) + } catch { + // The registry throws for an id outside the caller's set; after a removal that is the expected answer. + return undefined } } -/** Install checks over the current unowned projections and every settlement. */ +/** + * The announced projection must agree with the registry's own read: the + * immutable identity matches, and the announced output total is not ahead of + * the committed one. + */ +function checkAnnounced(read: JobView, announced: JobView, what: string, fail: InvariantFailure): void { + const id = String(announced.id) + for (const key of ['kind', 'label', 'owner', 'startedAt'] as const) { + if (read[key] !== announced[key]) { + fail(`${what} for job ${id} announces ${key} ${JSON.stringify(announced[key])} while the registry reads ${JSON.stringify(read[key])}`) + } + } + if (announced.output.total > read.output.total) { + fail(`${what} for job ${id} announces output total ${announced.output.total} ahead of the registry's ${read.output.total}`) + } +} + +/** Install the per-job protocol and event-versus-read checks over the whole event stream. */ const install: InvariantInstaller = Object.assign((ctx: Context, fail: InvariantFailure) => { - for (const job of ctx.jobs.visibleTo().list()) validateView(job, fail) - ctx.jobs.events.subscribe({ owners: 'all' }, (event) => { - if (event.type === 'settled') validateView(event.job, fail) + const phases = new Map() + ctx.jobs.events.subscribe({ owners: 'all' }, (event: JobEvent) => { + if (event.type === 'output') { + const id = String(event.id) + const read = readBack(ctx, event.id, event.owner) + if (read === undefined) fail(`output announced for job ${id} that the registry no longer returns`) + if (event.total > read.output.total) { + fail(`output announced for job ${id} at total ${event.total} ahead of the registry's ${read.output.total}`) + } + if (!phases.has(id)) phases.set(id, 'live') + return + } + const { job } = event + const id = String(job.id) + const phase = phases.get(id) + const terminal = TERMINAL_STATUSES.has(job.status) + switch (event.type) { + case 'registered': + if (phase !== undefined) fail(`registered announced for job ${id} after earlier events for the same id`) + if (terminal || job.finishedAt !== undefined) fail(`registered job ${id} must announce a live status without finishedAt`) + break + case 'progress': + case 'stopping': + if (phase === 'settled') fail(`${event.type} announced for job ${id} after its settlement`) + if (terminal) fail(`${event.type} announced for job ${id} with a terminal status`) + break + case 'settled': + if (phase === 'settled') fail(`settled announced twice for job ${id}`) + if (!terminal) fail(`settled job ${id} must announce a terminal status, got ${JSON.stringify(job.status)}`) + if (job.finishedAt === undefined || job.finishedAt < job.startedAt) { + fail(`settled job ${id} must announce finishedAt no earlier than startedAt`) + } + if (job.progress !== undefined) fail(`settled job ${id} must announce a cleared progress line`) + break + case 'removed': + if (phase === 'live') fail(`removed announced for job ${id} before its settlement`) + if (readBack(ctx, job.id, job.owner) !== undefined) fail(`removed announced for job ${id} that the registry still returns`) + phases.delete(id) + return + /* v8 ignore start -- defensive: the lifecycle event union is closed and exhaustive here; a new variant fails loudly */ + default: { + const exhaustive: never = event + fail(`unknown job event ${String(exhaustive)} for job ${id}`) + } + /* v8 ignore stop */ + } + const read = readBack(ctx, job.id, job.owner) + if (read === undefined) fail(`${event.type} announced for job ${id} that the registry does not return`) + checkAnnounced(read, job, event.type, fail) + if (event.type === 'settled' && (read.status !== job.status || read.finishedAt !== job.finishedAt)) { + fail(`settled job ${id} announces ${job.status} at ${String(job.finishedAt)}` + + ` while the registry reads ${read.status} at ${String(read.finishedAt)}`) + } + phases.set(id, event.type === 'settled' ? 'settled' : 'live') }) }, { inject: ['jobs'] }) diff --git a/packages/jobs/jobs/tests/invariant.spec.ts b/packages/jobs/jobs/tests/invariant.spec.ts index 12a6d48bee..51d0188dc4 100644 --- a/packages/jobs/jobs/tests/invariant.spec.ts +++ b/packages/jobs/jobs/tests/invariant.spec.ts @@ -2,22 +2,14 @@ import { describe, expect, it } from 'vitest' import { Context } from '@deepseek-ai/cordis' import { SessionId } from '@deepseek-ai/dsh-session' import JobRegistry, { JobId } from '@deepseek-ai/dsh-jobs' -import type { JobEventListener, JobView } from '@deepseek-ai/dsh-jobs' +import type { JobEvent, JobEventListener, JobView } from '@deepseek-ai/dsh-jobs' import * as JobsInvariant from '@deepseek-ai/dsh-jobs/invariant' import InvariantRegistry from '@deepseek-ai/dsh-invariants' -const BASE: JobView = { - id: JobId('bash-1'), - kind: 'bash', - label: 'compile', - status: 'completed', - startedAt: 10, - finishedAt: 20, - output: { total: 0, earliest: 0 }, -} +const ID = JobId('bash-1') const RUNNING: JobView = { - id: JobId('bash-1'), + id: ID, kind: 'bash', label: 'compile', status: 'running', @@ -25,20 +17,24 @@ const RUNNING: JobView = { output: { total: 0, earliest: 0 }, } -const TERMINAL_WITHOUT_FINISH: JobView = { - id: JobId('bash-1'), - kind: 'bash', - label: 'compile', - status: 'completed', - startedAt: 10, - output: { total: 0, earliest: 0 }, -} +const DONE: JobView = { ...RUNNING, status: 'completed', finishedAt: 20 } -async function setup(seed: JobView[] = []): Promise<(job: unknown) => void> { +/** + * A stub registry whose reads come from a map the test edits: the companion + * compares every announced projection against these reads. + */ +async function setup() { const ctx = new Context() + const reads = new Map() let listener: JobEventListener | undefined const probe = { - visibleTo: () => ({ list: () => seed }), + visibleTo: (caller?: SessionId) => ({ + get: (id: JobId) => { + const view = reads.get(String(id)) + if (view === undefined || view.owner !== caller) throw new Error(`unknown job ${String(id)}`) + return view + }, + }), events: { subscribe(_filter: unknown, value: JobEventListener) { listener = value @@ -52,54 +48,143 @@ async function setup(seed: JobView[] = []): Promise<(job: unknown) => void> { apply(child: Context) { child.provide('jobs', probe) }, }) await ctx.plugin(JobsInvariant) - if (listener === undefined) throw new Error('job invariant did not subscribe to settlements') - return (job) => { listener!({ type: 'settled', job: job as JobView, cause: 'producer' }) } + if (listener === undefined) throw new Error('job invariant did not subscribe to the event stream') + const emit = (event: JobEvent): void => { listener!(event) } + const read = (view: JobView | undefined): void => { + if (view === undefined) reads.delete(String(ID)) + else reads.set(String(view.id), view) + } + return { emit, read } } describe('job-registry invariants', () => { - it('accepts coherent current and terminal projections', async () => { - const notify = await setup([RUNNING]) - expect(() => { notify(BASE) }).not.toThrow() - expect(() => { notify({ ...BASE, id: JobId('subagent-2'), kind: 'subagent', owner: SessionId('owner') }) }) - .not.toThrow() - expect(() => { notify({ ...BASE, output: { total: 8, earliest: 4 } }) }).not.toThrow() + it('accepts one job announced from registration to removal, each event agreeing with the read', async () => { + const { emit, read } = await setup() + read(RUNNING) + emit({ type: 'registered', job: RUNNING }) + read({ ...RUNNING, progress: '1/2' }) + emit({ type: 'progress', job: { ...RUNNING, progress: '1/2' } }) + read({ ...RUNNING, progress: '1/2', output: { total: 4, earliest: 0 } }) + emit({ type: 'output', id: ID, total: 4 }) + read({ ...RUNNING, status: 'stopping', output: { total: 4, earliest: 0 } }) + emit({ type: 'stopping', job: { ...RUNNING, status: 'stopping', output: { total: 4, earliest: 0 } } }) + read({ ...DONE, status: 'killed', output: { total: 4, earliest: 0 } }) + emit({ type: 'settled', job: { ...DONE, status: 'killed', output: { total: 4, earliest: 0 } }, cause: 'kill' }) + emit({ type: 'output', id: ID, total: 4 }) + read(undefined) + expect(() => { emit({ type: 'removed', job: { ...DONE, status: 'killed', output: { total: 4, earliest: 0 } } }) }).not.toThrow() }) - it('ignores non-settlement events', async () => { - const ctx = new Context() - let listener: JobEventListener | undefined - const probe = { - visibleTo: () => ({ list: () => [] }), - events: { subscribe(_filter: unknown, value: JobEventListener) { listener = value; return () => {} } }, - } as unknown as JobRegistry - await ctx.plugin(InvariantRegistry) - await ctx.plugin({ name: 'job-invariant-probe', apply(child: Context) { child.provide('jobs', probe) } }) - await ctx.plugin(JobsInvariant) - expect(() => { listener!({ type: 'registered', job: { ...BASE, label: '' } }) }).not.toThrow() + it('adopts a job first seen through a later event, so a companion mounted late raises nothing', async () => { + const { emit, read } = await setup() + read({ ...RUNNING, progress: '3/4' }) + expect(() => { emit({ type: 'progress', job: { ...RUNNING, progress: '3/4' } }) }).not.toThrow() + read(DONE) + expect(() => { emit({ type: 'settled', job: DONE, cause: 'producer' }) }).not.toThrow() + read(undefined) + expect(() => { emit({ type: 'removed', job: DONE }) }).not.toThrow() + }) + + it('accepts an unowned job read through the unowned view and an owned job through its owner', async () => { + const { emit, read } = await setup() + const owner = SessionId('owner') + const owned: JobView = { ...RUNNING, id: JobId('subagent-2'), kind: 'subagent', owner } + read(RUNNING) + read(owned) + expect(() => { emit({ type: 'registered', job: RUNNING }) }).not.toThrow() + expect(() => { emit({ type: 'registered', job: owned }) }).not.toThrow() + expect(() => { emit({ type: 'output', id: owned.id, owner, total: 0 }) }).not.toThrow() }) it.each([ - [{ ...BASE, id: JobId('-1'), kind: '' }, /positive ordinal/], - [{ ...BASE, id: JobId('other-1') }, /must be "bash-" followed by a positive ordinal/], - [{ ...BASE, id: JobId('bash-x') }, /positive ordinal/], - [{ ...BASE, id: JobId('bash-0') }, /positive ordinal/], - [{ ...BASE, startedAt: -1 }, /startedAt must be a non-negative epoch integer/], - [{ ...BASE, startedAt: 0.5 }, /startedAt must be a non-negative epoch integer/], - [{ ...BASE, status: 'running' }, /finishedAt must be present exactly for a terminal status/], - [TERMINAL_WITHOUT_FINISH, /finishedAt must be present exactly for a terminal status/], - [{ ...BASE, finishedAt: 9 }, /no earlier than startedAt/], - [{ ...BASE, finishedAt: 20.5 }, /no earlier than startedAt/], - [{ ...BASE, progress: '3/10' }, /progress must be cleared once settled/], - [{ ...BASE, output: { total: 8, earliest: 9 } }, /0 <= earliest <= total/], - [{ ...BASE, output: { total: 8.5, earliest: 0 } }, /0 <= earliest <= total/], - [{ ...BASE, output: { total: 8, earliest: 0.5 } }, /0 <= earliest <= total/], - [{ ...BASE, output: { total: 8, earliest: -1 } }, /0 <= earliest <= total/], - ] as const)('rejects an incoherent registry projection', async (job, message) => { - const notify = await setup() - expect(() => { notify(job) }).toThrow(message) - }) - - it('rejects an incoherent projection already present at installation', async () => { - await expect(setup([{ ...BASE, label: '' }])).rejects.toThrow(/label must be non-empty/) - }) + ['registered after an earlier event for the same id', ({ emit, read }) => { + read(RUNNING) + emit({ type: 'output', id: ID, total: 0 }) + emit({ type: 'registered', job: RUNNING }) + }, /registered announced for job bash-1 after earlier events/], + ['registered twice', ({ emit, read }) => { + read(RUNNING) + emit({ type: 'registered', job: RUNNING }) + emit({ type: 'registered', job: RUNNING }) + }, /registered announced for job bash-1 after earlier events/], + ['registered with a terminal status', ({ emit, read }) => { + read(DONE) + emit({ type: 'registered', job: DONE }) + }, /must announce a live status without finishedAt/], + ['progress after settlement', ({ emit, read }) => { + read(DONE) + emit({ type: 'settled', job: DONE, cause: 'producer' }) + emit({ type: 'progress', job: { ...DONE, progress: 'late' } }) + }, /progress announced for job bash-1 after its settlement/], + ['stopping with a terminal status', ({ emit, read }) => { + read(DONE) + emit({ type: 'stopping', job: DONE }) + }, /stopping announced for job bash-1 with a terminal status/], + ['settled twice', ({ emit, read }) => { + read(DONE) + emit({ type: 'settled', job: DONE, cause: 'producer' }) + emit({ type: 'settled', job: DONE, cause: 'producer' }) + }, /settled announced twice for job bash-1/], + ['settled with a live status', ({ emit, read }) => { + read(RUNNING) + emit({ type: 'settled', job: { ...RUNNING, finishedAt: 20 }, cause: 'producer' }) + }, /must announce a terminal status, got "running"/], + ['settled without finishedAt', ({ emit, read }) => { + read(DONE) + emit({ type: 'settled', job: { ...RUNNING, status: 'completed' }, cause: 'producer' }) + }, /finishedAt no earlier than startedAt/], + ['settled before it started', ({ emit, read }) => { + read(DONE) + emit({ type: 'settled', job: { ...DONE, finishedAt: 9 }, cause: 'producer' }) + }, /finishedAt no earlier than startedAt/], + ['settled with a progress line', ({ emit, read }) => { + read(DONE) + emit({ type: 'settled', job: { ...DONE, progress: '9/10' }, cause: 'producer' }) + }, /must announce a cleared progress line/], + ['settled while the registry still reads it live', ({ emit, read }) => { + read(RUNNING) + emit({ type: 'settled', job: DONE, cause: 'producer' }) + }, /announces completed at 20 while the registry reads running at undefined/], + ['removed before settlement', ({ emit, read }) => { + read(RUNNING) + emit({ type: 'registered', job: RUNNING }) + read(undefined) + emit({ type: 'removed', job: RUNNING }) + }, /removed announced for job bash-1 before its settlement/], + ['removed while the registry still returns the job', ({ emit, read }) => { + read(DONE) + emit({ type: 'settled', job: DONE, cause: 'producer' }) + emit({ type: 'removed', job: DONE }) + }, /removed announced for job bash-1 that the registry still returns/], + ['output ahead of the read total', ({ emit, read }) => { + read(RUNNING) + emit({ type: 'output', id: ID, total: 8 }) + }, /output announced for job bash-1 at total 8 ahead of the registry's 0/], + ['output for a job the registry no longer returns', ({ emit }) => { + emit({ type: 'output', id: ID, total: 0 }) + }, /output announced for job bash-1 that the registry no longer returns/], + ['a lifecycle event for a job the registry does not return', ({ emit }) => { + emit({ type: 'registered', job: RUNNING }) + }, /registered announced for job bash-1 that the registry does not return/], + ['an announced label the registry does not read', ({ emit, read }) => { + read(RUNNING) + emit({ type: 'registered', job: { ...RUNNING, label: 'link' } }) + }, /announces label "link" while the registry reads "compile"/], + ['an announced owner the registry does not read', ({ emit, read }) => { + read({ ...RUNNING, owner: SessionId('alice') }) + emit({ type: 'registered', job: { ...RUNNING, owner: SessionId('alice') } }) + read({ ...RUNNING, owner: SessionId('alice'), startedAt: 11 }) + emit({ type: 'progress', job: { ...RUNNING, owner: SessionId('alice'), progress: 'x' } }) + }, /announces startedAt 10 while the registry reads 11/], + ['an announced output total ahead of the read', ({ emit, read }) => { + read(RUNNING) + emit({ type: 'registered', job: { ...RUNNING, output: { total: 2, earliest: 0 } } }) + }, /announces output total 2 ahead of the registry's 0/], + ] as const satisfies readonly (readonly [string, (probe: Awaited>) => void, RegExp])[])( + 'rejects %s', + async (_name, script, message) => { + const probe = await setup() + expect(() => { script(probe) }).toThrow(message) + }, + ) })