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.

.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

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

.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'`，从不导入提供方类型。
 

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

packages/api/job-controller/README.md

@@ -64,4 +64,4 @@ None.
 
 </details>
 
-**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.

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

@@ -64,4 +64,4 @@ Client 入口在 `ClientJobsModel` 之上提供 `ClientJobs`（`ctx.jobs`）。`
 
 </details>
 
-**运行时不变式：** 不发布伴生入口。控制器是 `ctx.jobs` 读取的无状态投影；这些流转发的偏移关系由注册表自己的 `@deepseek-ai/dsh-jobs/invariant` 拥有。
+**运行时不变式：** 不发布伴生入口。控制器是 `ctx.jobs` 读取的无状态投影；这些流转发的事件协议与事件对读取的关系由注册表自己的 `@deepseek-ai/dsh-jobs/invariant` 拥有。

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

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
 

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）以及每个通告的投影与注册表自身读取的一致性 |
 
 ### 服务操作
 

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<string, Phase>()
+  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'] })
 

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<string, JobView>()
   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<ReturnType<typeof setup>>) => void, RegExp])[])(
+    'rejects %s',
+    async (_name, script, message) => {
+      const probe = await setup()
+      expect(() => { script(probe) }).toThrow(message)
+    },
+  )
 })