Plan typed `EMS` and `brokerd` messaging
Define the schema-first migration from dictionary actor traffic to tagged `msgspec.Struct` payloads on both order-control boundaries. Deats, - normalize startup, status, identity and fill semantics; - port paper, Binance, Kraken and IB through one contract; - gate strict `tractor` payload specs on backend conformance; - layer deterministic state tests with real actor IPC checks; - split implementation into reviewable dependency-ordered commits. (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))wkt/typed_ems_messaging
parent
9115d24da0
commit
baf95bfb23
|
|
@ -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.<backend> / paperboi.<backend>
|
||||
```
|
||||
|
||||
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='<stable value>'` 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.
|
||||
Loading…
Reference in New Issue