diff --git a/plans/opencode/typed-ems-messaging.md b/plans/opencode/typed-ems-messaging.md new file mode 100644 index 00000000..4d574ccd --- /dev/null +++ b/plans/opencode/typed-ems-messaging.md @@ -0,0 +1,821 @@ +# Typed EMS and Brokerd Messaging + +## Purpose + +Replace EMS's dictionary-shaped actor traffic with explicit, tagged +`msgspec.Struct` protocols and modern Tractor payload enforcement on both +actor boundaries: + +```text +order client <-> emsd <-> brokerd. / paperboi. +``` + +The broker boundary is part of the formal protocol, not an implementation +detail behind the EMS endpoint. Paper, Binance, Kraken, and IB must all emit +the same startup type, consume the same request union, and produce the same +event union before strict Tractor payload limits are enabled. + +This plan deliberately normalizes schemas and backend behavior before adding +`pld_spec`. Enabling transport validation first would only turn existing +status drift, startup-shape drift, and raw dictionary reconstruction into +hard actor failures without first defining the intended behavior. + +## Scope + +This work includes: + +- typed client-to-EMS requests and EMS-to-client events; +- typed EMS-to-broker requests and broker-to-EMS events; +- one typed broker startup payload shared by live and paper endpoints; +- one typed EMS startup payload; +- stable tagged unions suitable for native `msgspec` decoding; +- normalized status, identifier, startup, and metadata semantics; +- per-direction `Context.pld_rx.limit_plds()` enforcement; +- backend-neutral conformance tests for paper, Binance, Kraken, and IB; +- deterministic actor-free tests and real Tractor actor/IPC tests; +- bounded handling of acknowledgement, status, cancellation, and teardown + races; +- removal of raw dictionary dispatch after all producers have migrated. + +This work does not include: + +- typing provider-native websocket or REST payloads inside each backend; +- changing the datad/brokerd actor split; +- spawning brokerd from any location other than `open_brokerd_dialog()`; +- redesigning accounting or replacing `BrokerdPosition` with the full + accounting `Position` type in the same patch series; +- live order submission in CI; +- preserving undocumented dictionary IPC between actors running different + Piker revisions. The actor tree is treated as a same-checkout deployment. + +## Current Contract Inventory + +### Client and EMS boundary + +| Location | Startup payload | Stream received | Stream sent | Current issue | +| --- | --- | --- | --- | --- | +| `_emsd_main()` | tuple of positions map, accounts list, dialogs map | dictionaries reconstructed as `Order` or `Cancel` | `Status` and `BrokerdPosition`, often decoded as dictionaries by clients | no payload spec, tuple startup, raw matching, loose nested fields | +| `open_ems()` / `OrderClient` | destructures tuple | `Status` and `BrokerdPosition` dictionaries | annotated as `Order | dict`, but implementation assumes struct attributes | public shape and wire shape are conflated | + +### EMS and broker boundary + +| Endpoint | Startup payload | Stream received by backend | Stream sent by backend | Current issue | +| --- | --- | --- | --- | --- | +| paper `open_trade_dialog()` | `(list[BrokerdPosition], ['paper'])` | raw `BrokerdOrder` / `BrokerdCancel` dictionaries | ack, status, fill, error, position structs | raw dispatch and network-dependent simulator setup | +| Binance `open_trade_dialog()` | `'paper'` sentinel or `(positions, accounts)` | raw request dictionaries | structs plus client-side `Status` for existing orders | mixed protocol layers and `fill` status drift | +| Kraken `open_trade_dialog()` | `'paper'` sentinel or `(positions, accounts)` | raw request dictionaries | structs plus client-side `Status` for existing orders | `filled` status drift and race workarounds | +| IB `open_trade_dialog()` | `(positions, tuple(accounts))` | raw request dictionaries | structs plus client-side `Status` for existing orders | status can race ack and internal event types leak into sequencing | + +The shared broker boundary already exists operationally: all four endpoints +are long-lived Tractor contexts opened by `open_brokerd_dialog()`. The upgrade +formalizes that de facto interface instead of allowing each backend to define +its own wire dialect. + +## Target Protocol + +### Tagging and decoding policy + +Every union member must have an explicit stable tag. Use one common tag field, +preferably `name`, with protocol-level values such as `order`, `cancel`, +`ack`, `status`, `fill`, `error`, `position`, and `open_order`. Do not rely on +Python class names as wire tags. + +All newly formalized wire structs should use: + +- `tag=''` and `tag_field='name'`; +- `frozen=True` for wire snapshots; +- `kw_only=True` where it improves constructor safety; +- factory-based mutable defaults; +- finite `Literal` values for protocol enums; +- `forbid_unknown_fields=True` only after the compatibility-producing phase + has been removed and every backend passes conformance. + +The tag is the dispatch discriminator. Fields such as `action`, `status`, and +`resp` retain only domain meaning and are not overloaded as message type tags. +Mutable EMS dialog bookkeeping must use a separate internal state type. Do not +make a wire struct mutable merely because current code mutates and re-sends the +same `Status` or `Order` instance. + +Define shared scalar aliases and use them consistently: + +```python +RequestId = int | str +FillId = str +``` + +### Client-to-EMS requests + +Define: + +```python +EmsRequest = Order | Cancel +``` + +Normalize the request structs as follows: + +- `Order.action` is exactly `buy | sell | alert`. +- `Order.exec_mode` is exactly `dark | live`. +- `Order.oid`, `symbol`, and `account` remain required strings. +- `Order.price` remains `Decimal`; tests must prove transport round-tripping. +- `Order.size` remains signed `float` for this migration. Resolving the + duplicated side/action representation is a later domain change. +- `Cancel` has its own tag; `action='cancel'` can be removed once no local UI + matching depends on it. +- Actor IPC accepts structs only. Any user-facing convenience constructor for + mappings must validate and construct a struct before entering the stream. + +### EMS-to-client events + +Define: + +```python +EmsEvent = Status | BrokerdPosition +``` + +Retain `BrokerdPosition` as the wire-level position summary for this series, +but document it as shared by the broker and client event protocols. A later +accounting patch may rename it to `PositionSummary` without coupling that +change to IPC enforcement. + +Normalize `Status` so that: + +- `resp` is exactly `pending | open | dark_open | triggered | fill | closed | + canceled | error`; +- `req` is `Order | None`, never an unvalidated nested dictionary; +- `brokerd_request` is `BrokerdRequest | None` and records the command that + caused the status transition; +- `brokerd_event` is `BrokerdEvent | None` and records the backend response; +- the current mixed-purpose `brokerd_msg` field is removed after callers have + migrated; neither replacement accepts an untyped dictionary; +- an error is represented by `Status(resp='error', ...)` or by a distinct + tagged client event, not the current incomplete `Error(Status)` subclass; +- internal mutable dialog state is not the same object sent repeatedly over + the wire. Build a status snapshot when broadcasting so prior history and + multi-client delivery cannot be changed by later in-place mutation. + +### EMS-to-broker requests + +Define: + +```python +BrokerdRequest = BrokerdOrder | BrokerdCancel +``` + +Normalize both variants so that: + +- they have explicit `order` and `cancel` protocol tags independent of + `action`; +- `oid` always identifies the EMS dialog; +- `reqid` is `RequestId | None` on requests because allocation policy differs + by backend; +- account and symbol validation failures produce a typed `BrokerdError` + rather than logging and continuing silently; +- updates remain `BrokerdOrder` with an existing `reqid`; a separate modify + variant is deferred unless conformance tests prove the current distinction + ambiguous. + +### Broker-to-EMS events + +Define: + +```python +BrokerdEvent = ( + BrokerdOrderAck + | BrokerdStatus + | BrokerdFill + | BrokerdError + | BrokerdPosition + | BrokerdOpenOrder + | BrokerdOpenOrdersComplete +) +``` + +`BrokerdOpenOrder` replaces the current backend-to-EMS use of client-facing +`Status`. It carries the provider request id, account, source backend, and a +fully typed embedded `Order`. EMS alone translates this event into a client +`Status` and registers it in the dialog book. + +Require `oid` on ack and on all events produced for Piker-originated orders. +For an externally originated order, the backend adapter creates a +collision-safe stable `oid` from broker, account, and provider `reqid` before +emitting `BrokerdOpenOrder`; its embedded `Order.oid` is the same value. Later +status/fill events reuse that identity. EMS validates and registers it but does +not allocate a new id requiring an unmodelled reverse message. Service-level +errors that cannot name a dialog may keep `oid=None`. + +`BrokerdFill` has a required `fill_id: FillId`. Use the provider execution id +or provider event sequence. The four current implementations must use Binance +trade id, Kraken transaction id, IB execution id, and a distinct paper-ledger +transaction id respectively. Price/size/time are not a safe deduplication key +because two legitimate executions may share them. A future provider with no +execution identity must define an adapter session-id plus monotonic sequence +and document that deduplication does not survive brokerd restart; it may not +silently synthesize identity from equal-valued fill fields. + +Use one broker status vocabulary: + +```python +BrokerdOrderState = Literal[ + 'pending', + 'open', + 'canceled', + 'closed', +] +``` + +Fills are events, not status values. A partial execution emits a +`BrokerdFill` and may then emit `BrokerdStatus(status='open', ...)`. A complete +execution emits the final `BrokerdFill` followed by +`BrokerdStatus(status='closed', remaining=0)`. Binance `fill`, Kraken +`filled`, and IB provider status names must be translated inside their +backend adapters and must never cross the brokerd IPC boundary. + +### Startup payloads + +Replace every tuple and string sentinel with tagged startup structs. + +`BrokerdStartup` contains: + +- `mode: Literal['live', 'paper', 'paper_fallback']`; +- `broker: str`; +- `accounts: tuple[str, ...]`; +- `positions: tuple[BrokerdPosition, ...]`. + +A live endpoint without credentials returns +`BrokerdStartup(mode='paper_fallback', broker=..., accounts=(), positions=())`. +`open_brokerd_dialog()` then opens the paper endpoint, which returns the same +type with `mode='paper'`, its paper account, and persisted positions. The +fallback value is a typed delegation result, not an active broker snapshot: it +is exempt from account/position and open-order-completion requirements, and +its context returns after startup. This removes the raw `'paper'` union branch +without hiding fallback behavior. + +Existing open orders remain `BrokerdOpenOrder` stream events because some +providers discover them only after authenticated stream subscription. The +contract must state that startup positions/accounts are a complete snapshot, +while open-order synchronization completes through typed stream events. Every +active live or paper backend emits exactly one `BrokerdOpenOrdersComplete` +after the initial set, including when it found zero orders, so EMS can +distinguish completion from delay and can order initial client readiness +deterministically. A `paper_fallback` delegation does not open a stream and +does not emit this event; the subsequently opened paper endpoint does. + +`EmsStartup` contains: + +- `positions: tuple[BrokerdPosition, ...]`; +- `accounts: tuple[str, ...]`; +- `dialogs: tuple[Status, ...]`. + +`open_ems()` may continue returning the existing nested lookup maps to callers, +but it builds them locally from `EmsStartup`. The public convenience shape is +therefore separated from the wire contract. + +Treat `BrokerdStartup` as the brokerd equivalent of datad's `FeedInit`, but do +not reproduce `validate_backend()`'s long-lived dictionary compatibility +branches. Add a broker initialization validator beside the clearing protocol +that checks backend name, unique/non-empty accounts, position account +membership, `position.broker == startup.broker`, unique +`(broker, account, symbol)` position keys, and mode-specific requirements. +Give malformed startup its own registered initialization error so failures +retain a useful remote exception type before strict Tractor decoding takes +over. Each account name must be non-empty; the accounts collection may be +empty only for `paper_fallback`. + +Replace `open_brokerd_dialog()`'s manual probing with one broker order-control +endpoint resolver. It may use `get_eps(mod, 'brokerd')` as input, but it must +select exactly one of `open_trade_dialog` or the temporary legacy +`trades_dialogue` alias and must not confuse unrelated `norm_trade` support +with a trading dialog. The runtime validator proves startup; the conformance +suite proves request/event behavior. Neither mechanism should import or enable +datad modules inside the credentialed broker actor. + +### Payload aliases and Tractor limits + +The endpoint-wide payload aliases include startup, stream traffic, and the +final `None` result: + +```python +EmsPayload = EmsStartup | EmsRequest | EmsEvent | None +BrokerdPayload = BrokerdStartup | BrokerdRequest | BrokerdEvent | None +``` + +Apply broad endpoint specs and narrow receive-side specs as follows: + +| Actor side | Endpoint `pld_spec` | Narrow receive spec | +| --- | --- | --- | +| emsd `_emsd_main()` | `EmsPayload` | `EmsRequest` around the client command loop | +| order client | inherited from endpoint | `EmsEvent` around trade event receives | +| backend/paper `open_trade_dialog()` | `BrokerdPayload` | `BrokerdRequest` around backend request loops | +| emsd broker relay | inherited from endpoint | `BrokerdEvent` around broker event receives | + +The endpoint payload spec supplies the broad decoder used for `Started`, stream, +and return payloads. `ctx.started(..., validate_pld_spec=True)` can also perform +local started-value validation. Ordinary `MsgStream.send()` does not validate +locally: malformed stream data raises `tractor.MsgTypeError` when the peer +receives/decodes it. `ctx.pld_rx.limit_plds(spec=...)` narrows that receiving +direction so a valid message from the wrong direction still fails there. + +Do not add either decorator spec until the corresponding boundary has no +remaining legacy sender. The final enforcement must follow the proven pattern +in `remote_gap_overlays()` and Tractor's own payload-limiting tests. + +### Provider metadata + +Core routing and state fields must be typed. Provider-native diagnostics may +remain behind one explicit extension point, for example a +`BrokerEventMeta(source, broker_time, raw)` struct where `raw` is +`dict[str, Any]`. + +No EMS decision may depend on keys hidden in `raw`. Fields currently used by +EMS, including source backend, broker timestamp, symbol, account, `oid`, and +`reqid`, must move into typed top-level fields. Tests should demonstrate that +malformed extension data cannot alter routing or dialog state. + +## Ordering and State Semantics + +Typed messages alone do not repair ordering defects. Define these invariants +before enforcement: + +1. A backend sends exactly one `BrokerdOrderAck` for each accepted new order. +2. The ack establishes `oid <-> reqid` before status/fill events are exposed + to EMS whenever the provider API permits that ordering. +3. A backend adapter that receives provider events before it can send the ack + buffers them by `reqid` and flushes them after ack. +4. EMS keeps a bounded defensive pre-ack buffer because actor task scheduling + can still expose ordering bugs. The buffer has a size/age limit and turns + expiry into a typed error plus a diagnostic, never silent message loss. +5. A cancel received before ack marks the dialog `cancel_called`; the first ack + immediately sends a typed `BrokerdCancel` exactly once. +6. Duplicate ack, fill, and terminal status events are idempotent or rejected + deterministically. Fill deduplication uses `fill_id`. They never resurrect + a closed dialog. +7. `closed` and `canceled` are terminal. A late fill may be recorded and + reported according to an explicit test, but it cannot recreate mutable + active state. +8. Existing external orders enter through `BrokerdOpenOrder` with the backend's + stable namespaced `oid` and are registered before broadcast. +9. Initial open-order discovery is complete only after + `BrokerdOpenOrdersComplete`; client readiness cannot infer completion from + an empty or temporarily quiet stream. +10. Order status is routed by dialog and FQME. Account authorization and + position privacy remain a separately scoped subscription design because + `_emsd_main()` currently receives no account authorization input. Tests in + this series document the current position broadcast and must not claim + unauthorized-account isolation. + +## Implementation Phases + +### Phase 0: Characterize the Existing Boundary + +Add tests that record current intended behavior before restructuring: + +- new order, update, cancel, ack, open, partial fill, close, and position flow; +- cancel-before-ack; +- status-before-ack; +- existing provider order discovery; +- bad account, zero size, and backend submission error as typed domain errors; +- bad action/tag/field type as protocol decode failures, not `BrokerdError`; +- startup with live credentials, no credentials, paper account, persisted + positions, and no positions; +- Binance startup containing a negative/short position; +- IB startup containing at least two accounts with distinct contracts; +- zero, one, and multiple existing orders followed by the explicit snapshot + completion event; +- disconnect while requests or events are in flight. + +Use fake streams and fake provider clients; do not require public market data +or credentials. Capture known undesirable outcomes as explicit expected +failures only when a same-phase fix would obscure the baseline. + +Gate: every behavior changed later has a deterministic assertion describing +the old behavior and the target behavior. + +### Phase 1: Define and Audit the Wire Schemas + +Refactor `piker/clearing/_messages.py` into a protocol-only module or keep that +file as the single protocol module if the resulting dependency order remains +clear. Add the tagged structs and aliases described above without enabling +Tractor limits. + +Replace mutable literal defaults with factories. Resolve the accidental tuple +default on `BrokerdStatus.account`. Remove or complete the unusable +`Error.from_status()` path. Add explicit startup types and +`BrokerdOpenOrder`. Add the broker initialization error/validator without a +legacy dictionary fallback. + +Inventory every current mutation of `Order`, `Status`, and embedded requests. +Introduce a mutable internal dialog-state record and construct frozen wire +snapshots from it. Replace mutation-based `Struct.copy(update=...)` usage with +new-instance replacement that is proven to work for frozen structs. + +Add actor-free serialization tests using the same msgpack semantics as +Tractor: + +- each concrete type round-trips with exact type equality; +- each union decodes to the correct concrete type by tag; +- nested `Order`, broker event, position, `Decimal`, optional id, and metadata + fields round-trip; +- unknown tag, missing required field, wrong literal, wrong primitive type, + malformed nested struct, and duplicate/ambiguous tag fail; +- strict unknown-field behavior is tested separately and enabled only at the + strictness gate; +- request and event aliases do not accept each other's members. + +Gate: all protocol unions have unique tags and can be decoded without manual +dictionary inspection. + +### Phase 2: Build a Backend Conformance Harness + +Create reusable conformance helpers that exercise a backend request handler +with a memory-backed stream and fake provider client. The harness supplies the +same typed request cases and validates only typed events. Every request and +emitted event must be encoded and decoded through its declared union before +assertion; `isinstance` checks alone are insufficient because direct +`msgspec.Struct` construction does not validate annotated field types. + +The shared suite must verify: + +- startup is one `BrokerdStartup`, including typed fallback-to-paper; +- malformed startup fails the shared broker initialization validator; +- new order produces one correlated ack; +- update preserves `oid` and `reqid`; +- cancel uses the correlated request id; +- invalid account/size produces `BrokerdError` without killing the dialog; +- invalid action/tag/type is rejected by codec or Tractor before the handler; +- provider statuses normalize to the four broker states; +- partial and complete fills use `BrokerdFill` plus normalized statuses; +- emitted positions and existing orders are typed; +- every active live/paper startup emits one open-order completion event, even + for zero orders; `paper_fallback` delegates without one; +- duplicate fills are rejected by `fill_id`, while distinct equal-valued fills + remain valid; +- provider-native dictionaries never cross the test stream; +- cancellation closes all handler tasks and leaves no blocked sender; +- one malformed request fails at the protocol boundary once limits are + enabled. + +Separate backend-neutral contract assertions from provider translation +fixtures. Provider fixtures should be the smallest recorded websocket/API +payload required to prove normalization. Include regressions for Binance's +tuple-shaped error reason, string fill size, short startup position, and +float-vs-`Decimal` open-order price; Kraken's `filled` status and open-order +price; and IB multi-account startup association. + +Gate: the harness can fail a deliberately nonconforming fake backend for each +contract rule. + +### Phase 3: Port the Paper Engine as the Reference Backend + +Make paper `open_trade_dialog()` the reference implementation of +`BrokerdPayload`: + +- return `BrokerdStartup`; +- consume `BrokerdRequest` instances directly; +- emit only `BrokerdEvent` instances; +- remove `BrokerdOrder(**request_msg)` and dictionary pattern dispatch; +- allocate a distinct deterministic `fill_id`/transaction id instead of + reusing `oid` as `tid`; +- separate its order protocol tests from market-feed simulation; +- allow deterministic synthetic ticks to drive fill tests without a public + feed. + +Do not enable endpoint `pld_spec` yet if live backends share code paths that +still send legacy payloads. + +Gate: the complete broker conformance suite passes for paper without network, +credentials, sleeps, or teardown suppression. + +### Phase 4: Port Live Backend Adapters Individually + +Port one backend per independently reviewable change. A backend is complete +only when its provider translation fixtures and shared conformance suite pass. + +Recommended order: + +1. Binance: replace raw request matching, type startup/fallback, replace + existing-order `Status` with `BrokerdOpenOrder`, and map `fill` to fill plus + `open/closed` events. +2. Kraken: replace raw request matching, type startup/fallback, replace + existing-order `Status`, map `filled` to fill plus `closed`, and formalize + the too-fast edit/cancel behavior. +3. IB: replace raw request indexing, type startup, replace existing-order + `Status`, and put an explicit sequencing gate between asyncio provider + events and ack publication. + +For every backend, core fields are extracted before constructing the event. +The provider's original message may be retained only in the metadata extension +point. Keep brokerd's enabled RPC module set minimal and do not reintroduce +datad modules into the credentialed actor. + +Each adapter must preserve a provider execution id as `fill_id`, emit the +initial open-order completion event, and populate a stable namespaced `oid` for +external orders. Binance must retain negative positions in startup. IB startup +must associate each position with the correct account/client rather than loop +state from the last account. + +Gate: paper, Binance, Kraken, and IB all pass the same offline conformance +contract; no backend emits a dictionary or client-facing `Status` to EMS. + +### Phase 5: Convert EMS Broker Translation and State + +Change `open_brokerd_dialog()` to consume `BrokerdStartup` and return a typed +relay object. Remove startup tuple/string branching and +`BrokerdPosition(**msg)` reconstruction. + +Remove the current `client_ready` dependency from initial broker-event +processing. Start the broker relay as soon as its stream opens, consume and +reduce `BrokerdOpenOrder` events through `BrokerdOpenOrdersComplete`, and only +then construct `EmsStartup` and call `_emsd_main().started()`. The resulting +startup dialogs contain the completed initial snapshot. Events arriving before +the first client stream exists remain in dialog state/history for startup or +later delivery rather than blocking the broker sender or being discarded. + +Split the current large translation loop into a small typed event reducer plus +async effects: + +- dispatch with class patterns, not mapping patterns; +- update correlation/dialog state in a deterministic function or narrowly + scoped state object; +- keep that state mutable internally but emit only new frozen `Status` + snapshots; +- return typed client events and typed broker requests to send; +- keep stream broadcast and logging in the async shell; +- implement bounded pre-ack buffering and terminal-state idempotence; +- translate `BrokerdOpenOrder` to `Status` in one place; +- stop reading routing fields from `broker_details`. + +The reducer must be directly testable with event sequences and no actors. + +Gate: all order-state permutations pass actor-free tests, and +`translate_and_relay_brokerd_events()` contains no struct reconstruction from +dictionaries. + +### Phase 6: Convert the Client Boundary + +Change `process_client_order_cmds()` to receive `EmsRequest` directly and use +class-pattern dispatch. Return `EmsStartup` from `_emsd_main()` and construct +the existing `open_ems()` convenience maps client-side. + +Make `OrderClient.send()` and `send_nowait()` accept `EmsRequest` only. If a +mapping convenience API is required by a known caller, put it in an explicit +constructor that performs local msgspec validation; do not put dictionaries on +the actor stream. Audit UI consumers and tests so receive loops work with +already-decoded `Status` and `BrokerdPosition` instances. + +Replace `OrderClient.update*()` and UI mutations with construction of a new +`Order`. Audit mutations of `Status.req`, `Status.resp`, and embedded orders in +order mode and clearing code; all should update internal state or replace a +wire value, never mutate a frozen payload. + +Gate: there are no `Order(**cmd)`, `Cancel(**cmd)`, `Status(**msg)`, or +`BrokerdPosition(**msg)` conversions on the EMS client boundary. + +### Phase 7: Enable Tractor Payload Enforcement + +Enable one boundary at a time: + +1. Add `BrokerdPayload` to every paper/live `open_trade_dialog()` decorator. +2. Limit backend receive scopes to `BrokerdRequest`. +3. Limit EMS's broker receive scope to `BrokerdEvent`. +4. Add `EmsPayload` to `_emsd_main()`. +5. Limit EMS's client receive scope to `EmsRequest`. +6. Limit order-client receive scopes to `EmsEvent`. +7. Enable `forbid_unknown_fields=True` on stable wire structs. + +At each step, test invalid startup, stream, wrong-direction, and return +payloads. Assert where decoding occurs, `tractor.MsgTypeError` provenance, +context cancellation, and bounded teardown rather than assuming every sender +raises locally. Add an actor-free metadata test that each paper/Binance/Kraken/ +IB endpoint's `_tractor_context_meta['pld_spec']` resolves to `BrokerdPayload` +and `_emsd_main()` resolves to `EmsPayload`. + +Gate: no valid wrong-direction type or malformed payload reaches business +logic, and malformed traffic cannot wedge the actor tree. + +### Phase 8: Real Actor and Lifecycle Qualification + +Add test-only Tractor context endpoints in a dedicated test fixture module to +exercise actual encoding, process boundaries, and cancellation. Do not mock +`MsgStream` in this layer. + +The real-actor matrix includes: + +- valid startup and bidirectional exchange for both protocol families; +- exact concrete types observed after remote decode; +- invalid `Started`, stream send, wrong-direction send, and return payload; +- optional local `Started` validation and peer receive-side stream/return + validation; +- child failure before and after `ctx.started()`; +- client cancellation while backend send is blocked; +- backend cancellation while EMS waits for an event; +- simultaneous stream close and terminal status; +- repeated open/close cycles with no stale subscribers, tasks, actors, or SHM; +- two clients with distinct dialogs and no cross-delivery of order status; +- deterministic ack/status and cancel/ack race barriers; +- brokerd restart/reconnect policy, explicitly failing if unsupported rather + than hanging. + +Use Trio events/channels as barriers. Do not use timing sleeps to force races. +Use bounded `fail_after()` scopes around every scenario. Do not use +`run_and_tollerate_cancels()` for these tests; expected cancellation must be +asserted precisely. + +Gate: the offline real-actor suite passes repeatedly under the repository's +supported Tractor spawn methods and fails loudly within a bounded timeout. + +### Phase 9: Remove Compatibility Paths + +After all strict tests pass: + +- delete mapping-pattern IPC dispatch and `Struct(**wire_dict)` conversions; +- delete the `'paper'` startup sentinel and startup tuple handling; +- delete backend emission of client `Status`; +- delete obsolete status aliases such as `fill` and `filled`; +- delete legacy `trades_dialogue` lookup if no backend implements it; +- remove stale dictionary type annotations and compatibility comments; +- remove expected failures introduced in Phase 0; +- update backend author documentation with the protocol aliases and + conformance command. + +Gate: searches for the removed patterns are empty or have an explicit local, +non-IPC justification. + +## Test Layout + +Use focused flat modules consistent with the existing test tree: + +| Test module | Responsibility | Network | +| --- | --- | --- | +| `tests/test_ems_messages.py` | union tags, codec round-trips, invalid schemas, strict fields | never | +| `tests/test_brokerd_protocol.py` | shared backend conformance and provider translation fixtures | never | +| `tests/test_ems_state.py` | reducer transitions, correlation, races, idempotence, routing | never | +| `tests/test_ems_ipc.py` | real Tractor actor payload limits and lifecycle | never | +| `tests/test_ems.py` | public `open_ems()` behavior, persistence, paper integration | deterministic by default | +| backend-specific live qualification | authenticated smoke tests only | opt-in | + +The default CI lane runs every actor-free and offline actor test. Authenticated +or public-network checks use explicit markers and are never the only coverage +for a protocol behavior. + +## Required Test Matrix + +### Schema validation + +- every valid concrete message; +- every startup mode; +- `int` and `str` request identifiers; +- zero, positive, and negative sizes where semantically permitted; +- `Decimal` prices; +- optional and populated metadata; +- unknown/missing/wrong tags; +- unknown fields at the strictness gate; +- wrong literal and primitive field types; +- malformed nested order/event/position; +- a request sent in the event direction and vice versa. + +### Dialog transitions + +- submit -> ack -> pending -> open; +- submit -> ack -> fill -> open -> fill -> closed; +- submit -> ack -> canceled; +- submit -> cancel -> ack -> broker cancel -> canceled; +- submit -> provider error; +- status -> ack reordered delivery; +- fill -> ack reordered delivery; +- duplicate ack; +- duplicate `fill_id` and distinct equal-valued fills; +- duplicate and conflicting terminal statuses; +- late fill after close; +- update before and after ack; +- cancel and update for unknown dialog; +- existing external open order; +- position update with and without an active order. + +### Backend conformance + +- paper, Binance, Kraken, and IB request acceptance; +- account rejection; +- provider status normalization; +- provider exception to `BrokerdError` conversion; +- open-order conversion; +- zero-order and populated open-order snapshot completion; +- startup account/position conversion; +- Binance negative-position startup and IB multi-account association; +- provider metadata isolation; +- backend handler cancellation and stream closure. + +### Actor semantics + +- valid typed startup and streams; +- local `Started` and peer receive-side `MsgTypeError` provenance; +- malformed startup cancels context cleanly; +- malformed stream traffic cancels both ends cleanly; +- caller and callee cancellation; +- no swallowed teardown errors; +- no actor/task/channel leaks; +- repeated execution to expose scheduling-sensitive failures. + +### Routing and isolation + +- event reaches the originating dialog subscribers; +- unrelated FQME subscriber does not receive it; +- current all-client position broadcast is explicitly characterized as a + known limitation pending an account authorization/subscription schema; +- disconnected clients are removed from subscriber/dialog tables; +- reconnect receives only the documented startup snapshot and active dialogs. + +## CI and Qualification Gates + +Run gates in increasing cost order: + +1. schema and reducer unit tests; +2. backend conformance tests with fake provider clients; +3. real Tractor IPC tests; +4. deterministic paper EMS integration; +5. the broader clearing/service regression set; +6. repeated race tests under each supported spawn method; +7. opt-in live read-only startup for Binance, Kraken, and IB; +8. manually authorized live order qualification with minimal size. + +No live credential or public-network result can waive an offline failure. +Never place an order as part of default CI. + +## Commit Boundaries + +Keep the implementation reviewable and bisectable: + +1. characterize current protocol behavior; +2. add tagged schemas and codec tests; +3. add backend conformance harness; +4. port paper backend; +5. port Binance backend; +6. port Kraken backend; +7. port IB backend; +8. refactor EMS broker-event state reducer; +9. type the client/EMS boundary; +10. enable broker payload enforcement; +11. enable client payload enforcement; +12. add lifecycle/race/isolation qualification; +13. remove compatibility paths and document backend requirements. + +Every commit must leave the deterministic tests green. Do not combine schema +normalization, backend provider behavior changes, and strict Tractor +enforcement in one commit. + +## Rollback Strategy + +Before Phase 7, each backend port can be reverted independently because the +broad stream remains unenforced. After enforcement begins, rollback must +revert the matching decorator/receive limit and its schema producer together; +never leave one actor side strict while restoring a dictionary sender. + +No long-lived persisted wire data requires a schema migration. Accounting and +ledger files remain outside this protocol change. If a strictness regression +is discovered, temporarily relaxing `forbid_unknown_fields` is safer than +restoring raw dictionary dispatch, provided the tagged union and core typed +fields remain enforced. + +## Exit Criteria + +The formal EMS messaging upgrade is complete when: + +- all client/EMS and EMS/backend IPC payloads are tagged structs; +- paper, Binance, Kraken, and IB share one broker startup/request/event + contract; +- both Tractor context families declare broad `pld_spec` aliases and narrow + each receive direction; +- startup, stream, and return payload failures produce tested + `MsgTypeError` behavior; +- no broker status outside `pending/open/canceled/closed` crosses IPC; +- fills and existing orders use dedicated variants; +- no business logic reconstructs wire dictionaries into message structs; +- ack/status/cancel races are deterministic and bounded; +- multi-client order routing is tested and the unresolved position-broadcast + privacy limitation is explicit; +- default EMS tests run without network or credentials; +- actor teardown failures are asserted rather than tolerated; +- backend authors have one conformance suite and protocol definition to follow. + +## Primary References + +- `piker/clearing/_messages.py`: current EMS and broker message structs. +- `piker/clearing/_ems.py`: broker dialog opening, translation, client command + processing, and `_emsd_main()`. +- `piker/clearing/_client.py`: public EMS client and startup handling. +- `piker/clearing/_paper_engine.py`: paper reference backend. +- `piker/brokers/binance/broker.py`: Binance request and event adapter. +- `piker/brokers/kraken/broker.py`: Kraken request and event adapter. +- `piker/brokers/ib/broker.py`: IB request and event adapter. +- `piker/data/validate.py`: existing datad-side `FeedInit` backend validation + direction to improve upon rather than duplicate. +- `piker/ui/_remote_ctl.py`: in-tree `pld_spec` and per-direction + `limit_plds()` precedent. +- `piker/ui/_gaps.py`: in-tree tagged union precedent. +- `tests/test_ems.py`: current network-dependent clearing coverage and empty + order tests. +- `/home/goodboy/repos/tractor/tests/msg/test_pldrx_limiting.py`: canonical + payload limiting and `MsgTypeError` behavior. +- `/home/goodboy/repos/tractor/tests/msg/test_ext_types_msgspec.py`: canonical + msgspec codec behavior. +- `/home/goodboy/repos/tractor/tests/test_context_stream_semantics.py`: + context stream cancellation and serialization semantics.