diff --git a/AGENTS.md b/AGENTS.md index 723a781..c246d39 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,7 +41,7 @@ Guidance for any agent (or human) working in this repo. `CLAUDE.md` is a symlink 2. **Fixture-semantics** — *"does this fixture actually test what it claims?"* See the **Fixture quality bar** in `fixtures/asyncapi/README.md` (no vacuous values; claim↔content; full-path/both-direction coverage; internal consistency; negative cases). The validity-only angle misses all of these. ## Status & next -All build tiers and v1 gates are `tested` (37 of 39 requirements): `model` → `broker`/`registry`/`ingestion` → `engine`/`validation` → `scenarios`/`control-plane` → `cli` (the full verb set incl. `up`/`down` process management over the G14 runfile, watch modes, `init`), plus the four cross-cutting v1 gates (R-028–R-031). The open items: the two empirical spikes, `R-006` (WS-fidelity) and `R-007` (capture the browser application's `connect()`), which need the real browser application against `broker/`'s Aedes defaults and are hard gates on calling v1 done; plus `R-033` (`tested`): the `demo-app/` spike-harness webapp + connect fingerprint (`docs/specs/demo-app.md`), which rehearses both spikes — the at-work capture is now a no-app-change procedure (point the real client at offbook, read `offbook logs`). The adoption surface is `tested` (R-034–R-036: README + guides with executable quickstart/cookbook gates, `offbook doctor`, the first-run error audit — docs/specs/adoption.md). The AsyncAPI support range is declared and hardened (R-037–R-039, D-018): **2.0.0–2.6.0, 3.0.0, 3.1.0**, payloads validated under **draft-07**, with the R-028 gate extended over `multi-format.yaml` (3.1.0) and `v2-oldest.yaml` (2.0.0). +All build tiers and v1 gates are `tested` (38 of 40 requirements): `model` → `broker`/`registry`/`ingestion` → `engine`/`validation` → `scenarios`/`control-plane` → `cli` (the full verb set incl. `up`/`down` process management over the G14 runfile, watch modes, `init`), plus the four cross-cutting v1 gates (R-028–R-031). The open items: the two empirical spikes, `R-006` (WS-fidelity) and `R-007` (capture the browser application's `connect()`), which need the real browser application against `broker/`'s Aedes defaults and are hard gates on calling v1 done; plus `R-033` (`tested`): the `demo-app/` spike-harness webapp + connect fingerprint (`docs/specs/demo-app.md`), which rehearses both spikes — the at-work capture is now a no-app-change procedure (point the real client at offbook, read `offbook logs`). The adoption surface is `tested` (R-034–R-036: README + guides with executable quickstart/cookbook gates, `offbook doctor`, the first-run error audit — docs/specs/adoption.md). The AsyncAPI support range is declared and hardened (R-037–R-039, D-018): **2.0.0–2.6.0, 3.0.0, 3.1.0**, payloads validated under **draft-07**, with the R-028 gate extended over `multi-format.yaml` (3.1.0) and `v2-oldest.yaml` (2.0.0). The per-channel initial-state opt-out is `tested` (R-040, D-025): `topicOverrides.
.initialState: false` declares a reactive-only channel (no L1 floor; four spec-load warnings; handler-wins warn-log; `GET /v1/topics` marks suppressed channels). ## Working notes - **Git identity is the user's to set** — don't run `git config user.*` on their behalf. Commit/push **only when asked**. diff --git a/DECISIONS.md b/DECISIONS.md index 3cf9b2c..eb98a13 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -235,3 +235,13 @@ Append-only. Each decision has a stable never-reused `D-###` id, what was decide **Obligations**: none. **From**: the "matters of aedes" follow-up to D-021 (2026-08-01): source reads of `aedes-server-factory@0.2.1`, tarball reads of `aedes@1.1.1` and `aedes-persistence@10.3.1`, and the Bun probe described above. **Folds into**: package.json, bun.lock, src/broker/index.ts, src/broker/fingerprint.test.ts + +### D-025: Per-channel initial-state opt-out — `topicOverrides.
.initialState: false` +**Date**: 2026-08-01 +**What**: `ServiceConfig.topicOverrides` values grow `initialState?: boolean` (absent ⇒ true; only `false` is meaningful). The registry resolves it onto `Channel.initialState` (toClient records only — the floor never runs elsewhere; no spec-binding tier — the override is the field's only author), and the engine's proactive floor (`materializeAndPublish`) returns before the L1 draw when `initialState === false`, silencing every materialization leg (eager startup, concrete subscribe, `seedInstances`, `reset` republish) while the instance ledger, L2/L3 emissions, wildcard retained replay, and the explicit example surfaces (`GET /v1/topics` examples, `POST /v1/publish {example:true}`) stay untouched. An L3 `initialState` handler still wins, with a compose-root warn-log naming channel + handler, re-run after `POST /v1/specs/refresh`. Four `spec-load` warnings make misconfiguration loud: `override-dangling-key`, `initial-state-on-from-client` (address-scoped — a dual-direction address must not warn), `initial-state-non-boolean` (warn + ignore), `initial-state-cross-service` (exact-address duplicates at the merge seam, naming the true match winner — a same-address fromClient record can win, in which case the floor never runs there and every declaration is dead). `TopicInfo.initialState?: false` appears only when suppressed; the CLI topics views carry a `[no initial state]` marker. +**Why**: Reactive-only channels (error/notification topics) have no initial state; the always-on floor emits a synthetic draw on subscribe that can drive a stateful client into a bad state — behavior a real broker (silent on subscribe for non-retained topics) would never produce. The opt-out keeps the zero-config floor as the default: a retain-keyed "faithful" default was rejected because `retain` resolves `false` at the bottom of the §2 chain, so it would silence nearly every channel out of the box. Adjacent prior art: D-009 declined a tick-varying L1 draw partly because churn "would immediately demand a quiet-toggle" — this is that toggle, per channel, for the subscribe-leg floor D-009 ratified. +**Mitigations / notes**: Retained residue is deliberately out of the flag's reach: an L2/L3/`/publish` retained payload on a flagged channel survives `reset` un-overwritten (contracts §2/§5 caveats; the wiring guide recommends `retain: false` there). The flag is boot-time-only like all of `topicOverrides` (`specs update` re-resolves from the boot-time ServiceConfig and never re-reads services.yaml; changing the flag takes a restart). The F21 compiled-registry cache key omits ServiceConfig — safe while services.yaml is read once per process; the invariant is commented at the cache site (`src/cli/boot.ts`). Parametrized cross-service shadowing (a literal address in one service shadowing a flagged `{param}` address in another) stays a known residual — the merge warning covers exact duplicates only. +**Consequences for earlier entries**: none changed — D-008 (drop-and-surface) and D-009 (no tick leg) stand untouched; this narrows where the floor runs, not how it draws or fails. +**Obligations**: if services.yaml ever becomes re-readable mid-process, grow the F21 cache key with a ServiceConfig fingerprint (see the `src/cli/boot.ts` comment). +**From**: docs/archive/intake/2026-08-01-initial-state-optout.md (design dialog + 4-lens adversarial review, 2026-08-01). +**Folds into**: docs/specs/contracts.md §1/§2/§5/§6, docs/specs/design.md §7a, docs/guides/wiring-your-service.md, REQUIREMENTS.md (R-040), src/model/index.ts, src/config/fixtures/services.yaml, src/registry/index.ts, src/engine/index.ts, src/compose/index.ts, src/control-plane/index.ts, src/cli/index.ts, src/cli/boot.ts diff --git a/REQUIREMENTS.md b/REQUIREMENTS.md index c884ca1..c96c148 100644 --- a/REQUIREMENTS.md +++ b/REQUIREMENTS.md @@ -318,6 +318,14 @@ Every error reachable on the clone→demo→init→wire→up→first-publish pat **TEST**: src/registry/index.test.ts, test/gate-validation.test.ts, test/upstream-drift.test.ts `registry/` guards binding-supplied `qos`/`retain` values (falling through the §2 precedence chain on a bad value), reports unknown keys against a hand-authored mqtt operation-binding key set that honors the schema's `x-` vendor-extension pattern and is drift-tested against `@asyncapi/specs` (a devDependency, never imported from `src/`; D-019), reports an mqtt CHANNEL binding as ignored, and reports MQTT-5-only binding fields as unhonored under the MQTT 3.1.1-only constraint. +#### Per-channel initial-state opt-out (reactive-only channels) +**UID**: R-040 +**STATUS**: tested +**COVERS**: docs/specs/contracts.md#R-040 +**IMPL**: src/model/index.ts, src/registry/index.ts, src/engine/index.ts, src/compose/index.ts, src/control-plane/index.ts, src/cli/index.ts +**TEST**: src/config/index.test.ts, src/registry/index.test.ts, src/engine/index.test.ts, src/compose/initial-state.test.ts, src/control-plane/index.test.ts, src/cli/doctor.test.ts, test/cli-dispatch.test.ts +`topicOverrides.
.initialState: false` (services.yaml) declares a reactive-only channel: the registry resolves the flag onto `Channel.initialState` (toClient records only; no spec-binding tier; only `false` is meaningful), the engine's L1 proactive floor skips the channel on every materialization leg (concrete subscribe, eager startup, `seedInstances`, `reset` republish) while the instance ledger, L2/L3 emissions, wildcard retained replay, and the explicit example surfaces stay untouched; an L3 `initialState` handler still wins, with a compose-root warn-log naming channel and handler re-run after a specs refresh; four `spec-load` warnings (`override-dangling-key`, `initial-state-on-from-client`, `initial-state-non-boolean`, `initial-state-cross-service`) make misconfiguration loud; `GET /v1/topics` exposes `initialState: false` on suppressed channels only. + ` markers inside frozen contracts). + +**Files:** +- Modify: `docs/specs/contracts.md` (one comment line above the §2 G3 bullet, ~line 127) +- Modify: `REQUIREMENTS.md` (insert entry after R-039, before the trailing ` +- **`onSubscribe` & the initial-state materialization policy (G3).** Retained initial state +``` + +- [ ] **Step 2: Insert the R-040 entry in REQUIREMENTS.md** + +Directly after R-039's statement line (the long sentence starting `` `registry/` guards binding-supplied `qos`/`retain` values ``) and before the ` - **`onSubscribe` & the initial-state materialization policy (G3).** Retained initial state for `toClient` channels is published by the **engine**, which owns materialization end-to-end: it consumes `broker.onSubscribe` and, on a **concrete** subscribe, calls `InstanceRegistry.materialize` then republishes — the broker only *reports* the subscribe, it never materializes (F6). **When** the publish happens depends on whether the channel address is parametrized — these are the **normative rules** (`design.md` §7a elaborates them with examples + rationale): - **Non-parametrized** `toClient` channels → published **eagerly at startup** (one concrete topic, nothing to de-wildcard). - - **Parametrized** `toClient` channels → an instance is **materialized lazily** when a concrete subscribe binds its params **or** a `fromClient` command first references a concrete param; the engine keeps a **materialized-instance set** — the engine-owned `InstanceRegistry` (F1), the single owner of all five rules in this policy. + - **Parametrized** `toClient` channels → an instance is **materialized lazily** when a concrete subscribe binds its params **or** a `fromClient` command first references a concrete param; the engine keeps a **materialized-instance set** — the engine-owned `InstanceRegistry` (F1), the single owner of all six rules in this policy. - A **wildcard subscribe** (`+`/`#`) replays the **existing retained state for every topic matching the filter** — sourced from **Aedes' own retained store** (R3, the single source of truth) via the broker's **native retained delivery** to the subscribing client (filter tested by `matchesFilter`, F6), **not** a parallel materialized-instance set. It **never invents** params: a topic is replayed iff it currently holds retained state, so a cleared (zero-byte-evicted) topic is excluded and an off-ledger L3/L2 retained publish is included — strictly more correct than a ledger could be. - Optional **`seedInstances`** (typed on `ServiceConfig`, §6 — channel address → list of param-maps) pre-materializes a deterministic demo set at startup (so onboarding isn't a blank UI). - - **`reset`** re-materializes via `InstanceRegistry.restore(snapshot())` — **exactly the recorded set** (seed instances + those materialized since the last reset), re-seeded — so post-`reset` `/state` is deterministic **by construction**, not empty. + - **`initialState: false`** — a `topicOverrides` declaration (§6) marking a **reactive-only** channel (error/notification topics: nothing to materialize): the engine still records instances per the rules above but **never publishes the L1 floor** on any leg of this policy (eager startup, concrete subscribe, `seedInstances`, `reset` republish). An L3 `initialState` handler still runs (most-specific wins), with the contradiction warn-logged. The flag gates **engine emissions only** — it neither blocks Aedes' native retained delivery nor scrubs retained residue (R-040/D-025). + - **`reset`** re-materializes via `InstanceRegistry.restore(snapshot())` — **exactly the recorded set** (seed instances + those materialized since the last reset), re-seeded — so post-`reset` `/state` is deterministic **by construction**, not empty. (On a channel with `initialState: false` the republish is skipped, so retained residue there — an L2/L3/`/publish` retained payload — persists as-is across `reset`; R-040.) ```ts // Engine-owned instance lifecycle (F1) — the ONE owner of the materialization policy above; declared in model/, driven by the engine. @@ -273,7 +276,7 @@ interface Violation { | Endpoint | Returns | Notes | |---|---|---| | `GET /v1/topics` | `{ topics: TopicInfo[] }` — or `{ topics: Omit[] }` under `?schema=false` | dereferenced **schema + seeded example inline** (via the injected `Faker`, F11); `?direction=` / `?service=` filters; **`?schema=false`** is the slim discovery view — drops the bulky `schema` field **only** (keeps `example` + all else), so `schema` stays **required** on the full `TopicInfo` | -| `GET /v1/state` | `{ state: StateEntry[] }` | lean, **concrete** topics; `?topic=` prefix filter | +| `GET /v1/state` | `{ state: StateEntry[] }` | lean, **concrete** topics; `?topic=` prefix filter; reports Aedes' store as-is — on an `initialState: false` channel (§2) retained residue can persist across `reset` (R-040) | | `GET /v1/validation` | `{ violations: Violation[]; summary: ValidationSummary }` | `?sinceSeq=` (strictly-greater) `?origin=` `?severity=` `?kind=`; ordered by `seq` alone (now a total order — G6); violations-only. **Bounded ring buffer** — see note below | | `GET /v1/specs` | `{ specs: SpecInfo[]; resolutionMode; warnings? }` | `resolutionMode: 'branch' \| 'pinned'` honesty flag (design §7); in **branch** mode `warnings?` carries the version-not-honored notice — "requested versions in `environments.yaml` are recorded but NOT honored; fetching branch tips (serviceA→main, serviceB→dev)", naming each service's actual branch; suppressed under `pinned`/`--frozen` (EQ2) — **both v2; v1's `resolutionMode` is always `'branch'`, so the notice always shows**. Each `SpecInfo` also carries `source` + `fetchedAt` — the **content-axis** trust surface (design §7, Mode 3): validation is against the spec **as fetched**, so age shows **neutrally** (no stale threshold) for the dev to weigh; `offbook status` composes this | | `GET /v1/diagnostics` | `{ diagnostics: Diagnostic[]; summary: DiagnosticSummary }` | load/hot-reload-populated; dev-time surface | @@ -283,7 +286,8 @@ interface Violation { ```ts interface TopicInfo { topic: string; direction: Direction; service: string; - title?: string; description?: string; schema: object; example?: unknown; qos?: 0|1|2; retain?: boolean; } + title?: string; description?: string; schema: object; example?: unknown; qos?: 0|1|2; retain?: boolean; + initialState?: false; } // present ONLY when the channel declares initialState: false (§2/§6, R-040) — absent otherwise; survives ?schema=false (that view drops `schema` alone) interface StateEntry { topic: string; payload: unknown; qos?: 0|1|2; retain: true; } // retain is always true — clearing a retained topic EVICTS it (§2), so /state never returns tombstones; a decode-failure (§2) is never stored, so payload is always a successfully-decoded value interface SpecInfo { service: string; declaredVersion?: string; specVersion?: string; source: string; contentHash: string; channelCount: number; fetchedAt: string; } // declaredVersion = info.version, read parser-free by ingestion/ (shallow yaml read, G12) — NOT the requested version (they differ in v1 branch mode). specVersion = the AsyncAPI DOCUMENT version (the `asyncapi` field, e.g. '3.1.0'), read in the same parser-free pass — which spec major this service is on (D-018), not the service's own info.version. fetchedAt (ISO8601, propagated from the lockfile `fetched-at`) = spec provenance/age for TRUST CALIBRATION — the tool validates against the spec AS FETCHED, never the live service; surfaced NEUTRALLY (no stale threshold) by GET /specs + status (design §7, Mode 3) interface ScenarioInfo { name: string; when?: string; stepCount: number; source: string; } // GET /scenarios discovery (P8): `when` = the reactive trigger topic (absent ⇒ on-demand/trigger-only); source = scenario file path @@ -304,7 +308,9 @@ interface Diagnostic { kind: 'scenario-load' | 'overlap' | 'spec-load' | 'uninst // catalog build can see, so they cannot be recomputed from a Channel. The kind union stays closed (four values, and // DiagnosticSummary.byKind keeps exactly those four keys, zero-filled); each finding is instead machine-identified by // a stable tag prefix on `detail` (the tag, then `: `, then the sentence): 'binding-on-channel', -// 'binding-invalid-value', 'binding-unknown-key', 'mqtt5-field-ignored', 'dialect-mismatch', 'schema-compile-failed'. +// 'binding-invalid-value', 'binding-unknown-key', 'mqtt5-field-ignored', 'dialect-mismatch', 'schema-compile-failed', +// 'override-dangling-key', 'initial-state-on-from-client', 'initial-state-non-boolean', +// 'initial-state-cross-service' (the last four: the R-040 topicOverrides sweep + the cross-service merge check). // Channel ADDRESS in `source?` as above, so filtering by tag and by address both work. interface ValidationSummary { // the CI-facing payload of GET /v1/validation @@ -360,7 +366,7 @@ interface ServiceConfig { branch?: string; // v1 ref selection; default 'main' qosDefault?: 0 | 1 | 2; // per-service default qos — tier 3 of the §2 precedence chain (above the global qos 1 fallback; the last config tier, consulted just before global) (G13) retainDefault?: boolean; // per-service default retain — tier 3 - topicOverrides?: Record; // per-topic override — tier 2 (above the per-service default, below the spec binding); key = channel address, matched by STRING-EQUALITY against channel.topic (the {param} form) — not routed through SpecRegistry.match, not concrete topics (F14) + topicOverrides?: Record; // per-topic override — qos/retain are tier 2 (above the per-service default, below the spec binding); initialState rides the SAME map but has NO other tier (no binding above, no service default below; only `false` is meaningful — §2, R-040/D-025); key = channel address, matched by STRING-EQUALITY against channel.topic (the {param} form) — not routed through SpecRegistry.match, not concrete topics (F14) seedInstances?: Record[]>; // channel address → list of param-maps; pre-materializes a deterministic demo set at startup (F1; §2 InstanceRegistry). Each map binds ALL of a channel's {params}, so multi-param channels work // v2: versionToSha strategy, specPath glob strategy, range policy, manual override } diff --git a/docs/specs/design.md b/docs/specs/design.md index b3ef1c0..7ce0427 100644 --- a/docs/specs/design.md +++ b/docs/specs/design.md @@ -199,11 +199,12 @@ Ordering control, QoS1 **duplicates** (at-least-once genuinely means the browser ### Liveliness, cleanly separated Two behaviors that were initially conflated, now decided independently: -- **(a · §7a) Initial state on connect — always on.** The engine guarantees retained state exists for any **materialized** instance, so the UI renders populated immediately (no blank UI until the next tick). **Whether** that state pre-exists or is created on demand splits by whether the `toClient` channel address is parametrized — **eager** for non-parametrized channels, **lazy** (created at first concrete subscribe/command) for parametrized ones. L1/L3 publish initial state with `retain: true`; L1's initial-state faking is **per-instance** — keyed by the instance's params (`fake(channel, params)`), so a multi-device `seedInstances` set renders **distinct** devices, not N identical ones (contracts §3, CR1). This is the materialization policy, owned by the engine-side `InstanceRegistry` (the **normative rules + types are in `contracts.md` §2** — this elaborates them with examples + rationale): +- **(a · §7a) Initial state on connect — always on (with a per-channel opt-out, below).** The engine guarantees retained state exists for any **materialized** instance, so the UI renders populated immediately (no blank UI until the next tick). **Whether** that state pre-exists or is created on demand splits by whether the `toClient` channel address is parametrized — **eager** for non-parametrized channels, **lazy** (created at first concrete subscribe/command) for parametrized ones. L1/L3 publish initial state with `retain: true`; L1's initial-state faking is **per-instance** — keyed by the instance's params (`fake(channel, params)`), so a multi-device `seedInstances` set renders **distinct** devices, not N identical ones (contracts §3, CR1). This is the materialization policy, owned by the engine-side `InstanceRegistry` (the **normative rules + types are in `contracts.md` §2** — this elaborates them with examples + rationale): - **Non-parametrized `toClient` channels** (e.g. `status/all`): retained initial state is published **eagerly at startup**. There is exactly one concrete topic, so there is nothing to de-wildcard. - **Parametrized `toClient` channels** (e.g. `state/{deviceId}`): a concrete instance has no value until one is bound — the spec declares the param, never enumerates ids — so an instance is **materialized lazily** when either (i) a **concrete** subscribe binds its params (`SUBSCRIBE state/thermostat-1`), or (ii) a `fromClient` command **first references** a concrete param (`command/thermostat-1/set` → the reactive path publishes `state/thermostat-1`). The engine keeps a **materialized-instance set** in the engine-owned `InstanceRegistry` (`contracts.md` §2) — the recorded concrete ids. - **Wildcard subscribe** (`SUBSCRIBE state/+` or `state/#`): emit the **existing retained state for every topic matching the filter** — via Aedes' **native retained delivery** from its own retained store (filter tested by `matchesFilter`, F6; R3 — the subscribe hot-path does **not** call `getState()`), **not** a parallel materialized-instance ledger — and **never invent** a param value. A wildcard carries no binding to de-wildcard, so it can only replay what already exists (a cleared topic is excluded; an off-ledger L3/L2 retained publish is included). *(MQTT `+`/`#` are subscribe-side filters here, distinct from the channel-address `{param}` the registry matcher resolves — `contracts.md` §1.)* - **Optional `seedInstances`** — a `ServiceConfig` field (`contracts.md` §6) keyed by channel address, whose value is a **list of param-maps** so multi-param channels work (e.g. `state/{deviceId}: [{ deviceId: thermostat-1 }, { deviceId: thermostat-2 }]`) — pre-materializes a deterministic demo set at startup, so onboarding isn't a blank UI even before any subscribe or command. + - **Reactive-only opt-out (`initialState: false`, R-040/D-025)**: error/notification-style `toClient` channels carry no initial state — a synthetic draw there can corrupt a stateful client, and against a real broker a subscriber to a non-retained topic hears nothing until the service publishes. "Always on" deliberately trades that fidelity for a populated first render; `topicOverrides.
.initialState: false` (`contracts.md` §6) restores per-channel silence where the trade is wrong. The materialization ledger and L2/L3 behavior are untouched; an L3 `initialState` handler still wins, warn-logged. - **`reset`** re-materializes via `InstanceRegistry.restore(snapshot())` — **exactly the recorded set** (any `seedInstances` plus instances materialized since the last reset), re-seeded — so post-`reset` `/state` is deterministic **by construction**, not empty (the moment-4 CI primitive holds). - **(b · §7b) Autonomous emission over time — a toggleable, seeded mode.** Whether the mock keeps generating new messages on its own (e.g. telemetry ticking) is a **mode**, on by default for normal startup (onboarding/daily-driver benefit) and off/forced-off under test (to avoid CI flake). When on, emission is **seeded** for run-to-run reproducibility. diff --git a/src/cli/boot.ts b/src/cli/boot.ts index 8fd2b54..bedd3fb 100644 --- a/src/cli/boot.ts +++ b/src/cli/boot.ts @@ -39,7 +39,11 @@ export async function bootProject(opts: ProjectBootOptions): Promise { : {}; const resolver = new GitRefResolver({ gitHost }); - // per-service compiled registries keyed by content-hash (the F21 skip) + // per-service compiled registries keyed by content-hash (the F21 skip). + // R-040 invariant: the key deliberately omits ServiceConfig — safe because + // `services` is read once at boot and immutable in-process; if services.yaml + // ever becomes re-readable mid-process, this key must grow a config + // fingerprint or a stale Channel.initialState is served for an unchanged spec. const compiled = new Map(); async function resolveAll(): Promise<{ diff --git a/src/cli/doctor.test.ts b/src/cli/doctor.test.ts index 5ace861..d1f78c6 100644 --- a/src/cli/doctor.test.ts +++ b/src/cli/doctor.test.ts @@ -141,6 +141,18 @@ test("project: valid services.yaml + environments.yaml passes", async () => { expect(byName(report, "project").status).toBe("pass"); }); +// [utest->R-040] +test("project: a services.yaml carrying topicOverrides.initialState parses clean", async () => { + const dir = projectWith({ + "services.yaml": + "services:\n svc:\n repo: org/svc\n specPath: asyncapi.yaml\n topicOverrides:\n 'errors/{id}': { initialState: false }\n", + }); + const report = await runDoctor( + ctxWith({ repoRoot: GOOD_REPO_ROOT, projectDir: dir }), + ); + expect(byName(report, "project").status).toBe("pass"); +}); + test("specs-reachable: --offline and empty services both warn, never fetch", async () => { const dir = projectWith({ "services.yaml": "services: {}\n" }); const offline = await runDoctor( diff --git a/src/cli/index.ts b/src/cli/index.ts index 476cb0a..b522826 100644 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -224,13 +224,22 @@ export function renderTopicList( topics: TopicInfo[], opts: TopicRenderOpts = {}, ): string { + // R-040: the human views carry the reactive-only marker; --json shows the + // TopicInfo field itself + const quietMark = (t: TopicInfo) => + t.initialState === false ? " [no initial state]" : ""; if (opts.compact) return topics - .map((t) => `${t.topic} [${phraseDirection(t.direction)}] ${t.service}`) + .map( + (t) => + `${t.topic} [${phraseDirection(t.direction)}] ${t.service}${quietMark(t)}`, + ) .join("\n"); return topics .map((t) => { - const lines = [`${t.topic} [${phraseDirection(t.direction)}]`]; + const lines = [ + `${t.topic} [${phraseDirection(t.direction)}]${quietMark(t)}`, + ]; lines.push(...fieldLines(t.schema)); if (opts.examples !== false && t.example !== undefined) lines.push(` example: ${JSON.stringify(t.example)}`); diff --git a/src/compose/index.ts b/src/compose/index.ts index 1b60aa9..e29fb38 100644 --- a/src/compose/index.ts +++ b/src/compose/index.ts @@ -69,6 +69,28 @@ export async function compose(parts: ComposeParts) { log, }); + // R-040: config says "no initial state", the handler says otherwise — the + // handler wins (L3 stays most-specific on every path); surface the + // contradiction, never silently prefer either side. Pure over + // (loaded handlers × current registry): re-run after every registry swap. + const warnInitialStateContradictions = () => { + for (const h of engine.handlers()) { + if (!h.hasInitialState) continue; + const flagged = registry + .channels() + .some( + (c) => + c.topic === h.pattern && + c.direction === "toClient" && + c.initialState === false, + ); + if (flagged) + log( + `channel '${h.pattern}' has initialState: false but handler '${h.modulePath}' defines initialState — the handler wins`, + ); + } + }; + // the ONE inbound pipeline (G9): classification (validation, never // blocking) + reactive dispatch — shared verbatim by real broker clients // and HTTP-injected publishes @@ -206,6 +228,7 @@ export async function compose(parts: ComposeParts) { const next = await parts.resolveSpecs(); registry = next.registry; // hot-swap; F19 lazy dispatch survives it specs = next.specs; + warnInitialStateContradictions(); // R-040: the flag set may have changed return specs; }, @@ -258,6 +281,7 @@ export async function compose(parts: ComposeParts) { await broker.start(); if (parts.handlersDir !== undefined) await engine.loadHandlers(parts.handlersDir); + warnInitialStateContradictions(); // strict mode: a scenario-load error aborts startup in the // foreground (l2 §7) — the throw propagates to the caller await runtime?.load(); diff --git a/src/compose/initial-state.test.ts b/src/compose/initial-state.test.ts new file mode 100644 index 0000000..35252d2 --- /dev/null +++ b/src/compose/initial-state.test.ts @@ -0,0 +1,88 @@ +// R-040 — the compose-root contradiction warn-log: an L3 initialState handler +// on an initialState:false channel wins, loudly; re-checked after a specs +// refresh swaps the registry. +// [utest->R-040] +import { afterEach, expect, test } from "bun:test"; +import { mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import Ajv2020 from "ajv/dist/2020"; +import { type Composed, compose } from "#src/compose/index.ts"; +import { loadConfig } from "#src/config/index.ts"; +import type { Channel, SpecRegistry } from "#src/model/index.ts"; + +const servers: Composed[] = []; +afterEach(async () => { + while (servers.length) await servers.pop()?.stop(); +}); + +function chan(topic: string, initialState?: boolean): Channel { + const v = new Ajv2020({ allErrors: true, strict: false }).compile({ + type: "object", + }); + return { + topic, + direction: "toClient", + service: "t", + schema: { type: "object" }, + validate: (p) => (v(p) ? [] : (v.errors ?? [])), + qos: 1, + retain: false, + initialState, + }; +} + +function regOf(...channels: Channel[]): SpecRegistry { + return { + diagnostics: () => [], + channels: () => channels, + match: (topic) => { + const c = channels.find((ch) => ch.topic === topic); + return c ? { channel: c, params: {} } : undefined; + }, + matchesFilter: () => false, + }; +} + +test("contradiction warn-log: fires for a flagged channel's initialState handler, stays silent otherwise, re-checks on refresh", async () => { + const dir = mkdtempSync(join(tmpdir(), "offbook-r040-handlers-")); + const dispatchPath = new URL("../engine/dispatch.ts", import.meta.url) + .pathname; + writeFileSync( + join(dir, "10-quiet.ts"), + [ + `import { register } from "${dispatchPath}";`, + `register("alerts/off", () => ({ initialState() {} }));`, + `register("alerts/on", () => ({ initialState() {} }));`, + "", + ].join("\n"), + ); + const logs: string[] = []; + const server = await compose({ + config: loadConfig({ + brokerWsPort: 18120, + brokerTcpPort: 12920, + controlPlanePort: 18920, + }), + registry: regOf(chan("alerts/off"), chan("alerts/on")), // unflagged at boot + handlersDir: dir, + resolveSpecs: async () => ({ + registry: regOf(chan("alerts/off", false), chan("alerts/on")), + specs: [], + }), + log: (l) => logs.push(l), + }); + servers.push(server); + await server.start(); + // boot registry is unflagged: no contradiction line + expect(logs.filter((l) => l.includes("the handler wins"))).toEqual([]); + // refresh swaps in the flagged registry: the re-check fires exactly once, + // naming the flagged channel and the handler file — never the unflagged one + await server.app.request("/v1/specs/refresh", { method: "POST" }); + const lines = logs.filter((l) => l.includes("the handler wins")); + expect(lines.length).toBe(1); + expect(lines[0]).toContain("'alerts/off'"); + expect(lines[0]).toContain("10-quiet.ts"); + expect(lines[0]).toContain("initialState: false"); + expect(lines.some((l) => l.includes("'alerts/on'"))).toBe(false); +}); diff --git a/src/config/fixtures/services.yaml b/src/config/fixtures/services.yaml index 223a475..edb8deb 100644 --- a/src/config/fixtures/services.yaml +++ b/src/config/fixtures/services.yaml @@ -12,3 +12,8 @@ services: topicOverrides: # tier 2: a per-topic override beats the per-service default telemetry/{deviceId}: { qos: 0, retain: false } serviceD: { repo: https://other.example.com/org/service-d.git, specPath: asyncapi.yaml } # full URL used as-is (ignores gitHost) + serviceE: # reactive-only declaration (R-040) + repo: org/service-e + specPath: asyncapi.yaml + topicOverrides: + alerts/{deviceId}: { initialState: false } # the initial-state floor is off on this channel diff --git a/src/config/index.test.ts b/src/config/index.test.ts index e594c71..371bcf1 100644 --- a/src/config/index.test.ts +++ b/src/config/index.test.ts @@ -136,3 +136,11 @@ test("loadEnvironments reads an environments.yaml file into typed objects", asyn serviceB: "2.0.1", }); }); + +// [utest->R-040] +test("loadServices carries topicOverrides.initialState through as a typed boolean (serviceE)", async () => { + const cfg = await loadServices(`${import.meta.dir}/fixtures/services.yaml`); + expect(cfg.services.serviceE?.topicOverrides).toEqual({ + "alerts/{deviceId}": { initialState: false }, + }); +}); diff --git a/src/control-plane/index.test.ts b/src/control-plane/index.test.ts index 6ed769e..08fdfca 100644 --- a/src/control-plane/index.test.ts +++ b/src/control-plane/index.test.ts @@ -583,3 +583,54 @@ test("POST /v1/specs/refresh: hot-swaps the running registry through the thunk ( }; expect(specs.specs).toEqual(newSpecs); }); + +// [itest->R-040] +test("GET /v1/topics: initialState:false is exposed only on suppressed channels and survives ?schema=false", async () => { + const flagged = { + ...makeChannel("quiet/errors", "toClient", { type: "object" }), + initialState: false, + }; + const normal = makeChannel("loud/state", "toClient", { type: "object" }); + const { req } = await boot(20, { + registry: fakeRegistry([flagged, normal]), + scenarios: false, + }); + const full = (await (await req("/v1/topics")).json()) as { + topics: Array>; + }; + const quiet = full.topics.find((t) => t.topic === "quiet/errors"); + const loud = full.topics.find((t) => t.topic === "loud/state"); + expect(quiet?.initialState).toBe(false); + expect(loud !== undefined && "initialState" in loud).toBe(false); + const slim = (await (await req("/v1/topics?schema=false")).json()) as { + topics: Array>; + }; + expect( + slim.topics.find((t) => t.topic === "quiet/errors")?.initialState, + ).toBe(false); +}); + +// [itest->R-040] +test("retained residue on an initialState:false channel survives reset and stays in /state", async () => { + const flagged = { + ...makeChannel("quiet/errors", "toClient", { type: "object" }), + initialState: false, + }; + const { server, req, post } = await boot(21, { + registry: fakeRegistry([flagged]), + scenarios: false, + }); + await post("/v1/publish", { + topic: "quiet/errors", + payload: { level: "warn" }, + retain: true, + }); + await server.engine.idle(); + await post("/v1/reset", {}); + await server.engine.idle(); + const state = (await (await req("/v1/state")).json()) as { + state: Array<{ topic: string; payload: unknown }>; + }; + const entry = state.state.find((e) => e.topic === "quiet/errors"); + expect(entry?.payload).toEqual({ level: "warn" }); // NOT overwritten by a floor republish +}); diff --git a/src/control-plane/index.ts b/src/control-plane/index.ts index c9f86f8..d2e7730 100644 --- a/src/control-plane/index.ts +++ b/src/control-plane/index.ts @@ -111,6 +111,8 @@ export async function buildTopicInfo( example: "payload" in floor ? floor.payload : undefined, qos: c.qos, retain: c.retain, + // R-040: only-when-suppressed — undefined serializes as absent + initialState: c.initialState === false ? (false as const) : undefined, }); } return infos; diff --git a/src/engine/index.test.ts b/src/engine/index.test.ts index c54a7fa..5ab0efc 100644 --- a/src/engine/index.test.ts +++ b/src/engine/index.test.ts @@ -52,6 +52,24 @@ function makeRegistry(): SpecRegistry { }; } +// makeRegistry()'s state/{deviceId} channel, but declared reactive-only (R-040) +function flaggedRegistry(): SpecRegistry { + const state = { + ...makeChannel("state/{deviceId}", stateSchema, 2, true), + initialState: false, + }; + return { + diagnostics: () => [], + match(topic: string) { + const m = topic.match(/^state\/([^/]+)$/); + if (m?.[1]) return { channel: state, params: { deviceId: m[1] } }; + return undefined; + }, + matchesFilter: () => false, + channels: () => [state], + }; +} + function buildEngine( overrides: Parameters[0] = {}, registry: SpecRegistry = makeRegistry(), @@ -778,3 +796,99 @@ test("start(): a seed entry resolving to a fromClient channel surfaces loudly an "does not resolve to a toClient channel instance", ); }); + +// [utest->R-040] +test("subscribe on an initialState:false channel records the instance and emits nothing", async () => { + const { engine, emitted, violations } = buildEngine({}, flaggedRegistry()); + engine.onSubscribe("state/d7"); + await engine.idle(); + expect(emitted).toEqual([]); + expect(violations).toEqual([]); + expect(engine.instances.snapshot()).toEqual({ + instances: [ + { channelAddress: "state/{deviceId}", params: { deviceId: "d7" } }, + ], + }); +}); + +// [utest->R-040] +test("start(): an initialState:false literal channel is skipped by the eager sweep", async () => { + const flagged = { + ...makeChannel("plain/topic", stateSchema, 1, true), + initialState: false, + }; + const reg: SpecRegistry = { + diagnostics: () => [], + match: (topic) => + topic === "plain/topic" ? { channel: flagged, params: {} } : undefined, + matchesFilter: () => false, + channels: () => [flagged], + }; + const { engine, emitted } = buildEngine({}, reg); + engine.start(); + await engine.idle(); + expect(emitted).toEqual([]); +}); + +// [utest->R-040] +test("start() + reset(): seeded instances on an initialState:false channel land in the ledger but never republish", async () => { + const { engine, emitted } = buildEngine({}, flaggedRegistry(), { + "state/{deviceId}": [{ deviceId: "d9" }], + }); + engine.start(); + await engine.idle(); + expect(engine.instances.snapshot()).toEqual({ + instances: [ + { channelAddress: "state/{deviceId}", params: { deviceId: "d9" } }, + ], + }); + expect(emitted).toEqual([]); + engine.reset(undefined); + await engine.idle(); + expect(emitted).toEqual([]); +}); + +// [utest->R-040] +test("engine.handlers() reports pattern, modulePath and initialState presence in precedence order", () => { + const { engine, dispatch } = buildEngine(); + dispatch.register("state/{deviceId}", () => ({ initialState() {} }), "a.ts"); + dispatch.register("state/{deviceId}", () => ({ onInbound() {} }), "b.ts"); + dispatch.instantiate(); + expect(engine.handlers()).toEqual([ + { pattern: "state/{deviceId}", modulePath: "a.ts", hasInitialState: true }, + { pattern: "state/{deviceId}", modulePath: "b.ts", hasInitialState: false }, + ]); +}); + +// [utest->R-040] +test("an L3 initialState handler still runs on an initialState:false channel (handler wins)", async () => { + const flagged = { + ...makeChannel("thing/{id}", { type: "object" }, 1, false), + initialState: false, + }; + const reg: SpecRegistry = { + diagnostics: () => [], + match(topic: string) { + const m = topic.match(/^thing\/([^/]+)$/); + if (m?.[1]) return { channel: flagged, params: { id: m[1] } }; + return undefined; + }, + matchesFilter: () => false, + channels: () => [flagged], + }; + const { engine, emitted, dispatch } = buildEngine({}, reg); + dispatch.register( + "thing/{id}", + () => ({ + initialState(topic, ctx) { + ctx.publish({ topic, payload: { marker: "authored" } }); + }, + }), + "h.ts", + ); + dispatch.instantiate(); + engine.onSubscribe("thing/t1"); + await engine.idle(); + expect(emitted.length).toBe(1); + expect(emitted[0]?.payload).toEqual({ marker: "authored" }); +}); diff --git a/src/engine/index.ts b/src/engine/index.ts index 9426f0a..1a9d716 100644 --- a/src/engine/index.ts +++ b/src/engine/index.ts @@ -48,6 +48,13 @@ export interface L2Dispatch { export interface Engine { loadHandlers(dir: string): Promise; + // R-040: read-only view over the dispatch registry for the compose root's + // flag-vs-handler contradiction warn-log (precedence-sorted, instantiate()-gated) + handlers(): { + pattern: string; + modulePath: string; + hasInitialState: boolean; + }[]; start(): void; onInbound(event: InboundEvent): void; onSubscribe(topic: string): void; @@ -200,6 +207,11 @@ export function createEngine(deps: EngineDeps): Engine { ); return; } + // R-040: a reactive-only channel declares it has no initial state + // (topicOverrides initialState: false) — the floor is off on every leg + // through this function; the ledger record above, L3 initialState + // handlers, and all L2/L3 emissions stay untouched + if (m.channel.initialState === false) return; // L1 is the proactive floor: keyed per instance params (F7) const out = await l1Floor(m.channel, (ch) => faker(ch, m.params)); if ("violation" in out) { @@ -231,6 +243,16 @@ export function createEngine(deps: EngineDeps): Engine { return paths; }, + // R-040: read-only view for the compose root's contradiction warn-log — + // which handlers exist, on which channel pattern, and whether they define + // initialState (dispatch.all() is instantiate()-gated and precedence-sorted) + handlers: () => + dispatch.all().map(({ handler, registration }) => ({ + pattern: registration.pattern, + modulePath: registration.modulePath, + hasInitialState: typeof handler.initialState === "function", + })), + start() { // seedInstances pre-materializes the deterministic demo set (§2/F1); // an entry that doesn't resolve to a toClient channel instance is diff --git a/src/model/index.ts b/src/model/index.ts index b876c22..d3ebb25 100644 --- a/src/model/index.ts +++ b/src/model/index.ts @@ -28,6 +28,10 @@ export interface Channel { validate: (payload: unknown) => SchemaError[]; qos?: 0 | 1 | 2; retain?: boolean; + // R-040: registry-resolved from topicOverrides.initialState ONLY (no spec-binding + // tier), onto toClient records only (the floor never runs elsewhere); absent ⇒ the + // §2 initial-state floor applies; false ⇒ reactive-only channel + initialState?: boolean; title?: string; description?: string; } @@ -90,7 +94,10 @@ export interface ServiceConfig { qosDefault?: 0 | 1 | 2; // per-service default qos — tier 3 of the §2 precedence chain retainDefault?: boolean; // per-service default retain — tier 3 // per-topic override — tier 2; key = channel address (the {param} form), string-equality matched (F14) - topicOverrides?: Record; + topicOverrides?: Record< + string, + { qos?: 0 | 1 | 2; retain?: boolean; initialState?: boolean } + >; // channel address → list of param-maps; pre-materializes a deterministic demo set (F1, §2) seedInstances?: Record[]>; } @@ -155,6 +162,9 @@ export interface TopicInfo { example?: unknown; qos?: 0 | 1 | 2; retain?: boolean; + // R-040: present ONLY when the channel declares initialState: false; absent + // otherwise. Survives ?schema=false (that view drops `schema` alone). + initialState?: false; } export type ViolationKind = "schema" | "direction" | "unknown-topic" | "decode"; diff --git a/src/registry/index.test.ts b/src/registry/index.test.ts index b7c9282..02231d8 100644 --- a/src/registry/index.test.ts +++ b/src/registry/index.test.ts @@ -1,9 +1,14 @@ import { expect, test } from "bun:test"; import { readdirSync } from "node:fs"; import { loadConfig, loadServices } from "#src/config/index.ts"; -import { DEFAULT_CONFIG, type ServiceConfig } from "#src/model/index.ts"; +import { + type Channel, + DEFAULT_CONFIG, + type Diagnostic, + type ServiceConfig, +} from "#src/model/index.ts"; import { SUPPORTED_SPEC_VERSIONS } from "#src/model/spec-version.ts"; -import { buildRegistry } from "./index.ts"; +import { buildRegistry, mergeRegistries } from "./index.ts"; // [utest->R-004] // [utest->R-026] @@ -596,6 +601,207 @@ test("the oldest supported major parses, inverts direction, and its binding is r expect(reg.diagnostics()).toEqual([]); }); +// [utest->R-040] +test("topicOverrides.initialState resolves onto the Channel; absent stays undefined", async () => { + const spec = `asyncapi: 2.6.0 +info: { title: T, version: 1.0.0 } +channels: + errors/{sessionId}: + parameters: + sessionId: { schema: { type: string } } + subscribe: + operationId: err + message: + payload: { type: object, properties: { msg: { type: string } } } + state/{sessionId}: + parameters: + sessionId: { schema: { type: string } } + subscribe: + operationId: st + message: + payload: { type: object, properties: { v: { type: string } } } +`; + const reg = await buildRegistry({ + specText: spec, + service: "s", + config: DEFAULT_CONFIG, + serviceConfig: { + name: "s", + repo: "x", + specPath: "y", + topicOverrides: { "errors/{sessionId}": { initialState: false } }, + }, + }); + expect(reg.match("errors/abc")?.channel.initialState).toBe(false); + expect(reg.match("state/abc")?.channel.initialState).toBeUndefined(); + expect(reg.diagnostics()).toEqual([]); +}); + +// [utest->R-040] +test("a dangling topicOverrides key warns once and is otherwise ignored", async () => { + const spec = `asyncapi: 2.6.0 +info: { title: T, version: 1.0.0 } +channels: + t/real: + subscribe: + operationId: s + message: + payload: { type: object, properties: { a: { type: string } } } +`; + const reg = await buildRegistry({ + specText: spec, + service: "s", + config: DEFAULT_CONFIG, + serviceConfig: { + name: "s", + repo: "x", + specPath: "y", + topicOverrides: { "t/nope": { qos: 0, initialState: false } }, + }, + }); + const warns = reg + .diagnostics() + .filter((d) => d.detail.startsWith("override-dangling-key:")); + expect(warns.length).toBe(1); + expect(warns[0]?.severity).toBe("warning"); + expect(warns[0]?.source).toBe("t/nope"); + // dangling ⇒ ONLY the dangling warning, not the direction/type warnings too + expect(reg.diagnostics().length).toBe(1); +}); + +// [utest->R-040] +test("a non-boolean initialState warns and is ignored (the floor applies)", async () => { + const spec = `asyncapi: 2.6.0 +info: { title: T, version: 1.0.0 } +channels: + t/one: + subscribe: + operationId: s + message: + payload: { type: object, properties: { a: { type: string } } } +`; + const reg = await buildRegistry({ + specText: spec, + service: "s", + config: DEFAULT_CONFIG, + serviceConfig: { + name: "s", + repo: "x", + specPath: "y", + topicOverrides: { "t/one": { initialState: "false" } }, + } as unknown as ServiceConfig, + }); + const warns = reg + .diagnostics() + .filter((d) => d.detail.startsWith("initial-state-non-boolean:")); + expect(warns.length).toBe(1); + expect(warns[0]?.source).toBe("t/one"); + expect(reg.match("t/one")?.channel.initialState).toBeUndefined(); +}); + +// [utest->R-040] +test("initialState:false on an address with no toClient operation warns; a dual-direction address does not", async () => { + // v2: one channel with BOTH subscribe (toClient) and publish (fromClient) + // operations = two Channel records sharing the address; plus a publish-only + // (fromClient-only) channel + const spec = `asyncapi: 2.6.0 +info: { title: T, version: 1.0.0 } +channels: + duplex/{id}: + parameters: + id: { schema: { type: string } } + subscribe: + operationId: out + message: + payload: { type: object, properties: { a: { type: string } } } + publish: + operationId: inbound + message: + payload: { type: object, properties: { a: { type: string } } } + cmd/{id}: + parameters: + id: { schema: { type: string } } + publish: + operationId: cmd + message: + payload: { type: object, properties: { a: { type: string } } } +`; + const reg = await buildRegistry({ + specText: spec, + service: "s", + config: DEFAULT_CONFIG, + serviceConfig: { + name: "s", + repo: "x", + specPath: "y", + topicOverrides: { + "duplex/{id}": { initialState: false }, + "cmd/{id}": { initialState: false }, + }, + }, + }); + const warns = reg + .diagnostics() + .filter((d) => d.detail.startsWith("initial-state-on-from-client:")); + expect(warns.length).toBe(1); + expect(warns[0]?.source).toBe("cmd/{id}"); + // the toClient record of the dual-direction address carries the flag + const duplex = reg.channels().filter((c) => c.topic === "duplex/{id}"); + expect( + duplex.some((c) => c.direction === "toClient" && c.initialState === false), + ).toBe(true); +}); + +// [utest->R-040] +test("initialState never resolves onto a fromClient record, even on a dual-direction address", async () => { + // v2: one channel with BOTH subscribe (toClient) and publish (fromClient) + // operations = two Channel records sharing the address; plus a publish-only + // (fromClient-only) channel + const spec = `asyncapi: 2.6.0 +info: { title: T, version: 1.0.0 } +channels: + duplex/{id}: + parameters: + id: { schema: { type: string } } + subscribe: + operationId: out + message: + payload: { type: object, properties: { a: { type: string } } } + publish: + operationId: inbound + message: + payload: { type: object, properties: { a: { type: string } } } + cmd/{id}: + parameters: + id: { schema: { type: string } } + publish: + operationId: cmd + message: + payload: { type: object, properties: { a: { type: string } } } +`; + const reg = await buildRegistry({ + specText: spec, + service: "s", + config: DEFAULT_CONFIG, + serviceConfig: { + name: "s", + repo: "x", + specPath: "y", + topicOverrides: { + "duplex/{id}": { initialState: false }, + "cmd/{id}": { initialState: false }, + }, + }, + }); + const fromClient = reg.channels().filter((c) => c.direction === "fromClient"); + expect(fromClient.length).toBeGreaterThan(0); + expect(fromClient.every((c) => c.initialState === undefined)).toBe(true); + const duplex = reg.channels().filter((c) => c.topic === "duplex/{id}"); + expect( + duplex.some((c) => c.direction === "toClient" && c.initialState === false), + ).toBe(true); +}); + // [utest->R-039] test("an out-of-range binding qos is rejected and falls through the precedence chain", async () => { // 2.x maps `mqtt` to an empty schema, so qos 9 parses clean upstream and @@ -712,3 +918,98 @@ operations: // from the global default of 1, so this assertion is not vacuous expect(reg.match("t/ext")?.channel.qos).toBe(2); }); + +function mergeChan( + topic: string, + service: string, + initialState?: boolean, + direction: "toClient" | "fromClient" = "toClient", +): Channel { + return { + topic, + direction, + service, + schema: {}, + validate: () => [], + initialState, + } as unknown as Channel; +} + +function mergeReg(diags: Diagnostic[], ...channels: Channel[]) { + return { + diagnostics: () => diags, + channels: () => channels, + match: () => undefined, + matchesFilter: () => false, + }; +} + +// [utest->R-040] +test("mergeRegistries warns on an exact-address initialState disagreement, naming the winning service", () => { + const merged = mergeRegistries([ + mergeReg([], mergeChan("errors/all", "first")), + mergeReg([], mergeChan("errors/all", "second", false)), + ]); + const warns = merged + .diagnostics() + .filter((d) => d.detail.startsWith("initial-state-cross-service:")); + expect(warns.length).toBe(1); + expect(warns[0]?.severity).toBe("warning"); + expect(warns[0]?.source).toBe("errors/all"); + expect(warns[0]?.detail).toContain("'first' wins the match"); +}); + +// [utest->R-040] +test("mergeRegistries names the fromClient record winning the match, not the first toClient service", () => { + // 'gate' declares the same address fromClient FIRST: it wins the match (same + // param count, earlier flatMap order), so the floor never runs on this + // address and BOTH toClient declarations are dead — the warning must not + // name 'first' as the winner. + const merged = mergeRegistries([ + mergeReg([], mergeChan("errors/all", "gate", undefined, "fromClient")), + mergeReg([], mergeChan("errors/all", "first")), + mergeReg([], mergeChan("errors/all", "second", false)), + ]); + // pin the premise: the match winner IS the fromClient record + expect(merged.match("errors/all")?.channel.service).toBe("gate"); + expect(merged.match("errors/all")?.channel.direction).toBe("fromClient"); + const warns = merged + .diagnostics() + .filter((d) => d.detail.startsWith("initial-state-cross-service:")); + expect(warns.length).toBe(1); + expect(warns[0]?.detail).toContain( + "a fromClient record from 'gate' wins the match", + ); + expect(warns[0]?.detail).toContain("every initialState declaration is dead"); + expect(warns[0]?.detail).not.toContain("'first' wins"); +}); + +// [utest->R-040] +test("mergeRegistries: agreement, single-service duplicates, and child diagnostics pass through unwarned", () => { + const childDiag: Diagnostic = { + kind: "spec-load", + severity: "warning", + detail: + "override-dangling-key: 'x' matches no channel address in service 'a', so this topicOverrides entry is ignored", + source: "x", + }; + const merged = mergeRegistries([ + mergeReg([childDiag], mergeChan("errors/all", "a", false)), + mergeReg([], mergeChan("errors/all", "b", false)), // agreement: both false + mergeReg( + [], + mergeChan("dup/one", "c"), + mergeChan("dup/one", "c"), // same service twice: not cross-service + ), + mergeReg( + [], + mergeChan("dup/two", "c"), + mergeChan("dup/two", "c", false), // same service, DISAGREEING — must still not warn + ), + ]); + const cross = merged + .diagnostics() + .filter((d) => d.detail.startsWith("initial-state-cross-service:")); + expect(cross).toEqual([]); + expect(merged.diagnostics()).toContainEqual(childDiag); +}); diff --git a/src/registry/index.ts b/src/registry/index.ts index a5a9faa..bcfaeea 100644 --- a/src/registry/index.ts +++ b/src/registry/index.ts @@ -313,19 +313,63 @@ export async function buildRegistry(opts: { override?.retain ?? opts.serviceConfig?.retainDefault ?? false; + const direction = directionOf(op.action()); + // R-040: resolved onto toClient records only — the §2 floor is a toClient + // concept, and surfacing the flag on a fromClient row would contradict the + // initial-state-on-from-client warning + const initialState = + direction === "toClient" && typeof override?.initialState === "boolean" + ? override.initialState + : undefined; channels.push({ topic: address, - direction: directionOf(op.action()), + direction, service: opts.service, schema, validate, qos, retain, + initialState, title: msg?.title() ?? undefined, description: ch.description() ?? msg?.description() ?? undefined, }); } + // R-040: topicOverrides is a pure lookup above, so a mistyped key or value + // is silent there — this sweep is the loud counterpart (one warning per key, + // never per operation, so a dual-direction address cannot double-fire) + for (const [key, value] of Object.entries( + opts.serviceConfig?.topicOverrides ?? {}, + )) { + const matching = channels.filter((c) => c.topic === key); + if (matching.length === 0) { + diagnostics.push({ + kind: "spec-load", + severity: "warning", + detail: `override-dangling-key: '${key}' matches no channel address in service '${opts.service}', so this topicOverrides entry is ignored`, + source: key, + }); + continue; + } + const raw = value.initialState; + if (raw !== undefined && typeof raw !== "boolean") { + diagnostics.push({ + kind: "spec-load", + severity: "warning", + detail: `initial-state-non-boolean: '${key}' topicOverrides initialState is ${JSON.stringify(raw)}; initialState MUST be a boolean, so it is ignored and the floor applies`, + source: key, + }); + } + if (raw === false && !matching.some((c) => c.direction === "toClient")) { + diagnostics.push({ + kind: "spec-load", + severity: "warning", + detail: `initial-state-on-from-client: '${key}' has initialState: false but no toClient operation, and the initial-state floor only runs toClient, so the flag is ignored`, + source: key, + }); + } + } + // most-specific first (fewer params = more literal segments), then declaration order const ordered = channels .map((c, i) => ({ c, i })) @@ -355,6 +399,44 @@ export async function buildRegistry(opts: { // order, which across services is services.yaml key order. export function mergeRegistries(registries: SpecRegistry[]): SpecRegistry { const channels = registries.flatMap((r) => [...r.channels()]); + // R-040: an exact-address duplicate across services resolves by match order + // (for identical addresses: services.yaml key order), so a disagreeing + // initialState on the losing record is silently dead — surface it at the + // only cross-service seam. Parametrized shadowing (a literal address in one + // service shadowing a flagged {param} address in another) stays a known + // residual, recorded in D-025. + const crossService: Diagnostic[] = []; + const byTopic = new Map(); + for (const c of channels) { + // R-040: only toClient records carry the flag + if (c.direction !== "toClient") continue; + const group = byTopic.get(c.topic); + if (group) group.push(c); + else byTopic.set(c.topic, [c]); + } + for (const [topic, group] of byTopic) { + const services = [...new Set(group.map((c) => c.service))]; + if (services.length < 2) continue; + const stances = new Set(group.map((c) => c.initialState === false)); + if (stances.size < 2) continue; + // The match winner for an exact address is the first same-topic record in + // flatMap order, ANY direction — `group` holds only toClient records, so a + // same-address fromClient record declared earlier wins instead, and then + // the engine's floor never runs there (it returns for non-toClient matches). + const winner = channels.find((c) => c.topic === topic); + const verdict = + winner?.direction === "toClient" + ? `'${winner.service}' wins the match, so the other declaration is dead` + : `a fromClient record from '${winner?.service}' wins the match, so the floor never runs there and every initialState declaration is dead`; + crossService.push({ + kind: "spec-load", + severity: "warning", + detail: `initial-state-cross-service: '${topic}' is declared by ${services + .map((s) => `'${s}'`) + .join(" and ")} with disagreeing initialState; ${verdict}`, + source: topic, + }); + } const ordered = channels .map((c, i) => ({ c, i })) .sort((a, b) => { @@ -364,7 +446,10 @@ export function mergeRegistries(registries: SpecRegistry[]): SpecRegistry { }); return { channels: () => channels, - diagnostics: () => registries.flatMap((r) => [...r.diagnostics()]), + diagnostics: () => [ + ...registries.flatMap((r) => [...r.diagnostics()]), + ...crossService, + ], matchesFilter: (filter, topic) => matches(filter, topic), match: (topic) => { for (const { c } of ordered) { diff --git a/test/cli-dispatch.test.ts b/test/cli-dispatch.test.ts index 3b44e82..c9508f8 100644 --- a/test/cli-dispatch.test.ts +++ b/test/cli-dispatch.test.ts @@ -469,6 +469,28 @@ test("renderTopicList flattens allOf into one field list and marks oneOf variant expect(marked).toContain("- level: enum(low|high)"); }); +// [utest->R-040] +test("renderTopicList marks initialState:false channels in both views", () => { + const t: TopicInfo = { + topic: "alerts/x", + direction: "toClient", + service: "s", + schema: {}, + initialState: false, + }; + expect(renderTopicList([t])).toContain("[no initial state]"); + expect(renderTopicList([t], { compact: true })).toContain( + "[no initial state]", + ); + const plain: TopicInfo = { + topic: "state/x", + direction: "toClient", + service: "s", + schema: {}, + }; + expect(renderTopicList([plain])).not.toContain("[no initial state]"); +}); + test("validation collapses repeats to ×N with the EQ6 composed headline; -v expands; --json round-trips (ER2)", async () => { const r = io(); await run(["reset", ...CTRL_FLAG], r.io);