From b5f5e5aea4cfd44455a44f0fd3020d0621fb8195 Mon Sep 17 00:00:00 2001 From: nzneit Date: Sat, 1 Aug 2026 19:26:46 -0400 Subject: [PATCH 01/17] docs(intake): per-channel initial-state opt-out for reactive-only channels Design intake for suppressing the L1 floor on channels with no initial state (error/notification topics), where a synthetic on-subscribe draw can corrupt a stateful client. Resolved via design dialog: topicOverrides.initialState:false, handler-wins-with-warning, four spec-load warnings; allocates D-025/R-040 with the implementation PR. --- .../intake/2026-08-01-initial-state-optout.md | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 docs/intake/2026-08-01-initial-state-optout.md diff --git a/docs/intake/2026-08-01-initial-state-optout.md b/docs/intake/2026-08-01-initial-state-optout.md new file mode 100644 index 0000000..eba6da8 --- /dev/null +++ b/docs/intake/2026-08-01-initial-state-optout.md @@ -0,0 +1,43 @@ +# 2026-08-01: per-channel initial-state opt-out for reactive-only channels (intake) +**Status**: open +**Owner**: nzneit + +Problem: the L1 floor materializes initial state on every concrete subscribe to a `toClient` channel (and again in the startup eager sweep and on every `reset` republish — one emission site, `materializeAndPublish`, `src/engine/index.ts:184-212`). For **reactive-only channels** — error and notification topics — there is no initial state to materialize, and a stateful client can be driven into a bad state by a synthetic error it never asked for. Today the only way to silence the floor on a channel is an L3 handler with an empty `initialState` hook: real per-channel code for what should be a declaration. L2 cannot preempt the proactive path at all (contracts §3: proactive resolves L3 → L1). Adjacent prior art: D-009 declined a tick-varying L1 draw partly because random churn "would immediately demand a quiet-toggle"; this item supplies that toggle, per channel, for the subscribe-leg floor D-009 ratified. + +## a — Should the always-on floor gain a per-channel opt-out, flip to a retain-keyed "faithful" default, or flip to global opt-in? +Against a real broker, a subscriber to a non-retained topic receives nothing until the service publishes. The always-on floor (design §7a, which records the decision and its populated-UI first-run rationale) deliberately departs from that; the fidelity trade is implicit today, and this item's doc impact makes it explicit in §7a. A retain-keyed default (materialize only retained channels) is the principled alternative, but `retain` resolves to `false` at the bottom of the §2 precedence chain, so out of the box nearly every channel would go silent — it guts the floor's demo value and breaks frozen G3 behavior for every existing setup. Global opt-in is the same problem, worse. +→ Resolution: **per-channel opt-out; the always-on default is preserved** (decided 2026-08-01; the fidelity argument moves to the docs as "why this flag exists", not into the default) → allocates D-025 with the implementation PR. + +## b — Where does the declaration live, and what is it called? +Options considered: (1) grow the `topicOverrides` value type; (2) a new `ServiceConfig` role field (`reactiveOnly: [addr, …]`); (3) an `x-offbook` extension in the AsyncAPI doc itself. Precedents: F1 put channel-scoped config on `ServiceConfig` keyed by channel address; F14 pinned key matching to string equality on the `{param}`-form address; G13 resolves per-channel config onto the `Channel` record. The contract's own vocabulary for this path is the "initial-state materialization policy (G3)" (contracts §2), and the L3 hook that shadows it is named `initialState` — the flag reuses that word rather than inventing "suppress"/"reactive" vocabulary. Spec-side annotation is philosophically right (the spec is where "reactive-only" belongs) but requires editing other teams' specs — exactly the cross-team friction offbook exists to route around. +→ Resolution: **`topicOverrides` values grow `initialState?: boolean`** — absent means `true` (today's behavior); only `false` is meaningful. The registry resolves it onto the `Channel` record as `initialState: boolean` alongside `qos`/`retain`; unlike those there is no spec-binding tier above it — the override is the field's only author in v1. No service-level `initialStateDefault` (no purely-reactive service exists to justify it); the spec-side extension is deferred as a possible v2 read-through. Amends contracts §1 (`Channel` gains the resolved field), §2 (G3 policy sentence), §5 (`TopicInfo`, see Observability), and §6 (`ServiceConfig`) → folds into D-025. + +## c — Flag vs a registered L3 `initialState` handler on the same channel: who wins? +Config says "this channel has no initial state"; code says "here is its initial state". Options: handler wins with a warning / handler wins silently / load error / config wins. Config-over-code silently disables code the adopter wrote; a load error punishes the legitimate move of temporarily dropping a debugging handler onto a suppressed channel. +→ Resolution: **the handler wins, with a load-time warning** naming the channel and handler file ("`initialState: false` but handler defines `initialState`; the handler wins"). Mechanics: the check runs in the compose root after handlers and registry are loaded — the registry never sees handlers and the engine has no log channel, so the natural surface is the compose root's injected `log` (offbook.log), the same path `caps.warn` already rides (`src/compose/index.ts:63-70`, `:240`). It is a pure function of (loaded handlers × current registry): it re-runs on `POST /v1/specs/refresh` (registry rebuild), and under `up --watch` it recurs naturally on each restart-watch respawn (l2 §8 vocabulary; there is no in-process handler reload to hook). F19 is untouched: the check resolves handler patterns against the registry non-bindingly, for diagnosis only; dispatch-time binding stays lazy. The layer model stays intact: the flag governs only the L1 floor; L3 remains most-specific on every path → folds into D-025. + +## d — How loud is a mistyped or misplaced flag? +`topicOverrides` is consumed as a pure lookup (`src/registry/index.ts:308`); nothing walks the keys, so a typo'd key is silent today. For `qos`/`retain` that is a minor fidelity leak; a typo'd `initialState: false` silently resurrects the exact client-state bug the flag exists to fix. YAML also invites `initialState: "false"` (a truthy string). And because the boot merge resolves cross-service collisions by match order, a flag on a shadowed record would be silently dead. +→ Resolution: **four load-time warnings**, all `kind: "spec-load"` diagnostics in the existing registry idiom (cf. `binding-invalid-value`, `src/registry/index.ts:296`), with pinned detail-tag prefixes added to contracts §5's tag enumeration: +1. `override-dangling-key` — any `topicOverrides` key matching no channel address in that service's spec (covers `qos`/`retain` typos too). +2. `initial-state-on-from-client` — `initialState: false` on an address with **no `toClient` operation** in that service's spec. Address-scoped on purpose: a dual-direction address (one `send` + one `receive` operation) yields two `Channel` records sharing the topic, and the flag is meaningful for the `toClient` one, so its presence alone must not warn. +3. `initial-state-non-boolean` — non-boolean `initialState` value: warn and ignore rather than misread. +4. `initial-state-cross-service` — at the cross-service merge, two services' records can match the same concrete address with disagreeing resolved `initialState` (suppression is decided by whichever record wins `registry.match`, so the loser's flag is dead). Exact-address duplicates only in v1; the parametrized-shadowing variant (a literal address in one service shadowing a flagged `{param}` address in another under fewest-params-first ordering) is recorded below as a known residual. +→ folds into D-025. + +## Resolved design (mechanics) +- **Engine**: one gate in `materializeAndPublish`, after instance recording and the L3 `initialState` dispatch check, immediately before the `l1Floor` call (`src/engine/index.ts:204`): `channel.initialState === false` ⇒ return without drawing. The single site silences all three legs (subscribe, startup sweep, `reset`/`seedInstances` republish) at once. +- **Boot-time-only, by design**: like all of `topicOverrides`, the flag is read once at `offbook up`. `offbook specs update` / `POST /v1/specs/refresh` re-resolves it from the boot-time `ServiceConfig` during the registry rebuild but does **not** re-read services.yaml; changing the flag takes `down && up` (or an EH1 `--watch` respawn, which happens to re-read config). Plan-time verification: the F21 compiled-registry cache key must cover `ServiceConfig` (or be per-boot), so a changed flag can never be served from a stale compiled `Channel`. +- **Untouched by the flag**: the params ledger (instances still record), L2/L3 emissions to the channel, wildcard retained-store replay, `GET /v1/topics` example generation and `POST /v1/publish {example:true}` (explicit requests, not synthetic pushes). +- **Retained residue and `reset`**: the flag gates engine emissions only. It neither shields Aedes' native retained delivery on subscribe nor scrubs retained residue at reset — and on a flagged channel `republishInitialState` no longer overwrites that residue, so a retained payload (an L2/L3 emit or `POST /v1/publish` with `retain`, on a channel whose resolved `retain` is true) survives `reset` and is served to every subsequent subscriber until cleared. Contracts §2's "post-reset `/state` is deterministic by construction" note (and the §5 `/state` reads-table row) gain a flagged-channel caveat; the cookbook recommends `retain: false` for flagged channels unless retained mock state is intended. +- **Instances on a flagged channel** (EQ5 interaction, stated honestly): the "uninstantiated" info notice keeps firing while the channel has no instances, but on a flagged channel two of its three remedies (subscribe to a concrete topic, add `seedInstances`) clear the notice without rendering anything — instances record, nothing publishes; only "send a matching command" (a reactive L2/L3 emit) produces output. After materialization, the `/topics` `initialState` field is the surviving breadcrumb for "why is this channel quiet". Accepted as the designed observability story; the cookbook note covers it. +- **Observability**: `TopicInfo` (contracts §5) gains `initialState?: false` — present **only when suppressed**, absent otherwise (matches "only `false` is meaningful"), and it survives the `?schema=false` slim view (that flag drops only `schema`). Verify `offbook doctor` (which re-reads services.yaml independently, `src/cli/doctor.ts:180-219`) tolerates — ideally surfaces — the new key. +- **Doc impact**: contracts §1 (`Channel` type), §2 (G3 policy sentence + the post-reset determinism note), §5 (`TopicInfo`, the spec-load tag-prefix enumeration, the `/state` caveat), §6 (`ServiceConfig` type); design.md §7a gains the reactive-channel caveat and the now-explicit fidelity-trade rationale; a cookbook note in `docs/guides/` (derived; contracts stay canonical) covering the flag, the retain recommendation, and the flagged-channel EQ5 behavior; R-040 (arrow-tagged tests) and D-025 (citing D-009 as related prior art: the decision that named the quiet-toggle demand in the tick-churn context it declined) allocated with the implementation PR, which also resolves and archives this item. + +## Test plan +Engine: flagged channel subscribe emits nothing and records no violation; startup sweep and `reset` skip flagged channels; flagged channel with an L3 `initialState` handler emits the handler's payload and raises the contradiction warning; a retained publish on a flagged channel before `reset` is still present in `getState()` after `reset` and is delivered on a fresh concrete subscribe; unflagged channels behave byte-identically to today. Registry: the flag lands on `Channel`; warnings 1–3 each fire on their trigger and not otherwise, including the dual-direction address non-warning case; warning 4 fires at the merge on exact-address disagreement. Config: loader passthrough plus fixture update (`src/config/fixtures/services.yaml`). Control plane: `GET /v1/topics` exposes `initialState: false` on a flagged channel and omits the field elsewhere (both with and without `?schema=false`); `offbook doctor` runs clean over a services.yaml carrying the key. All tests arrow-tagged `[utest->R-040]` (or `itest`/`stest` as placed); gate on full `bun test` exit code. + +## Out of scope (recorded deliberately) +`initialStateDefault` per service; the spec-side `x-offbook` extension (v2 candidate); the parametrized cross-service shadowing residual (warning 4 covers exact-address duplicates only; the matcher-overlap analysis in `src/scenarios/matcher.ts` exists if it ever earns an item); a `seedInstances` lint (dangling entries are already loud — a startup `unknown-topic` error violation per entry, `src/engine/index.ts:246-258`, test-pinned at `src/engine/index.test.ts:709` — only an empty param-list entry is silent today; separate, smaller item); any L2-side suppression vocabulary. + +Source: design dialog 2026-08-01 (brainstorming session), grounded by a code/spec sweep of the emission path, config surface, and decision-log prior art (D-009, F1, F14, F19, F21, G13, G3, EQ5), then hardened by a 4-lens adversarial review (citations, corpus consistency, design holes, template/prose; 22 confirmed findings folded back in). From 05e7b5f343c8ef5c92199572d16b773e198a9fd4 Mon Sep 17 00:00:00 2001 From: nzneit Date: Sat, 1 Aug 2026 20:22:33 -0400 Subject: [PATCH 02/17] docs(plans): per-channel initial-state opt-out implementation plan --- docs/plans/2026-08-01-initial-state-optout.md | 1515 +++++++++++++++++ 1 file changed, 1515 insertions(+) create mode 100644 docs/plans/2026-08-01-initial-state-optout.md diff --git a/docs/plans/2026-08-01-initial-state-optout.md b/docs/plans/2026-08-01-initial-state-optout.md new file mode 100644 index 0000000..3cab9e2 --- /dev/null +++ b/docs/plans/2026-08-01-initial-state-optout.md @@ -0,0 +1,1515 @@ +# Per-Channel Initial-State Opt-Out Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** `topicOverrides.
.initialState: false` in services.yaml declares a reactive-only channel: the L1 proactive floor never publishes there, on any leg, while everything else (ledger, L2/L3, explicit surfaces) is untouched — with four spec-load warnings and a handler-contradiction warn-log making every misconfiguration loud. + +**Architecture:** The flag rides the existing per-channel config path end to end: config loader passes it through untyped (it already does — only the type widens), the registry resolves it onto `Channel.initialState` next to `qos`/`retain` and sweeps `topicOverrides` keys for warnings, `mergeRegistries` adds the cross-service disagreement warning, the engine's single emission site (`materializeAndPublish`) gains one early return, the compose root warn-logs flag-vs-L3-handler contradictions at handler load and after every specs refresh, and the control plane exposes `initialState: false` on suppressed `TopicInfo` rows. Contracts §1/§2/§5/§6, design §7a, the wiring guide, R-040, and D-025 are amended in-repo per the doc system. + +**Tech Stack:** Bun (runtime + test), TypeScript (`bun run typecheck` = `tsc --noEmit`), Biome (`bun run lint`), `bun scripts/check-docs.ts` (doc gate). + +**Spec:** `docs/intake/2026-08-01-initial-state-optout.md` (approved 2026-08-01; moves to `docs/archive/intake/` in Task 10). + +## Global Constraints + +- Branch: all work lands on `initial-state-optout` (it already carries the intake commit `b5f5e5a`). Do not push unless asked. +- Bun is the only runtime: tests via `bun test`, scripts via `bun scripts/.ts`. +- Full `bun test` is the authoritative gate, judged by **exit code**. A focused run (`bun test `) may exit 1 with ZERO test failures because of the per-file coverage floor in `bunfig.toml`; on focused runs trust the printed fail count, gate on full runs only. +- `bun scripts/check-docs.ts` must exit 0 before every commit (it is the pre-commit gate). +- Commit exactly at the plan's commit steps, no others. Never run `git config user.*`. Do NOT add any Co-Authored-By or AI-attribution trailer to commits. +- `docs/specs/contracts.md` wins every interface conflict. +- All four new diagnostics use `kind: "spec-load"` exactly. The `Diagnostic.kind` union and `DiagnosticSummary.byKind` are closed four-value sets (`src/model/index.ts:269-285`); `diagnosticSummary` in `src/control-plane/index.ts` increments `byKind[d.kind]` and would throw on a new kind. Never invent a kind. +- `Channel.initialState` is OPTIONAL (`initialState?: boolean`): absent ⇒ the floor applies; the engine gates on `=== false` only. Never make it required — hand-built `Channel` literals in `src/engine/index.test.ts`, `src/control-plane/index.test.ts`, `test/cli-dispatch.test.ts`, and `src/compose/initial-state.test.ts` (new) omit it. +- Arrow-tag grammar is strict: `// [utest->R-040]` / `// [itest->R-040]` — exactly three digits, no spaces, inside a comment, on its own line directly above the `test(` call (or in the file header block). Malformed and dangling tags fail the gate. +- Statuses stay honest: R-040 is allocated `specified` in Task 1 and flips to `tested` only in Task 10, when every clause of its statement is covered by the named TEST traces. +- TDD per task: write the failing test first, watch it fail, implement, watch it pass. +- Never run `biome migrate`. If `bun run lint -- --write` (or any `--write`) is used, read the diff before trusting it. +- Do not modify the transport-isolation or lint-gate rules; nothing here imports `aedes` outside `src/broker/`. + +## File Structure + +- Modify: `REQUIREMENTS.md` (R-040 entry), `DECISIONS.md` (D-025), `docs/specs/contracts.md` (§1 `Channel`, §2 G3 policy + reset bullet + back-anchor, §5 `TopicInfo` + `/state` row + tag list, §6 `ServiceConfig`), `docs/specs/design.md` (§7a), `docs/guides/wiring-your-service.md` (new §7) +- Move: `docs/intake/2026-08-01-initial-state-optout.md` → `docs/archive/intake/` (Task 10, same commit as the status flip) +- Modify: `src/model/index.ts` (`Channel`, `ServiceConfig`, `TopicInfo`) +- Modify: `src/config/fixtures/services.yaml` (new `serviceE` entry; `serviceC` untouched) +- Modify: `src/registry/index.ts` (resolution + key sweep + `mergeRegistries` warning) +- Modify: `src/engine/index.ts` (the gate + `handlers()` accessor) +- Modify: `src/compose/index.ts` (contradiction warn-log at start + refresh) +- Modify: `src/cli/boot.ts` (F21 invariant comment only) +- Modify: `src/control-plane/index.ts` (`buildTopicInfo`), `src/cli/index.ts` (`renderTopicList` marker) +- Create: `src/compose/initial-state.test.ts` (contradiction warn-log; isolated because handler files register on the process-global `defaultDispatch`) +- Test modifications: `src/config/index.test.ts`, `src/registry/index.test.ts`, `src/engine/index.test.ts`, `src/control-plane/index.test.ts`, `src/cli/doctor.test.ts`, `test/cli-dispatch.test.ts` + +--- + +### Task 1: Allocate R-040 (`specified`) + the contracts back-anchor + +The doc gate rejects arrow tags pointing at a nonexistent UID, so R-040 must exist before any tagged test is committed. Its `COVERS` anchor must resolve at the same commit, so the additive back-anchor lands in `contracts.md` now (doc-system.md §154 explicitly blesses additive `` 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. From 72c4f6871f9091d750ea4d3c69651273efa7f452 Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 14:11:46 -0400 Subject: [PATCH 04/17] feat(model,config): topicOverrides.initialState passthrough (R-040) --- src/config/fixtures/services.yaml | 5 +++++ src/config/index.test.ts | 8 ++++++++ src/model/index.ts | 8 +++++++- 3 files changed, 20 insertions(+), 1 deletion(-) 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/model/index.ts b/src/model/index.ts index b876c22..7103c00 100644 --- a/src/model/index.ts +++ b/src/model/index.ts @@ -28,6 +28,9 @@ 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); absent ⇒ the §2 initial-state floor applies; false ⇒ reactive-only channel + initialState?: boolean; title?: string; description?: string; } @@ -90,7 +93,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[]>; } From d10c9b1e22ecb1fa91a564584fd838fa3074215d Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 14:17:39 -0400 Subject: [PATCH 05/17] feat(registry): resolve initialState onto Channel + the topicOverrides key sweep (R-040) --- src/registry/index.test.ts | 151 +++++++++++++++++++++++++++++++++++++ src/registry/index.ts | 43 +++++++++++ 2 files changed, 194 insertions(+) diff --git a/src/registry/index.test.ts b/src/registry/index.test.ts index b7c9282..3b90e09 100644 --- a/src/registry/index.test.ts +++ b/src/registry/index.test.ts @@ -596,6 +596,157 @@ 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-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 diff --git a/src/registry/index.ts b/src/registry/index.ts index a5a9faa..0a63d36 100644 --- a/src/registry/index.ts +++ b/src/registry/index.ts @@ -313,6 +313,13 @@ export async function buildRegistry(opts: { override?.retain ?? opts.serviceConfig?.retainDefault ?? false; + // R-040: initialState rides topicOverrides alone — no binding tier above + // it, no service default below it; non-boolean values are warned by the + // post-loop key sweep, so only a real boolean lands on the Channel + const initialState = + typeof override?.initialState === "boolean" + ? override.initialState + : undefined; channels.push({ topic: address, direction: directionOf(op.action()), @@ -321,11 +328,47 @@ export async function buildRegistry(opts: { 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 })) From 8b67cea20352a696d2e4104d30de6d620c823292 Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 14:25:48 -0400 Subject: [PATCH 06/17] feat(registry): initial-state-cross-service warning at the merge seam (R-040) --- src/registry/index.test.ts | 72 ++++++++++++++++++++++++++++++++++++-- src/registry/index.ts | 32 ++++++++++++++++- 2 files changed, 101 insertions(+), 3 deletions(-) diff --git a/src/registry/index.test.ts b/src/registry/index.test.ts index 3b90e09..083ef4d 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 { + DEFAULT_CONFIG, + type Channel, + 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] @@ -863,3 +868,66 @@ 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, +): Channel { + return { + topic, + direction: "toClient", + 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'"); // the winner (earlier services.yaml key) +}); + +// [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 + ), + ]); + 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 0a63d36..2ac9d7d 100644 --- a/src/registry/index.ts +++ b/src/registry/index.ts @@ -398,6 +398,33 @@ 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) { + 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; + 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; '${group[0]?.service}' wins the match, so the other declaration is dead`, + source: topic, + }); + } const ordered = channels .map((c, i) => ({ c, i })) .sort((a, b) => { @@ -407,7 +434,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) { From 73df52c07d5ca86ae5e18e484720ffcb8e893075 Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 14:35:19 -0400 Subject: [PATCH 07/17] test(registry): strengthen cross-service warning assertions (R-040) --- src/registry/index.test.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/registry/index.test.ts b/src/registry/index.test.ts index 083ef4d..8bb8601 100644 --- a/src/registry/index.test.ts +++ b/src/registry/index.test.ts @@ -905,7 +905,7 @@ test("mergeRegistries warns on an exact-address initialState disagreement, namin expect(warns.length).toBe(1); expect(warns[0]?.severity).toBe("warning"); expect(warns[0]?.source).toBe("errors/all"); - expect(warns[0]?.detail).toContain("'first'"); // the winner (earlier services.yaml key) + expect(warns[0]?.detail).toContain("'first' wins the match"); }); // [utest->R-040] @@ -924,6 +924,11 @@ test("mergeRegistries: agreement, single-service duplicates, and child diagnosti 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() From ba7b4d8fffc2eba5c034342a4ee39e428bfeca19 Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 14:41:20 -0400 Subject: [PATCH 08/17] feat(engine): initialState:false gates the L1 floor at the one emission site (R-040) --- src/engine/index.test.ts | 102 +++++++++++++++++++++++++++++++++++++++ src/engine/index.ts | 5 ++ 2 files changed, 107 insertions(+) diff --git a/src/engine/index.test.ts b/src/engine/index.test.ts index c54a7fa..64fcff4 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,87 @@ 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("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..074edae 100644 --- a/src/engine/index.ts +++ b/src/engine/index.ts @@ -200,6 +200,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) { From 4b32d48b5e6c8ffbfe3e4e865b42b10f23d437af Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 14:48:00 -0400 Subject: [PATCH 09/17] docs(plans): unjoin Task 6 heading from separator --- docs/plans/2026-08-01-initial-state-optout.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/plans/2026-08-01-initial-state-optout.md b/docs/plans/2026-08-01-initial-state-optout.md index 3cab9e2..8b4c335 100644 --- a/docs/plans/2026-08-01-initial-state-optout.md +++ b/docs/plans/2026-08-01-initial-state-optout.md @@ -792,7 +792,9 @@ git add src/engine/index.ts src/engine/index.test.ts git commit -m "feat(engine): initialState:false gates the L1 floor at the one emission site (R-040)" ``` ----### Task 6: `engine.handlers()` + the compose contradiction warn-log + refresh re-check +--- + +### Task 6: `engine.handlers()` + the compose contradiction warn-log + refresh re-check The compose root never holds the dispatch registry (the engine falls back to the `defaultDispatch` singleton), so the engine grows a read-only `handlers()` view over `dispatch.all()`. Compose warn-logs a contradiction (flag says "no initial state", a loaded handler defines one — the handler wins) after handler load in `start()` and again after every `refreshSpecs` registry hot-swap. The warning goes through the injected `log` (offbook.log via serve.ts's sink), matching the house idiom: bare sentence, single-quoted identifiers. From fed86a3d9a114d3bfd58cd45ab810edad0631958 Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 14:50:38 -0400 Subject: [PATCH 10/17] feat(compose,engine): flag-vs-handler contradiction warn-log, re-checked on refresh (R-040) --- src/cli/boot.ts | 6 ++- src/compose/index.ts | 24 +++++++++ src/compose/initial-state.test.ts | 88 +++++++++++++++++++++++++++++++ src/engine/index.test.ts | 12 +++++ src/engine/index.ts | 13 +++++ 5 files changed, 142 insertions(+), 1 deletion(-) create mode 100644 src/compose/initial-state.test.ts 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/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/engine/index.test.ts b/src/engine/index.test.ts index 64fcff4..5ab0efc 100644 --- a/src/engine/index.test.ts +++ b/src/engine/index.test.ts @@ -848,6 +848,18 @@ test("start() + reset(): seeded instances on an initialState:false channel land 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 = { diff --git a/src/engine/index.ts b/src/engine/index.ts index 074edae..39c9f46 100644 --- a/src/engine/index.ts +++ b/src/engine/index.ts @@ -48,6 +48,9 @@ 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; @@ -236,6 +239,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 From 82b5369ba3837ca3bad80f7105bb67e2219fc48f Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 14:59:07 -0400 Subject: [PATCH 11/17] feat(control-plane,cli): expose initialState:false on /v1/topics + the topics views; pin residue + doctor tolerance (R-040) --- src/cli/doctor.test.ts | 12 ++++++++ src/cli/index.ts | 13 +++++++-- src/control-plane/index.test.ts | 51 +++++++++++++++++++++++++++++++++ src/control-plane/index.ts | 2 ++ src/model/index.ts | 3 ++ test/cli-dispatch.test.ts | 22 ++++++++++++++ 6 files changed, 101 insertions(+), 2 deletions(-) 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/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/model/index.ts b/src/model/index.ts index 7103c00..d99c645 100644 --- a/src/model/index.ts +++ b/src/model/index.ts @@ -161,6 +161,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/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); From d7d4cd8615c658bd6093eb7edd528be7a8af22b8 Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 15:04:04 -0400 Subject: [PATCH 12/17] style: apply biome formatting to R-040 template literals --- src/engine/index.ts | 6 +++++- src/registry/index.test.ts | 13 +++++++------ src/registry/index.ts | 4 +++- 3 files changed, 15 insertions(+), 8 deletions(-) diff --git a/src/engine/index.ts b/src/engine/index.ts index 39c9f46..1a9d716 100644 --- a/src/engine/index.ts +++ b/src/engine/index.ts @@ -50,7 +50,11 @@ 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 }[]; + handlers(): { + pattern: string; + modulePath: string; + hasInitialState: boolean; + }[]; start(): void; onInbound(event: InboundEvent): void; onSubscribe(topic: string): void; diff --git a/src/registry/index.test.ts b/src/registry/index.test.ts index 8bb8601..6cddd1e 100644 --- a/src/registry/index.test.ts +++ b/src/registry/index.test.ts @@ -2,8 +2,8 @@ import { expect, test } from "bun:test"; import { readdirSync } from "node:fs"; import { loadConfig, loadServices } from "#src/config/index.ts"; import { - DEFAULT_CONFIG, type Channel, + DEFAULT_CONFIG, type Diagnostic, type ServiceConfig, } from "#src/model/index.ts"; @@ -746,10 +746,10 @@ channels: 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); + 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] @@ -913,7 +913,8 @@ test("mergeRegistries: agreement, single-service duplicates, and child diagnosti 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", + detail: + "override-dangling-key: 'x' matches no channel address in service 'a', so this topicOverrides entry is ignored", source: "x", }; const merged = mergeRegistries([ diff --git a/src/registry/index.ts b/src/registry/index.ts index 2ac9d7d..cb70df8 100644 --- a/src/registry/index.ts +++ b/src/registry/index.ts @@ -421,7 +421,9 @@ export function mergeRegistries(registries: SpecRegistry[]): SpecRegistry { severity: "warning", detail: `initial-state-cross-service: '${topic}' is declared by ${services .map((s) => `'${s}'`) - .join(" and ")} with disagreeing initialState; '${group[0]?.service}' wins the match, so the other declaration is dead`, + .join( + " and ", + )} with disagreeing initialState; '${group[0]?.service}' wins the match, so the other declaration is dead`, source: topic, }); } From 2534caa8b121182d17fc1b305e0d733556f2799a Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 15:06:35 -0400 Subject: [PATCH 13/17] =?UTF-8?q?docs(specs):=20contracts=20=C2=A71/=C2=A7?= =?UTF-8?q?2/=C2=A75/=C2=A76=20+=20design=20=C2=A77a=20=E2=80=94=20initial?= =?UTF-8?q?State:=20false=20(R-040)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/specs/contracts.md | 17 +++++++++++------ docs/specs/design.md | 3 ++- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/docs/specs/contracts.md b/docs/specs/contracts.md index 79edb36..29fec16 100644 --- a/docs/specs/contracts.md +++ b/docs/specs/contracts.md @@ -38,6 +38,7 @@ interface Channel { // produced by the Spec Registry validate: (payload: unknown) => SchemaError[]; // compiled from `schema` qos?: 0 | 1 | 2; // RESOLVED by the registry per the §2 precedence chain (G13) retain?: boolean; // RESOLVED by the registry per the §2 precedence chain (G13) + initialState?: boolean; // RESOLVED by the registry from topicOverrides.initialState ONLY (no spec-binding tier; R-040/D-025) — absent ⇒ true (the §2 initial-state floor applies); false ⇒ reactive-only channel: the floor is OFF, everything else (§2 ledger, L2/L3, explicit surfaces) untouched title?: string; // from the AsyncAPI message/channel, when present description?: string; // " " " } @@ -127,10 +128,11 @@ interface BrokerModule { - **`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. @@ -274,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 | @@ -284,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 @@ -305,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 @@ -361,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. From 5a749c12b1f318d4be4576f21d69d1e32f8cb523 Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 17:54:21 -0400 Subject: [PATCH 14/17] =?UTF-8?q?docs(guides):=20wiring=20=C2=A77=20?= =?UTF-8?q?=E2=80=94=20reactive-only=20channels=20(R-040)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/guides/wiring-your-service.md | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/docs/guides/wiring-your-service.md b/docs/guides/wiring-your-service.md index 14f1d56..3dbc948 100644 --- a/docs/guides/wiring-your-service.md +++ b/docs/guides/wiring-your-service.md @@ -74,3 +74,33 @@ offbook specs update # re-resolve branch tips + hot-swap the running mock An ingested spec gives you topics, retained state, and validation. To make the mock *react* (ack commands, chain state changes), add L2 scenarios: [scenario cookbook](scenario-cookbook.md). + +## 7. Reactive-only channels + +Some `toClient` channels carry events, not state: error topics, +notifications. Offbook's default floor publishes a schema-valid example +when such a channel is subscribed (so UIs render populated), but a +synthetic error can drive a stateful client into a bad state. Declare +those channels reactive-only and the floor stays off: + +```yaml +services: + my-service: + repo: org/my-service + specPath: asyncapi.yaml + topicOverrides: + "errors/{sessionId}": { initialState: false } +``` + +- The channel stays silent until a scenario, a handler, or `offbook + publish` emits to it; validation is unaffected, and `offbook topics` + marks it (`GET /v1/topics` carries `initialState: false`). +- Typos are loud: a key matching no channel address, a non-boolean + value, or a flag on a channel with no `toClient` operation each + surface in `offbook diagnostics`. +- A handler that defines `initialState` on a flagged channel wins — the + contradiction is warn-logged, not silent. +- Prefer `retain: false` on flagged channels: a retained payload + published there survives `offbook reset` (nothing overwrites it). +- The flag is read at `offbook up`; `offbook specs update` does not + re-read services.yaml, so change it with a restart. From f2cbc63cca057d9229934a45fbc5c5ac5fd0df2d Mon Sep 17 00:00:00 2001 From: nzneit Date: Sun, 2 Aug 2026 18:01:14 -0400 Subject: [PATCH 15/17] =?UTF-8?q?docs:=20D-025=20=E2=80=94=20per-channel?= =?UTF-8?q?=20initial-state=20opt-out;=20resolve=20initial-state=20intake?= =?UTF-8?q?=20(R-040=20tested)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DECISIONS.md | 10 ++++++++++ REQUIREMENTS.md | 4 +++- .../intake/2026-08-01-initial-state-optout.md | 16 +++++++++------- 3 files changed, 22 insertions(+), 8 deletions(-) rename docs/{ => archive}/intake/2026-08-01-initial-state-optout.md (96%) diff --git a/DECISIONS.md b/DECISIONS.md index 3cf9b2c..31ea12c 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` (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). `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 ee9793a..71aa04d 100644 --- a/REQUIREMENTS.md +++ b/REQUIREMENTS.md @@ -320,8 +320,10 @@ Every error reachable on the clone→demo→init→wire→up→first-publish pat #### Per-channel initial-state opt-out (reactive-only channels) **UID**: R-040 -**STATUS**: specified +**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` (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.