feat: capture the whole wire surface, not just declarations

.env.example

@@ -1,4 +1,4 @@
-# Local-developer environment variables for the SPAN simulator.
+# Local-developer environment variables for panelbench.
 #
 # Copy to `.env` and fill in. `.env` is gitignored and must stay that way — it holds
 # a live access token.
@@ -21,6 +21,19 @@
 HA_URL=https://homeassistant.example:8123
 HA_TOKEN=
 
+# A checkout of the eBus specification, read by scripts/check-spec-provenance.py.
+#
+#   git clone https://github.com/electrification-bus/specification
+#
+# Optional and purely an optimisation: without it the script clones the spec to a
+# temp directory on every run, which works and needs no setup. A path that no
+# longer exists falls back to cloning rather than failing. `--spec` overrides it.
+#
+# The same variable name is used by span-panel-api, whose schema_1 conformance
+# suite byte-compares its own vendored catalogs against a checkout. One name for
+# one thing, so a single line in your shell serves both repositories.
+#EBUS_SPEC_DIR=/path/to/specification
+
 # EBUS_EMITTER_PATH is obsolete. The emitter is vendored under
 # `src/panelbench/ebus_emitter` and every dependency resolves from PyPI,
 # so nothing reads it. Safe to delete from any .env that still carries it.

DEVELOPER.md

@@ -197,6 +197,30 @@ uv run scripts/check-conformance.py \
 a change to our published surface should always be a reviewed edit to the expected profile, never something that slips through because no rule happened to cover
 it.
 
+### The wire capture, and why it is a second fixture
+
+`golden_tree.json` holds `$description` documents, because conformance is a question about what a device *declares*. That is not enough to exercise a **consumer**:
+a parser fed only descriptions can be asked whether it understands the shape of a panel, never whether it builds the right snapshot from one — which is the part
+that reaches a user. `golden_wire.json` holds the whole retained surface instead: descriptions, `$state`, and every property value, keyed the way a consumer
+receives them.
+
+```bash
+# Recapture. No broker needed — the emitter is assembled with the MQTT client substituted.
+uv run scripts/capture-wire.py
+```
+
+It goes through `start_clone`, the same assembly a real panel uses, with `publisher=` supplying a recorder in place of the aiomqtt client. Reassembling the
+emitter inside the script would make it a second copy of that wiring, free to drift, and a capture taken through different wiring than a panel uses proves less
+than it appears to.
+
+**Its values are not reproducible and nothing asserts them.** The config carries `noise_factor` and the clock advances, so power and current differ every run.
+`test_wire_capture.py` compares *shape* — which devices, which topics — so a property added, removed or renamed fails it while a different wattage does not. That
+is the opposite policy to `golden_report.json`, deliberately: a conformance profile must not drift silently, and a fixture full of noisy floats cannot be held to
+byte equality without producing failures nobody can act on.
+
+`span-panel-api-schema-1` vendors this capture and drives its parser end to end from it, which is the point of publishing it as an artifact rather than keeping it
+internal.
+
 ### If you are changing the emitter
 
 Read [`docs/spec-conformance-design.md`](docs/spec-conformance-design.md) before adding a capability to a profile or changing how properties reach the wire. Two

scripts/capture-wire.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+"""Recapture the wire fixture — descriptions, `$state`, and every property value.
+
+A thin CLI over `panelbench.conformance.wire_capture`, which is where the work
+lives so a test can check the committed fixture against what the emitter
+currently emits.
+
+`check-conformance.py --from-stdin` captures `$description` documents only,
+because conformance is a question about declarations. This captures the whole
+retained surface, because a consumer cannot be exercised by declarations alone.
+
+**Values are not reproducible.** The config carries `noise_factor`, so power and
+current differ every run, and the clock advances. Nothing should assert byte
+equality against this file — assert structure and let the values be
+representative. That is why it is a separate artifact from `golden_tree.json`,
+whose report *is* compared exactly.
+
+    uv run scripts/capture-wire.py                       # default 40-space panel
+    uv run scripts/capture-wire.py --config configs/default_MAIN_16.yaml
+    uv run scripts/capture-wire.py -o /tmp/wire.json
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import pathlib
+import sys
+
+from panelbench.emitter_adapter.wire_capture import capture
+
+_REPO = pathlib.Path(__file__).resolve().parent.parent
+_DEFAULT_CONFIG = _REPO / "configs" / "default_MAIN_40.yaml"
+_DEFAULT_OUTPUT = _REPO / "tests" / "conformance" / "fixtures" / "golden_wire.json"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    parser.add_argument("--config", type=pathlib.Path, default=_DEFAULT_CONFIG)
+    parser.add_argument("-o", "--output", type=pathlib.Path, default=_DEFAULT_OUTPUT)
+    args = parser.parse_args()
+
+    if not args.config.exists():
+        print(f"no such config: {args.config}", file=sys.stderr)
+        return 1
+
+    captured = asyncio.run(capture(args.config))
+    if not captured:
+        print("captured nothing; did the emitter start?", file=sys.stderr)
+        return 1
+
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(captured, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+    described = sum(1 for device in captured.values() if "$description" in device)
+    values = sum(1 for device in captured.values() for key in device if not key.startswith("$"))
+    print(
+        f"{args.output}: {len(captured)} devices, {described} described, {values} property values"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/check-spec-provenance.py

@@ -16,15 +16,22 @@
    never fails the build.
 
 Usage:
-    scripts/check-spec-provenance.py                     # clones the spec to a temp dir
-    scripts/check-spec-provenance.py --spec ~/spec-repo  # reuse a local checkout
+    scripts/check-spec-provenance.py                     # $EBUS_SPEC_DIR, else clone to a temp dir
+    scripts/check-spec-provenance.py --spec ~/spec-repo  # reuse a particular checkout
+
+`EBUS_SPEC_DIR` belongs in `.env` (see `.env.example`); the flag overrides it.
+Cloning works and needs no setup, so the variable is an optimisation — it saves a
+network round trip per run — not a requirement. A path that no longer exists falls
+back to cloning rather than failing, since that is the same situation as not
+having set it.
 """
 
 from __future__ import annotations
 
 import argparse
 import filecmp
 import json
+import os
 import pathlib
 import subprocess
 import sys
@@ -75,9 +82,20 @@ def _catalog_at_commit(spec: pathlib.Path, commit: str, name: str, out: pathlib.
 
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--spec", help="path to an existing specification checkout")
+    parser.add_argument(
+        "--spec",
+        default=os.environ.get("EBUS_SPEC_DIR"),
+        help="existing specification checkout (default: $EBUS_SPEC_DIR, else clone)",
+    )
     args = parser.parse_args()
 
+    # A stale path is the same situation as no path — the checkout is not there —
+    # and cloning is the documented fallback, so fall back rather than failing on
+    # a variable someone set months ago and a directory that has since moved.
+    if args.spec and not pathlib.Path(args.spec).is_dir():
+        print(f"EBUS_SPEC_DIR={args.spec} does not exist; cloning instead", file=sys.stderr)
+        args.spec = None
+
     if not LOCKFILE.exists():
         print(f"no provenance lockfile at {LOCKFILE}", file=sys.stderr)
         return 2

src/panelbench/emitter_adapter/runtime.py

@@ -238,6 +238,7 @@ async def start_clone(
     engine: DynamicSimulationEngine,
     *,
     broker: BrokerConnection | None = None,
+    publisher: MqttPublisher | None = None,
 ) -> CloneRuntime:
     """Assemble the emitter for ``engine``: build manifest, open MQTT, run
     lifecycle. Returns a runtime the panel holds across ticks.
@@ -246,7 +247,14 @@ async def start_clone(
         1. ``broker:`` section in the YAML config (config explicitness wins).
         2. The supplied ``broker`` fallback (typically constructed once by
            ``SimulatorApp`` from CLI/env).
-        3. Default 127.0.0.1:1883 anonymous (no TLS)."""
+        3. Default 127.0.0.1:1883 anonymous (no TLS).
+
+    ``publisher`` substitutes the MQTT client, and when given no broker is
+    resolved or connected. It exists so a capture can be taken through this
+    function rather than beside it: a script that reassembled the emitter itself
+    would be a second, quietly diverging copy of the wiring below, and a capture
+    is only worth anything if it went through the same assembly a real panel
+    does."""
     manifest = build_manifest(engine.config)
 
     uuid_to_circuit_id = {stable_circuit_uuid(c["id"]): c["id"] for c in engine.config["circuits"]}
@@ -263,18 +271,23 @@ async def start_clone(
         retain=lwt_retain,
     )
 
-    broker_cfg: BrokerConfigYAML = engine.config.get("broker") or {}
-    resolved = _resolve_broker(broker_cfg, broker)
-    mqtt = _AiomqttPublisher(
-        host=resolved.host,
-        port=resolved.port,
-        client_id=f"span-sim-{engine.serial_number}",
-        username=resolved.username,
-        password=resolved.password,
-        will=will,
-        ca_cert_path=resolved.ca_cert_path,
-    )
-    await mqtt.connect()
+    mqtt: MqttPublisher
+    if publisher is not None:
+        mqtt = publisher
+    else:
+        broker_cfg: BrokerConfigYAML = engine.config.get("broker") or {}
+        resolved = _resolve_broker(broker_cfg, broker)
+        aiomqtt_publisher = _AiomqttPublisher(
+            host=resolved.host,
+            port=resolved.port,
+            client_id=f"span-sim-{engine.serial_number}",
+            username=resolved.username,
+            password=resolved.password,
+            will=will,
+            ca_cert_path=resolved.ca_cert_path,
+        )
+        await aiomqtt_publisher.connect()
+        mqtt = aiomqtt_publisher
 
     # The emitter Phase 2 reshape pluralised the BESS-config parameter:
     # ``bess_configs`` is a tuple keyed internally by ``instance_id``. The

src/panelbench/emitter_adapter/wire_capture.py

@@ -0,0 +1,88 @@
+"""Capture the full retained surface a consumer sees, not just declarations.
+
+The conformance checker works from `$description` documents, because conformance
+is a question about what a device declares. That is not enough to exercise a
+*consumer*: a parser fed only descriptions can be asked whether it understands
+the shape of a panel, never whether it builds the right snapshot from one — which
+is the part that reaches a user.
+
+So this captures descriptions, `$state`, and every property value, keyed the way
+a consumer receives them.
+
+It lives in the package rather than in `scripts/` so it can be imported by a test
+that checks the committed capture still matches what the emitter emits. A capture
+stranded in a script is one nobody notices going stale.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from panelbench.emitter_adapter import runtime as emitter_runtime
+from panelbench.engine import DynamicSimulationEngine
+
+if TYPE_CHECKING:
+    import pathlib
+
+# Homie topics are `ebus/<version>/<device-id>/<rest...>`; the device id is the
+# third segment and everything after it is the key a consumer sees.
+_DEVICE_SEGMENT = 2
+_MIN_SEGMENTS = 4
+
+
+class RecordingPublisher:
+    """Satisfies `MqttPublisher`, keeping the last payload seen per topic.
+
+    Last-wins rather than an append-only log, because that is what a broker's
+    retained store holds and therefore what a consumer replays on connect. A
+    value corrected within a single tick should leave only the correction here.
+    """
+
+    def __init__(self) -> None:
+        self.retained: dict[str, bytes] = {}
+
+    def is_connected(self) -> bool:
+        return True
+
+    async def publish(
+        self, topic: str, payload: bytes, qos: int = 0, retain: bool = False
+    ) -> None:
+        del qos, retain
+        self.retained[topic] = payload
+
+    async def subscribe(self, topic: str) -> None:
+        del topic
+
+    async def disconnect(self) -> None:
+        return None
+
+
+def as_capture(retained: dict[str, bytes]) -> dict[str, dict[str, str]]:
+    """Regroup flat topics into the device-keyed shape a consumer sees."""
+    devices: dict[str, dict[str, str]] = {}
+    for topic, payload in sorted(retained.items()):
+        parts = topic.split("/")
+        if len(parts) < _MIN_SEGMENTS:
+            continue
+        devices.setdefault(parts[_DEVICE_SEGMENT], {})["/".join(parts[_DEVICE_SEGMENT + 1 :])] = (
+            payload.decode()
+        )
+    return devices
+
+
+async def capture(config: pathlib.Path) -> dict[str, dict[str, str]]:
+    """Run one panel through the real assembly and return what it published.
+
+    Goes through `start_clone` with the MQTT client substituted, rather than
+    reassembling the emitter here: a capture taken through different wiring than
+    a real panel uses proves less than it appears to.
+    """
+    engine = DynamicSimulationEngine(config_path=config)
+    await engine.initialize_async()
+
+    recorder = RecordingPublisher()
+    runtime = await emitter_runtime.start_clone(engine, publisher=recorder)
+    # start() publishes the tree and its descriptions; one tick fills in values.
+    await emitter_runtime.publish_tick(runtime)
+
+    return as_capture(recorder.retained)