OM-M1.3 (protocol half): a persistent broadcast subscription

CLAUDE.md

@@ -18,6 +18,7 @@ Read `docs/STEERING.md` before non-trivial work (new operations, transport or fr
 - Grid mutations use the row/column-keyed pattern (`set_param` / `set_bypass`) - never extend the wholesale `write_preset` path.
 - Docstrings state their evidence: confirmed on hardware vs inferred from the schema. When you verify something on hardware, record it (docstring + coverage table) in the same change.
 - Code in the RX path preserves "the RX thread never dies": wrap every decode, skip unknown types at debug level, reset the reassembly buffer on anything malformed.
+- A `Transport.add_listener` listener runs ON the RX thread (ADR-0009). It applies what the push carries, notes what needs re-reading, and returns. It never reads from the device - `request`, `await_broadcast` and `collect` refuse to run on that thread, and that refusal is not to be relaxed for convenience. Anything registering a listener that must see the connect handshake's burst registers it through `protocol.connect(before_handshake=...)`; by the time `connect()` returns, the burst is still seconds away.
 - Hardware sessions: quit Cortex Control first - it holds the HID interface exclusively.
 - Describe the protocol work as documenting the device's protocol as-is (recovered schema, observed traffic). Do not call it "reverse engineering" in docs, comments, commit messages, or issues.
 - Changed code under a path listed in `docs/STEERING.md` § Owned Paths? Diff and update STEERING/CLAUDE/ADR in the same PR.

changelog.md

@@ -85,6 +85,52 @@ To use both layers in one script, wrap a connection you already have with
 `Device.from_client(qc)`. It does not take ownership: closing the `Device` leaves
 your connection open.
 
+### New: listen to everything the unit sends
+
+The unit talks without being asked. Turn a knob on its touchscreen, recall a
+preset, let the metronome run, and it pushes messages about it. Until now those
+messages were only reachable if you happened to be waiting for that exact one,
+and anything else was dropped. `add_listener` hands you all of them:
+
+```python
+from pyquadcortex import protocol
+
+def watch(message):
+    print(type(message).__name__)
+
+with protocol.connect() as qc:
+    stop = qc.add_listener(watch)
+    ...
+    stop()                     # or qc.remove_listener(watch)
+```
+
+Your function is called for every message, and it takes nothing away from the
+rest of the library: a call that was waiting for a reply still gets it.
+
+Two rules, because your function runs on the thread that reads from the USB
+device:
+
+- **Do not block in it.** Whatever it does delays the next message being read.
+- **Do not read from the device in it.** That thread is the one that would have to
+  deliver the answer, so the call could never be answered - and the connection
+  would stall behind it for as long as it waited. Rather than let that happen, the
+  library raises `RuntimeError` if you try. Note what you need and read it from
+  your own thread.
+- **Treat the message as read-only.** It is the same object the rest of the
+  library sees, not a copy.
+
+To hear the burst of state the unit sends when a client connects - nearly
+everything it knows, including the preset currently loaded - register before the
+handshake, because it arrives seconds after `connect()` returns:
+
+```python
+with protocol.connect(before_handshake=lambda t: t.add_listener(watch)) as qc:
+    ...
+```
+
+The decision behind the two rules is ADR-0009. This is the groundwork for the
+model keeping itself current without asking twice.
+
 ### Withdrawn: the Tempo menu's MODE is "not on the wire"
 
 The 0.23.0 entry below records, under **Settled**, that the Tempo menu's MODE

docs/ADR.md

@@ -110,3 +110,21 @@ Records are append-only once `Decided` and built upon: a shipped decision is nev
   - `compile_protos.sh` generates into a temporary directory and installs into the package only after the check passes, so a refusal leaves the tree exactly as it was.
   - The pin floor and the committed gencode must be equal, not merely compatible. A floor above the gencode still imports for every user, which is precisely the drift ADR-0001 exists to prevent, so the test treats it as a failure rather than a curiosity.
   - What CI proves is the bindings-to-pin half. Nothing machine-checks the `grpcio-tools` floor itself, because deciding whether a floor is high enough means running that generator, and the offline suite has none. A gencode bump that updates the bindings and the pin but forgets the floor therefore leaves CI green. The script is what catches it, one regeneration later, and its refusal names the stale floor as a cause - so the residual exposure is a delay, not a silent pass.
+
+## ADR-0009: Persistent listeners run on the RX thread, which may not read from the device
+
+- **Status:** Decided (2026-08-12)
+- **Decision:** A persistent inbound subscription (`Transport.add_listener`) is called on the transport's RX thread, synchronously, before the message reaches any collector or waiter. The transport enforces the other half of that bargain: `request`, `await_broadcast` and `collect` raise `RuntimeError` when called from the RX thread, so a listener cannot read from the device. A listener that raises is logged and skipped, costing its peers and the message's waiter nothing.
+- **Context:** The three existing inbound hooks are one-shot and scoped to a trigger, so a message nobody is expecting is dropped. A push-fed cache (`docs/domain-model.md` section 9) needs the opposite: every decoded message, for the life of the connection. Where that callback runs is the whole decision, because the RX thread is the one thread in this library that must never block and never die - a wedged read loop takes the connection with it, and every request outstanding.
+- **Options:**
+  - **(a) Call listeners on the RX thread, and forbid reads from it - chosen.** Cheapest, and it keeps arrival order exact. It also gives the ordering the cache design assumes: a listener has already applied a push before the caller that provoked it wakes up, so no caller can observe a reply that its own cache has not seen.
+  - **(b) A delivery thread with a queue.** The RX thread would be immune to a slow listener, at the cost of a thread, an unbounded queue, and a cache that lags the reply that fed it. It buys immunity from a listener that blocks while making one that blocks harmless enough to survive unnoticed.
+  - **(c) Call listeners on the RX thread and document the rules without enforcing them.** The rule that matters ("do not read from the device here") is invisible when broken: the read appears to work, then times out, having stopped the read loop for the whole timeout. This project's oldest lesson is that a failure which looks like success is the expensive kind.
+- **Open Questions:** Whether a listener should be notified when the device is lost. Today it simply stops receiving, which is enough for a cache whose owner learns about loss from the exception on its next call. Reconnect (issue #15) is where this gets answered.
+- **Rationale:** The design already says the RX thread applies pushes and notes what needs re-reading while the caller's thread does the re-reading. Option (a) is that sentence in code. The enforcement costs one thread-identity comparison per call and converts a silent, connection-wide stall into an immediate error naming the rule - and it can refuse nothing that ever worked, because a wait issued from the thread that delivers replies can never be satisfied. `request` and `await_broadcast` would time out; `collect` would return empty, having stalled the link for its full duration.
+- **Consequences:**
+  - A listener must return promptly. Anything expensive belongs on the caller's thread, reached by noting what needs doing rather than doing it.
+  - Nothing may weaken the refusal to keep a convenience. A future listener that wants a value it did not receive marks it for re-reading; it does not fetch it.
+  - `send` is deliberately not refused: it is fire-and-forget and cannot deadlock. A listener that writes owns the delay it adds to the read loop.
+  - Registering in time for the connect handshake's burst needs `protocol.connect(before_handshake=...)`, because the burst arrives after `connect()` returns.
+  - Existing behaviour is untouched: listeners consume nothing, and `send`/`request`/`collect`/`await_broadcast` answer exactly as they did with no listener registered.

docs/STEERING.md

@@ -37,7 +37,7 @@ Inside the protocol layer, a strict one-concern-per-file layering: `cli` → `se
 
 ### Data and state
 
-The protocol layer is stateless between calls: every read is a live exchange, and the unit is the source of truth. The model layer (design in [`domain-model.md`](domain-model.md)) introduces a broadcast-fed write-through cache above `protocol/client.py`; at the time of writing the model is a skeleton and that cache is not built, so callers still hold whatever state they need.
+The protocol layer is stateless between calls: every read is a live exchange, and the unit is the source of truth. It does carry one hook for a caller who wants to be told rather than to ask - `Transport.add_listener`, a subscription that sees every message the unit pushes for the life of the connection (ADR-0009) - but the transport stores none of it. The model layer (design in [`domain-model.md`](domain-model.md)) introduces a broadcast-fed write-through cache above `protocol/client.py`; at the time of writing the model is a skeleton and that cache is not built, so callers still hold whatever state they need.
 
 ## 4. Owned Paths
 
@@ -81,6 +81,7 @@ Decisions for this area are recorded in [`ADR.md`](ADR.md):
 | ADR-0006 | The domain model takes the top-level namespace; the protocol layer moves to `pyquadcortex.protocol` |
 | ADR-0007 | The model may represent a control whose wire path is still open |
 | ADR-0008 | The generator floor joins the bindings/pin unit, with a gate at regeneration and a CI check on the pin |
+| ADR-0009 | Persistent listeners run on the RX thread, which may not read from the device |
 
 ## 8. Open Questions
 
@@ -120,6 +121,32 @@ Single-device, single-connection USB HID at interactive rates (129-byte reports)
 
 ## Change Log
 
+### 2026-08-12 - A persistent broadcast subscription at the protocol layer, and ADR-0009
+
+**What changed:**
+- `Transport.add_listener` / `remove_listener`: a subscription that sees every decoded inbound message for the life of the connection, including the unsolicited pushes `_dispatch` used to drop for want of a waiter. `QuadCortex` passes both through so the layer above never reaches into `_t`
+- The transport now refuses `request`, `await_broadcast` and `collect` when they are called from the RX thread. That is what makes "a listener never reads from the device" enforced rather than requested
+- `protocol.connect(before_handshake=...)` calls back with the started transport before the handshake runs, which is the only moment early enough to hear the handshake's own state burst
+- ADR.md: ADR-0009 - listeners run on the RX thread, and the RX thread may not read; the queue-and-delivery-thread alternative and the document-but-do-not-enforce alternative are recorded with why each was rejected
+- Section 3's "Data and state" names the one hook that is not a live exchange; section 7's table gained the ADR-0009 row
+- `docs/protocol.md` "Connect burst, measured" gained the fact that decided the hook: `connect()` returns at 2.0 s, the ModelRepo lands at 4.9 s and the seed preset at 10.1 s, so a listener attached to the returned client has missed the burst it wanted
+- `tests/hardware/` gained `test_broadcast_listener.py`, and the suite's connection fixture now records the burst - it cannot be attached on demand later, because the burst happens during `connect()`
+
+**Why:**
+- M1 Epic (stokes-audio/pyquadcortex#8), Story #11. This is the protocol-layer half of that story, carved out because it is independent of the model work: `docs/domain-model.md` section 9 needs a push-fed cache, and a cache cannot be fed by three hooks that are all one-shot and scoped to a trigger
+
+**Scope of impact:**
+- **Updated:** `pyquadcortex/protocol/transport.py`, `client.py`, `session.py`, `tests/test_transport.py`, `tests/test_client.py`, `tests/test_session.py`, `tests/test_handshake_burst_recorder.py` (new), `tests/hardware/conftest.py`, `tests/hardware/test_broadcast_listener.py` (new), `tests/hardware/readme.md`, ADR.md, CLAUDE.md, STEERING.md, architecture.md, api.md, protocol.md, changelog.md
+- **Not updated (intentionally):** ADR-0002 - the offline suite still imports no `hid` and the new tests run against `FakeHid` like the rest; ADR-0005 - the new hardware tests only listen, so they write nothing and have nothing to restore, which meets the contract rather than changing it; `docs/domain-model.md` - section 9 designed this and needed no correction; the coverage table in `protocol.md` - no new message type is involved
+
+**Also in this branch:**
+- Merged main (PR #19) in. That change took ADR-0008 for the generator floor, so the listener record is ADR-0009; the two commit messages on this branch predate the renumber and still say 0008
+
+**Downstream to consider:**
+- The model-side cache (the other half of #11) is the intended consumer and is being written separately. It registers through `before_handshake` so the burst warms it for free
+- `tests/hardware/test_write_echo.py` still taps `Transport._dispatch` by monkeypatching it, which predates this and could now be an ordinary listener. Left alone deliberately: it is a working measurement harness, and `tests/test_scene_echo_predicates.py` imports it offline
+- ADR-0009 leaves one question open on purpose - whether a listener hears about device loss. It stops receiving today, and the answer belongs with reconnect (#15)
+
 ### 2026-08-12 - The generator floor joins the bindings/pin unit (ADR-0008)
 
 **What changed:**

docs/api.md

@@ -60,6 +60,7 @@ already read and need no connection; calling them as methods raises
 | **Per-beat accents** | `set_beat(n, MetronomeBeat.ACCENT)`, `set_beats([...])`, `protocol.beats(preset)` |
 | **Inspect a preset** (module functions) | `protocol.blocks(preset)`, `protocol.splits(preset)`, `protocol.free_rows(preset)`, `protocol.row_status(preset)`, `protocol.bypass_state(preset, row, column)`, `protocol.param_state(preset, row, column, index)`, `protocol.param_options(preset, row, column, index)`, `protocol.input_chain_rows(preset, input)`, `protocol.params_equal(a, b, option_count=)`, `protocol.field_present(msg, field)` |
 | **Wait for the device** | `wait_for_listing(setlist, until=...)` |
+| **Watch what the unit pushes** | `add_listener(fn)`, `remove_listener(fn)` - your `fn` is called with every message the unit sends, asked for or not. It runs on the transport's read thread, so it must not block and may not read from the device. To catch the connect handshake's own burst of state, register before it with `protocol.connect(before_handshake=...)` |
 | **Scenes** | `copy_scene(from_scene, to_scene, swap=False)`, `set_scene_label(scene, label)`, `set_scene_color(scene, argb)` |
 | **Global settings** | `settings()`, `update_settings(**fields)`, `set_scene_bypass_behavior()`, `set_global_bypass()`, `set_master_volume_assignment()`, `mode()`, `set_mode()`, `set_mode_cycle()`, `set_gig_view()` |
 | **Global EQ** | `global_eq()`, `set_global_eq(band, gain=, frequency=, q=, filter_type=, enabled=)`, `set_global_eq_output(level=, out12=, out34=)`, `set_global_eq_bypassed()` |

docs/architecture.md

@@ -106,6 +106,10 @@ concurrent:
   gunzips frame-level compressed payloads, parses the protobuf, and dispatches;
 - correlation: `request()` waiters keyed by `request_id`, plus
   `await_broadcast()` waiters keyed by message class and an optional predicate;
+- persistent subscriptions: `add_listener()` registers a callable that sees every
+  decoded message for as long as the connection lasts, including the unsolicited
+  pushes no waiter is expecting. It consumes nothing, so waiters and collectors
+  behave exactly as they do with no listener registered;
 - a keepalive thread;
 - tolerating the device's benign write STALL (see
   [protocol.md](protocol.md#the-benign-write-stall)): write errors are logged at
@@ -117,6 +121,16 @@ types and non-protobuf pushes are skipped at debug level, and the reassembly
 buffer is reset on anything malformed so one bad frame cannot wedge the stream.
 If you add code to the RX path, preserve that property.
 
+Listeners run on that thread, so the same rule covers them: one that raises is
+logged and skipped, its peers still get the message, and the message still
+reaches its waiter. A listener may also not read from the device -
+`request`, `await_broadcast` and `collect` raise `RuntimeError` when called from
+the RX thread, because the RX thread is the one that would have to deliver the
+answer, so such a call could only ever time out with the read loop stopped behind
+it. A listener applies what a push carries and notes what needs re-reading; the
+caller's thread does the re-reading (see [domain-model.md](domain-model.md)
+section 9, and ADR-0009).
+
 ### registry.py
 
 The only place that knows the mapping between the schema's
@@ -153,6 +167,14 @@ remembers what it opened (`_owned_resources`) so `close()` and the context
 manager tear down only what `connect()` created. A client built around a
 caller-supplied transport owns nothing and `close()` is a no-op.
 
+`connect(before_handshake=...)` is the hook for anything that has to be watching
+before the handshake runs. The subscription burst the handshake sends is what
+makes the unit start pushing state, and that state arrives AFTER `connect()` has
+returned (measured: the client comes back at 2 s, the ModelRepo lands at 4.9 s and
+the current preset at 10.1 s - see [protocol.md](protocol.md), "Connect burst,
+measured"). So a listener registered on the returned client has already missed it;
+one registered through this hook has not.
+
 `import hid` lives *inside* `open_device()`. That laziness is a contract, not an
 accident: see [Testing philosophy](#testing-philosophy).
 
@@ -200,18 +222,22 @@ device.read() -> one 129-byte input report
   -> framing.decode_reports(buffer) -> (message_type, payload)
   -> gunzip payload if it starts 1f 8b
   -> registry.class_for(message_type) -> parse
-  -> _dispatch: a request_id waiter, else a broadcast waiter, else dropped
+  -> _dispatch: every listener, then collectors, then a request_id waiter,
+     else a broadcast waiter, else dropped
 ```
 
 ## send vs request vs await_broadcast
 
-Choosing correctly is most of the work of adding an operation.
+Choosing correctly is most of the work of adding an operation. The first three
+rows serve ONE exchange, which is what an operation needs. The last one is not an
+operation at all: it is how a long-lived caller watches the link.
 
 | Transport method | Use when | Blocking | Correlation |
 |---|---|---|---|
 | `send(msg)` | The device acts on the message and you do not need its answer: scene switch, grid edits, recall, keepalive. | No | None |
 | `request(msg, timeout=)` | The device answers a message of the **same type**: `Version` READ, `ResetCommsBuffers`, the `File` mutations. | Yes | Fresh `request_id` is assigned and registered before the write. Reply is the first inbound message of the same type whose `request_id`, if present on both sides, matches. |
 | `await_broadcast(cls, trigger, timeout=, match=)` | The answer arrives as a **push of a different type**, or as an unsolicited broadcast the device emits in response to an action: the `RecallPreset` push that carries a full preset, the `File` folder listings. | Yes | By message class, plus your optional `match` predicate. A right-type message the predicate rejects is left undelivered so a later one can satisfy the waiter. |
+| `add_listener(fn)` | You want EVERY message for the life of the connection, not the answer to one call: a cache fed by the unit's own pushes, or a log of the link. | No, but `fn` runs on the RX thread | None. Every message, every type, whether or not a waiter also gets it. Removed with the returned callable or `remove_listener(fn)`. |
 
 Two gotchas the current code already encodes, and that new operations must
 respect: