diff --git a/piker/clearing/_codec.py b/piker/clearing/_codec.py
new file mode 100644
index 00000000..5f024bd3
--- /dev/null
+++ b/piker/clearing/_codec.py
@@ -0,0 +1,143 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public
+# License as published by the Free Software Foundation, either
+# version 3 of the License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program. If not, see
+# <https://www.gnu.org/licenses/>.
+
+'''
+Validation shared by local clearing queues and actor streams.
+
+``msgspec.Struct`` construction does not validate field values. New
+protocol sends validate outbound values first; Tractor independently
+validates inbound payloads using its scoped decoder.
+
+'''
+
+from decimal import Decimal
+from functools import lru_cache
+from math import isfinite
+from types import UnionType
+from typing import (
+    get_args,
+    Self,
+    TypeVar,
+)
+
+import msgspec
+from msgspec import msgpack
+
+from piker.types import Struct
+
+
+T = TypeVar('T')
+
+
+class ProtocolError(ValueError):
+    '''
+    A clearing peer violated its message direction or phase.
+
+    '''
+
+
+class Message(
+    Struct,
+    frozen=True,
+    kw_only=True,
+    tag=True,
+    tag_field='type',
+    forbid_unknown_fields=True,
+):
+    '''
+    Immutable clearing snapshot with an explicit wire discriminator.
+
+    ``OrderState.dialogs`` remains mutable in its owning task.
+    Messages contain snapshots rather than references to that
+    registry.
+
+    '''
+
+    def copy(self, update: dict[str, object]|None = None) -> Self:
+        '''
+        Return a validated replacement without mutating the source.
+
+        '''
+        return validate(
+            msgspec.structs.replace(self, **(update or {})),
+            type(self),
+        )
+
+
+@lru_cache(maxsize=None)
+def decoder(spec: type|UnionType) -> msgpack.Decoder:
+    '''
+    Cache the strict msgpack decoder for one payload spec.
+
+    '''
+    return msgpack.Decoder(spec, strict=True)
+
+
+def validate(value: T, spec: type[T]|UnionType) -> T:
+    '''
+    Check the local object class, then validate and detach its
+    fields.
+
+    A tagged map is wire-equivalent to a Struct. Rejecting
+    dictionaries here enforces the Python API contract before
+    serialization, while the decoder enforces the actual wire schema.
+
+    '''
+    allowed: tuple[type, ...] = get_args(spec) or (spec,)
+    if type(value) not in allowed:
+        name: str = type(value).__name__
+        raise ProtocolError(f'Unexpected {name}; expected {spec}')
+    decoded: T = decoder(spec).decode(msgpack.encode(value))
+    _check_objects(value, decoded)
+    return decoded
+
+
+def _check_objects(value: object, decoded: object) -> None:
+    '''
+    Reject nested maps masquerading as messages in local Python
+    calls.
+
+    Decimal and tuple wire encodings cannot prove their original
+    Python types. Compare the validated tree with the supplied tree
+    before admitting it to local state. Integer-to-float widening is
+    the sole accepted scalar coercion.
+
+    '''
+    if (
+        type(value) is not type(decoded)
+        and
+        not (
+            type(value) is int and type(decoded) is float
+        )
+    ):
+        raise ProtocolError('Local object differs from its schema')
+    if isinstance(decoded, Message):
+        name: str
+        for name in decoded.__struct_fields__:
+            _check_objects(
+                getattr(value, name),
+                getattr(decoded, name),
+            )
+    elif isinstance(decoded, tuple):
+        original: object
+        normalized: object
+        for original, normalized in zip(value, decoded, strict=True):
+            _check_objects(original, normalized)
+    elif isinstance(decoded, float) and not isfinite(decoded):
+        raise ProtocolError('Non-finite clearing number')
+    elif isinstance(decoded, Decimal) and not decoded.is_finite():
+        raise ProtocolError('Non-finite clearing price')
diff --git a/piker/clearing/_protocol.py b/piker/clearing/_protocol.py
new file mode 100644
index 00000000..4607bdec
--- /dev/null
+++ b/piker/clearing/_protocol.py
@@ -0,0 +1,210 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public
+# License as published by the Free Software Foundation, either
+# version 3 of the License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program. If not, see
+# <https://www.gnu.org/licenses/>.
+
+'''
+Directional clearing streams over Tractor's typed payload decoder.
+
+The context decorator declares the whole lifecycle. Each stream
+narrows reception to its peer's messages; startup records and return
+values cannot appear as streaming events. Phase barriers are checked
+by the EMS separately from this direction check.
+
+'''
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager as acm
+from types import UnionType
+from typing import Generic, Self, TypeVar
+
+import tractor
+import trio
+
+from piker.types import Struct
+
+from ._codec import ProtocolError, validate
+from ._schema import (
+    BrokerdEvent,
+    BrokerdOpenOrder,
+    BrokerdOpenOrdersComplete,
+    BrokerdPosition,
+    BrokerdRequest,
+    BrokerdStartup,
+    validate_startup,
+)
+
+
+Rx = TypeVar('Rx')
+Tx = TypeVar('Tx')
+
+
+class Stream(Generic[Rx, Tx]):
+    '''
+    Validate outbound objects before the actor transport sees them.
+
+    Tractor checks the wire payload on reception. The additional
+    local check also protects memory-channel test doubles and
+    disallows maps at the Python API boundary.
+
+    '''
+
+    def __init__(
+        self,
+        stream: tractor.MsgStream,
+        incoming: type[Rx]|UnionType,
+        outgoing: type[Tx]|UnionType,
+    ) -> None:
+        '''
+        Bind the raw stream and its directional payload specs.
+
+        '''
+        self.raw: tractor.MsgStream = stream
+        self.incoming: type[Rx]|UnionType = incoming
+        self.outgoing: type[Tx]|UnionType = outgoing
+
+    @property
+    def _closed(self) -> bool:
+        '''
+        Expose the underlying stream closure state.
+
+        '''
+        return self.raw._closed
+
+    @property
+    def chan(self) -> tractor.Channel:
+        '''
+        Expose the Tractor channel used by this stream.
+
+        '''
+        return self.raw.chan
+
+    async def send(self, msg: Tx) -> None:
+        '''
+        Validate and detach an outgoing object before IPC.
+
+        '''
+        await self.raw.send(validate(msg, self.outgoing))
+
+    async def receive(self) -> Rx:
+        '''
+        Receive and validate an object against the incoming spec.
+
+        '''
+        return validate(await self.raw.receive(), self.incoming)
+
+    def __aiter__(self) -> Self:
+        '''
+        Iterate the same directional receiving stream.
+
+        '''
+        return self
+
+    async def __anext__(self) -> Rx:
+        '''
+        Translate normal stream completion into iterator completion.
+
+        '''
+        try:
+            return await self.receive()
+        except trio.EndOfChannel:
+            raise StopAsyncIteration from None
+
+
+@acm
+async def open_stream(
+    ctx: tractor.Context,
+    incoming: type[Rx]|UnionType,
+    outgoing: type[Tx]|UnionType,
+) -> AsyncIterator[Stream[Rx, Tx]]:
+    '''
+    Give one receiving task ownership of the directional decoder.
+
+    '''
+    with ctx.pld_rx.limit_plds(spec=incoming):
+        raw: tractor.MsgStream
+        async with ctx.open_stream() as raw:
+            yield Stream(raw, incoming, outgoing)
+
+
+class BrokerdPhase(Struct):
+    '''
+    Require one startup barrier before accepting outgoing requests.
+
+    ``BrokerdPhase.complete`` belongs to the receiving task. Existing
+    orders are legal only before that barrier, including an empty
+    snapshot. Account and broker ownership are checked independently
+    of decoding.
+
+    '''
+
+    startup: BrokerdStartup
+    complete: bool = False
+
+    def __post_init__(self) -> None:
+        '''
+        Validate the startup snapshot before accepting events.
+
+        '''
+        self.startup = validate_startup(self.startup)
+
+    def receive(self, msg: BrokerdEvent) -> BrokerdEvent:
+        '''
+        Check event ownership and advance the startup barrier once.
+
+        '''
+        msg = validate(msg, BrokerdEvent)
+        if self.startup.mode == 'paper_fallback':
+            raise ProtocolError('Fallback has no live event stream')
+
+        broker: str = self.startup.broker
+        if getattr(msg, 'broker', broker) != broker:
+            raise ProtocolError('Event changed broker ownership')
+
+        if isinstance(msg, BrokerdOpenOrdersComplete):
+            if self.complete:
+                raise ProtocolError('Duplicate startup barrier')
+            self.complete = True
+            return msg
+        if isinstance(msg, BrokerdOpenOrder):
+            if self.complete:
+                raise ProtocolError('Existing order after startup')
+            account: str = msg.order.account
+            symbol: str = msg.order.symbol
+            if not symbol.endswith(f'.{broker}'):
+                raise ProtocolError('Wrong existing-order broker')
+
+        else:
+            account = msg.account
+        if account not in self.startup.accounts:
+            raise ProtocolError('Event has an unannounced account')
+        if isinstance(msg, BrokerdPosition) and not (
+            msg.symbol.endswith(f'.{broker}')
+        ):
+            raise ProtocolError('Position has wrong broker')
+        return msg
+
+    def request(self, msg: BrokerdRequest) -> BrokerdRequest:
+        '''
+        Require startup completion and an announced request account.
+
+        '''
+        msg = validate(msg, BrokerdRequest)
+        if not self.complete:
+            raise ProtocolError('Request before startup barrier')
+
+        if msg.account not in self.startup.accounts:
+            raise ProtocolError('Request has an unannounced account')
+        return msg
diff --git a/piker/clearing/_schema.py b/piker/clearing/_schema.py
new file mode 100644
index 00000000..063ea88f
--- /dev/null
+++ b/piker/clearing/_schema.py
@@ -0,0 +1,358 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public
+# License as published by the Free Software Foundation, either
+# version 3 of the License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program. If not, see
+# <https://www.gnu.org/licenses/>.
+
+'''
+Strict clearing protocol for the staged EMS migration.
+
+The legacy endpoints still use ``._messages`` until their producers
+and consumers move together. New endpoints import these schemas;
+there is deliberately no dictionary compatibility decoder.
+
+``oid`` belongs to the client/EMS dialog, ``reqid`` to a provider's
+account, and ``fill_id`` to an execution. Position summaries contain
+no ledger history. Provider-native records stay inside adapters.
+
+'''
+
+from decimal import Decimal
+from typing import Annotated, Literal
+
+from msgspec import Meta
+
+from ._codec import Message, ProtocolError, validate
+
+
+Identity = Annotated[str, Meta(min_length=1)]
+RequestId = int|Identity
+TradeSide = Literal['buy', 'sell']
+OrderResponse = Literal[
+    'pending',
+    'open',
+    'dark_open',
+    'triggered',
+    'closed',
+    'fill',
+    'canceled',
+    'error',
+]
+
+
+class Order(Message, tag='order', kw_only=True):
+    '''
+    Client intent for a live order, dark trigger, or alert.
+
+    '''
+
+    oid: Identity
+    symbol: Identity
+    account: Identity
+    action: Literal['buy', 'sell', 'alert']
+    exec_mode: Literal['dark', 'live']
+    price: Decimal
+    size: float  # signed: negative sells, positive buys
+    brokers: tuple[Identity, ...] = ()
+
+
+class Cancel(Message, tag='cancel', kw_only=True):
+    '''
+    Client intent to cancel the dialog identified by ``oid``.
+
+    '''
+
+    oid: Identity
+    symbol: Identity
+
+
+class BrokerdOrder(Message, tag='broker_order', kw_only=True):
+    '''
+    Submit or edit a provider order for an EMS dialog.
+
+    ``BrokerdOrder.reqid`` is absent on a new submission and names
+    the current provider order on an edit. ``BrokerdOrder.symbol``
+    contains the market endpoint without its broker suffix.
+
+    '''
+
+    oid: Identity
+    account: Identity
+    time_ns: int
+    symbol: Identity  # broker suffix removed at the EMS boundary
+    price: Decimal
+    size: float
+    action: TradeSide
+    reqid: RequestId|None = None
+
+
+class BrokerdCancel(Message, tag='broker_cancel', kw_only=True):
+    '''
+    Cancel an acknowledged provider order.
+
+    ``BrokerdCancel.reqid`` is required; EMS retains cancellation
+    intent until an acknowledgement establishes that provider ID.
+
+    '''
+
+    oid: Identity
+    account: Identity
+    time_ns: int
+    symbol: Identity
+    reqid: RequestId
+
+
+class BrokerdOrderAck(Message, tag='broker_ack', kw_only=True):
+    '''
+    Associate an EMS dialog with its account/provider ID.
+
+    '''
+
+    oid: Identity
+    account: Identity
+    reqid: RequestId
+
+
+class BrokerdStatus(Message, tag='broker_status', kw_only=True):
+    '''
+    Report provider control state independently of executions.
+
+    '''
+
+    time_ns: int
+    reqid: RequestId
+    account: Identity
+    broker: Identity
+    status: Literal['pending', 'open', 'canceled', 'closed']
+    filled: float = 0.0
+    remaining: float = 0.0
+    reason: str = ''
+
+
+class BrokerdFill(Message, tag='broker_fill', kw_only=True):
+    '''
+    One execution, independently identified from its order.
+
+    ``BrokerdFill.broker_time`` is Unix seconds for chart placement.
+    Adapters normalize native clocks and execution IDs before
+    sending. Repeated delivery retains ``BrokerdFill.fill_id``.
+
+    '''
+
+    time_ns: int
+    broker_time: float
+    reqid: RequestId
+    fill_id: Identity
+    account: Identity
+    broker: Identity
+    action: TradeSide
+    size: float
+    price: float
+
+
+class BrokerdError(Message, tag='broker_error', kw_only=True):
+    '''
+    Report a provider rejection with optional typed request context.
+
+    '''
+
+    account: Identity
+    broker: Identity
+    reason: Identity
+    oid: Identity|None = None
+    reqid: RequestId|None = None
+    symbol: Identity|None = None
+    request: BrokerdOrder|BrokerdCancel|None = None
+
+
+class BrokerdPosition(Message, tag='position', kw_only=True):
+    '''
+    Account projection; never a durable accounting transaction.
+
+    '''
+
+    broker: Identity
+    account: Identity
+    symbol: Identity  # full FQME
+    size: float
+    avg_price: float
+    currency: str = ''
+    bs_mktid: int|str|None = None
+
+
+class BrokerdOpenOrder(Message, tag='open_order', kw_only=True):
+    '''
+    An existing provider order emitted before the startup barrier.
+
+    ``BrokerdOpenOrder.order.oid`` names the provider and account.
+    Its symbol is a full FQME. ``BrokerdOpenOrder.reqid`` retains the
+    provider's control identity.
+
+    '''
+
+    reqid: RequestId
+    order: Order
+    broker: Identity
+    time_ns: int
+
+
+class BrokerdOpenOrdersComplete(
+    Message,
+    tag='open_orders_complete',
+    kw_only=True,
+):
+    '''
+    All existing orders have been emitted, including an empty set.
+
+    '''
+
+    broker: Identity
+
+
+BrokerdRequest = BrokerdOrder|BrokerdCancel
+BrokerdEvent = (
+    BrokerdOrderAck
+    |BrokerdStatus
+    |BrokerdFill
+    |BrokerdError
+    |BrokerdPosition
+    |BrokerdOpenOrder
+    |BrokerdOpenOrdersComplete
+)
+
+
+class Status(Message, tag='status', kw_only=True):
+    '''
+    Immutable EMS snapshot with a typed cause for each transition.
+
+    Mutable cancel and deduplication state belongs to ``Dialog``.
+    Request and event causes occupy separate fields, so neither needs
+    a catch-all dictionary or a recursive copy of provider history.
+
+    '''
+
+    time_ns: int
+    oid: Identity
+    resp: OrderResponse
+    req: Order
+    reqid: RequestId|None = None
+    src: str|None = None
+    cancel_called: bool = False
+    brokerd_request: BrokerdRequest|None = None
+    brokerd_event: BrokerdEvent|None = None
+
+
+class BrokerdStartup(Message, tag='broker_startup', kw_only=True):
+    '''
+    Announce owned accounts and positions, or delegate to paper.
+
+    '''
+
+    broker: Identity
+    mode: Literal['live', 'paper', 'paper_fallback']
+    accounts: tuple[Identity, ...] = ()
+    positions: tuple[BrokerdPosition, ...] = ()
+
+
+class EmsStartup(Message, tag='ems_startup', kw_only=True):
+    '''
+    Publish owned positions and active dialogs when EMS is ready.
+
+    '''
+
+    broker: Identity
+    positions: tuple[BrokerdPosition, ...]
+    accounts: tuple[Identity, ...]
+    dialogs: tuple[Status, ...]
+
+
+EmsRequest = Order|Cancel
+EmsEvent = Status|BrokerdPosition
+EmsPayload = EmsStartup|EmsRequest|EmsEvent|None
+BrokerdPayload = (
+    BrokerdStartup
+    |BrokerdRequest
+    |BrokerdEvent
+    |None
+)
+
+
+def validate_startup(msg: BrokerdStartup) -> BrokerdStartup:
+    '''
+    Validate account ownership before publishing a backend snapshot.
+
+    '''
+    msg = validate(msg, BrokerdStartup)
+    _validate_positions(msg.broker, msg.accounts, msg.positions)
+    if msg.mode == 'paper_fallback':
+        if msg.accounts or msg.positions:
+            raise ProtocolError('Fallback cannot publish live state')
+    elif not msg.accounts:
+        raise ProtocolError('Active startup requires an account')
+    elif msg.mode == 'paper' and msg.accounts != ('paper',):
+        raise ProtocolError(
+            'Paper startup requires the paper account'
+        )
+    return msg
+
+
+def validate_ems_startup(msg: EmsStartup) -> EmsStartup:
+    '''
+    Validate account, position, and active-dialog ownership.
+
+    '''
+    msg = validate(msg, EmsStartup)
+    if not msg.accounts:
+        raise ProtocolError('EMS startup requires an account')
+    broker: str = msg.broker
+    _validate_positions(broker, msg.accounts, msg.positions)
+    oids: set[str] = set()
+    status: Status
+    for status in msg.dialogs:
+        order: Order = status.req
+        if (
+            status.oid != order.oid
+            or status.oid in oids
+            or order.account not in msg.accounts
+            or not order.symbol.endswith(f'.{broker}')
+            or status.resp in ('closed', 'canceled')
+        ):
+            raise ProtocolError('Invalid startup dialog ownership')
+        oids.add(status.oid)
+    return msg
+
+
+def _validate_positions(
+    broker: str,
+    accounts: tuple[str, ...],
+    positions: tuple[BrokerdPosition, ...],
+) -> None:
+    '''
+    Reject duplicate accounts or positions outside their owner.
+
+    '''
+    if len(set(accounts)) != len(accounts):
+        raise ProtocolError('Duplicate startup accounts')
+    keys: set[tuple[str, str]] = set()
+    pos: BrokerdPosition
+    for pos in positions:
+        key: tuple[str, str] = (pos.account, pos.symbol)
+        if (
+            pos.broker != broker
+            or pos.account not in accounts
+            or not pos.symbol.endswith(f'.{broker}')
+            or key in keys
+        ):
+            raise ProtocolError('Invalid startup position ownership')
+        keys.add(key)
diff --git a/piker/clearing/_state.py b/piker/clearing/_state.py
new file mode 100644
index 00000000..a935fa9a
--- /dev/null
+++ b/piker/clearing/_state.py
@@ -0,0 +1,470 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public
+# License as published by the Free Software Foundation, either
+# version 3 of the License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program. If not, see
+# <https://www.gnu.org/licenses/>.
+
+'''
+EMS order-dialog transitions, independent of actor I/O.
+
+``Account`` owns accounting. ``OrderState`` only correlates order
+control IDs and produces immutable client snapshots and broker
+requests. Events can precede an ack or follow a terminal status;
+finite buffers retain those events without reopening finished orders.
+
+'''
+
+from collections import OrderedDict
+from time import time_ns
+from typing import Literal
+
+from msgspec import field
+from msgspec.structs import replace
+
+from piker.types import Struct
+from ._codec import ProtocolError, validate
+from ._schema import (
+    BrokerdCancel,
+    BrokerdError,
+    BrokerdEvent,
+    BrokerdFill,
+    BrokerdOpenOrder,
+    BrokerdOrder,
+    BrokerdOrderAck,
+    BrokerdRequest,
+    BrokerdStatus,
+    Order,
+    Status,
+)
+
+
+class Effects(Struct, frozen=True):
+    '''
+    Return immutable broker requests and client status snapshots.
+
+    '''
+
+    broker: tuple[BrokerdRequest, ...] = ()
+    clients: tuple[Status, ...] = ()
+
+
+class Dialog(Struct):
+    '''
+    Retain mutable cancellation and fill state for one EMS dialog.
+
+    '''
+
+    status: Status
+    cancel_sent: set[int|str] = field(default_factory=set)
+    fills: set[str] = field(default_factory=set)
+    terminal: bool = False
+
+
+class OrderState(Struct):
+    '''
+    Own active dialogs, bounded early events and terminal tombstones.
+
+    ``OrderState.reqids`` includes the account in each key: separate
+    broker sessions can allocate the same numeric order ID. Old
+    request IDs remain correlated until their tombstone is evicted
+    from ``OrderState.terminal``, so a late execution following an
+    edit still belongs to its original order. Capacity exhaustion
+    raises without discarding early fills.
+
+    '''
+
+    dialogs: dict[str, Dialog] = field(default_factory=dict)
+    active: dict[str, Status] = field(default_factory=dict)
+    reqids: dict[tuple[str, int|str], str] = field(
+        default_factory=dict,
+    )
+    early: dict[tuple[str, int|str], list[BrokerdEvent]] = field(
+        default_factory=dict,
+    )
+    terminal: OrderedDict[str, Dialog] = field(
+        default_factory=OrderedDict,
+    )
+    max_early: int = 256
+    max_terminal: int = 1024
+
+    def register(
+        self,
+        order: Order,
+        resp: Literal['pending', 'open', 'dark_open'] = 'pending',
+    ) -> Status:
+        '''
+        Register intent or replace an acknowledged active order.
+
+        Keep ``OrderState.dialogs`` and ``OrderState.active``
+        coherent while rejecting changes to the original order
+        ownership.
+
+        '''
+        order = validate(order, Order)
+        oid: str = order.oid
+        prior: Dialog|None = self.dialogs.get(oid)
+        if oid in self.terminal:
+            raise ProtocolError('Cannot reuse a completed order ID')
+
+        reqid: int|str|None = None
+        if prior:
+            previous: Order = prior.status.req
+            if (
+                previous.symbol != order.symbol
+                or previous.account != order.account
+                or previous.exec_mode != order.exec_mode
+            ):
+                raise ProtocolError('Edit changed order ownership')
+            reqid = prior.status.reqid
+            if prior.status.cancel_called:
+                raise ProtocolError('Cannot edit a canceling order')
+            if reqid is None and prior.status.resp != 'dark_open':
+                raise ProtocolError(
+                    'Cannot edit before acknowledgement',
+                )
+        status: Status = validate(
+            Status(
+                time_ns=time_ns(),
+                oid=oid,
+                req=order,
+                reqid=reqid,
+                resp=resp,
+            ),
+            Status,
+        )
+        if prior:
+            prior.status = status
+        else:
+            self.dialogs[oid] = Dialog(status=status)
+        self.active[oid] = status
+        return status
+
+    def _publish(self, status: Status) -> None:
+        '''
+        Refresh ``Dialog.status`` and ``OrderState.active``.
+
+        '''
+        dialog: Dialog = self.dialogs[status.oid]
+        dialog.status = status
+        if not dialog.terminal:
+            self.active[status.oid] = status
+
+    def _finish(self, status: Status) -> None:
+        '''
+        Move a completed dialog into ``OrderState.terminal``.
+
+        Evict its entries from ``OrderState.reqids`` when terminal
+        retention exceeds ``OrderState.max_terminal``.
+
+        '''
+        oid: str = status.oid
+        dialog: Dialog = self.dialogs.pop(oid)
+        dialog.status = status
+        dialog.terminal = True
+        self.active.pop(oid, None)
+        self.terminal[oid] = dialog
+        while len(self.terminal) > self.max_terminal:
+            old_oid: str
+            _: Dialog
+            old_oid, _ = self.terminal.popitem(last=False)
+            key: tuple[str, int|str]
+            for key in tuple(self.reqids):
+                if self.reqids[key] == old_oid:
+                    del self.reqids[key]
+
+    def _cancel(self, dialog: Dialog) -> tuple[BrokerdCancel, ...]:
+        '''
+        Emit a cancel once per ID recorded in ``Dialog.cancel_sent``.
+
+        '''
+        status: Status = dialog.status
+        reqid: int|str|None = status.reqid
+        if (
+            reqid is None
+            or reqid in dialog.cancel_sent
+            or dialog.terminal
+        ):
+            return ()
+        dialog.cancel_sent.add(reqid)
+        order: Order = status.req
+        return (
+            BrokerdCancel(
+                oid=status.oid,
+                reqid=reqid,
+                account=order.account,
+                symbol=order.symbol,
+                time_ns=time_ns(),
+            ),
+        )
+
+    def cancel(self, oid: str) -> Effects:
+        '''
+        Retain cancel intent or finish an unsubmitted dark order.
+
+        '''
+        dialog: Dialog|None = self.dialogs.get(oid)
+        if dialog is None:
+            return Effects()
+        status: Status = replace(
+            dialog.status,
+            cancel_called=True,
+            time_ns=time_ns(),
+        )
+        if status.resp == 'dark_open':
+            status = replace(status, resp='canceled')
+            self._finish(status)
+            return Effects(clients=(status,))
+        self._publish(status)
+        return Effects(broker=self._cancel(dialog))
+
+    def _correlate(
+        self,
+        account: str,
+        reqid: int|str,
+        oid: str,
+    ) -> None:
+        '''
+        Bind account/provider identity in ``OrderState.reqids`` once.
+
+        '''
+        key: tuple[str, int|str] = (account, reqid)
+        previous: str|None = self.reqids.get(key)
+        if previous is not None and previous != oid:
+            raise ProtocolError('Request ID changed ownership')
+        self.reqids[key] = oid
+
+    def accept(self, event: BrokerdEvent) -> Effects:
+        '''
+        Apply one broker event and return immutable transition
+        effects.
+
+        ``OrderState.early`` retains uncorrelated events until an ack
+        or startup order establishes their ``OrderState.reqids``
+        entry. ``Dialog.fills`` deduplicates executions without
+        reviving a completed dialog in ``OrderState.active``.
+
+        '''
+        event = validate(event, BrokerdEvent)
+        if isinstance(event, BrokerdOpenOrder):
+            order: Order = event.order
+            if (
+                order.oid in self.dialogs
+                or order.oid in self.terminal
+            ):
+                raise ProtocolError('Duplicate startup order')
+            key: tuple[str, int|str] = (order.account, event.reqid)
+            if key in self.reqids:
+                raise ProtocolError('Request ID changed ownership')
+            broker: str = event.broker
+            if not order.symbol.endswith(f'.{broker}'):
+                raise ProtocolError('Wrong existing-order broker')
+            status: Status = self.register(order, resp='open')
+            status = replace(
+                status,
+                reqid=event.reqid,
+                src=event.broker,
+                time_ns=event.time_ns,
+                brokerd_event=event,
+            )
+            self._publish(status)
+            self._correlate(order.account, event.reqid, order.oid)
+            return self._replay(
+                order.account,
+                event.reqid,
+                (status,),
+            )
+
+        if isinstance(event, BrokerdOrderAck):
+            oid: str|None = event.oid
+            dialog: Dialog|None = (
+                self.dialogs.get(oid) or self.terminal.get(oid)
+            )
+            if dialog is None:
+                raise ProtocolError('Ack for an unknown order')
+            if event.account != dialog.status.req.account:
+                raise ProtocolError('Ack changed account ownership')
+            key = (event.account, event.reqid)
+            known: bool = key in self.reqids
+            self._correlate(event.account, event.reqid, oid)
+            if known and dialog.status.reqid != event.reqid:
+                # A delayed duplicate ack for a replaced request ID
+                # cannot revert the current cancel target.
+                return Effects()
+            if not dialog.terminal:
+                self._publish(
+                    replace(
+                        dialog.status,
+                        reqid=event.reqid,
+                        time_ns=time_ns(),
+                    )
+                )
+            replayed: Effects = self._replay(
+                event.account,
+                event.reqid,
+            )
+            # Replay a terminal event first: an order already closed
+            # before its ack no longer needs a deferred cancel.
+            cancels: tuple[BrokerdCancel, ...] = (
+                self._cancel(dialog)
+                if dialog.status.cancel_called
+                else ()
+            )
+            return Effects(
+                broker=replayed.broker + cancels,
+                clients=replayed.clients,
+            )
+
+        if not isinstance(
+            event,
+            (BrokerdStatus, BrokerdFill, BrokerdError),
+        ):
+            raise ProtocolError('Not an order-dialog event')
+
+        account: str = event.account
+        broker: str = event.broker
+        reqid: int|str|None = event.reqid
+        oid = self.reqids.get((account, reqid))
+        if isinstance(event, BrokerdError):
+            error_oid: str|None = event.oid
+            request: BrokerdRequest|None = event.request
+            if request:
+                if (
+                    request.account != account
+                    or error_oid not in (None, request.oid)
+                ):
+                    raise ProtocolError(
+                        'Error changed request ownership',
+                    )
+                error_oid = request.oid
+            if oid and error_oid and oid != error_oid:
+                raise ProtocolError('Error changed order ownership')
+            oid = error_oid or oid
+        if oid is None:
+            if reqid is None:
+                raise ProtocolError('Event has no correlation ID')
+            count: int = sum(map(len, self.early.values()))
+            if count >= self.max_early:
+                raise ProtocolError('Pre-ack capacity exceeded')
+            self.early.setdefault((account, reqid), []).append(event)
+            return Effects()
+        dialog = self.dialogs.get(oid) or self.terminal.get(oid)
+        if dialog is None:
+            raise ProtocolError('Missing correlated dialog')
+
+        if account != dialog.status.req.account:
+            raise ProtocolError('Event changed account ownership')
+
+        if not dialog.status.req.symbol.endswith(f'.{broker}'):
+            raise ProtocolError('Event changed broker ownership')
+        status = replace(
+            dialog.status,
+            time_ns=time_ns(),
+            brokerd_event=event,
+            src=event.broker,
+        )
+        if isinstance(event, BrokerdFill):
+            if not event.fill_id:
+                raise ProtocolError('Fill has no execution identity')
+            if event.fill_id in dialog.fills:
+                return Effects()
+            dialog.fills.add(event.fill_id)
+            # The event retains its execution's reqid. The snapshot
+            # keeps the current control ID after a broker-side edit.
+            status = replace(status, resp='fill')
+            # A late fill is observable but never reopens its dialog.
+            if not dialog.terminal:
+                self._publish(status)
+            return Effects(clients=(status,))
+        if dialog.terminal:
+            return Effects()
+        if (
+            isinstance(event, BrokerdStatus)
+            and reqid != dialog.status.reqid
+        ):
+            # Replacing a provider order may cancel the old ID. Its
+            # delayed status cannot close the replacement dialog.
+            return Effects()
+        if isinstance(event, BrokerdError):
+            status = replace(status, resp='error')
+            if (
+                isinstance(event.request, BrokerdOrder)
+                and event.request.reqid is None
+                and dialog.status.reqid is None
+            ):
+                # A rejected new submission cannot receive a useful
+                # ack. Retain a tombstone instead of a pending order.
+                self._finish(status)
+            else:
+                self._publish(status)
+        else:
+            status = replace(
+                status, resp=event.status, reqid=reqid
+            )
+            if event.status in ('closed', 'canceled'):
+                self._finish(status)
+            else:
+                self._publish(status)
+        return Effects(clients=(status,))
+
+    def _replay(
+        self,
+        account: str,
+        reqid: int|str,
+        clients: tuple[Status, ...] = (),
+    ) -> Effects:
+        '''
+        Drain correlated events from ``OrderState.early``.
+
+        '''
+        requests: tuple[BrokerdRequest, ...] = ()
+        event: BrokerdEvent
+        for event in self.early.pop((account, reqid), ()):
+            effects: Effects = self.accept(event)
+            requests += effects.broker
+            clients += effects.clients
+        return Effects(broker=requests, clients=clients)
+
+
+def broker_request(
+    order: Order,
+    reqid: int|str|None = None,
+) -> BrokerdOrder:
+    '''
+    Remove exactly the final broker suffix for the provider API.
+
+    '''
+    order = validate(order, Order)
+    symbol: str
+    _: str
+    broker: str
+    (
+        symbol,
+        _,
+        broker,
+    ) = order.symbol.rpartition('.')
+    if not symbol or not broker:
+        raise ProtocolError('Order requires a full FQME')
+    return validate(
+        BrokerdOrder(
+            oid=order.oid,
+            reqid=reqid,
+            symbol=symbol,
+            time_ns=time_ns(),
+            account=order.account,
+            action=order.action,
+            price=order.price,
+            size=order.size,
+        ),
+        BrokerdOrder,
+    )
diff --git a/plans/opencode/typed-ems-messaging.md b/plans/opencode/typed-ems-messaging.md
new file mode 100644
index 00000000..dae3274d
--- /dev/null
+++ b/plans/opencode/typed-ems-messaging.md
@@ -0,0 +1,908 @@
+# Typed EMS and Brokerd Messaging
+
+## October 2026 implementation baseline
+
+This revision carries the August 28 design into the September Piker
+baseline (`eed58566`). Preserve IB execution-ID accounting (`5eb4fc9d`)
+and socket-disconnect propagation (`efa72980`) while changing payloads.
+The installed Tractor is `5c98ab1f`; qualification must distinguish
+that April revision from local `main` (`5cb600a8`) and the September
+context-fault development stack. Do not test against a dirty upstream
+checkout without recording its changes.
+
+The complete contract includes local client queues, EMS startup and
+service readiness, both order-control streams, quote normalization for
+dark/paper execution, and typed internal trigger/dialog records.
+Provider-native APIs and persistent ledger formats retain their own
+normalization boundaries. Runtime handles are not wire messages.
+
+Use explicit Struct validation before local state mutation or sending;
+annotations and Struct construction do not validate field values.
+Receive decoding proves wire shape, not the sender's Python type: a
+correctly tagged mapping has the same encoding as a Struct.
+
+Endpoint-wide unions need separate startup, receive-direction, and
+final-result checks. Validate RPC arguments explicitly until Tractor
+issue #514 supplies signature-derived `Start` contracts. Never admit
+`Any` or raw diagnostic mappings into the clearing wire schema.
+
+Frozen snapshots must use replacement rather than the inherited
+`Struct.copy(update=...)`, which mutates its source. Keep mutable EMS
+dialog state separate from published messages and retain execution IDs
+for fill deduplication. Position messages remain explicit summaries;
+they must never become synthetic entries in durable trade accounting.
+
+Port each boundary's producers and consumers together before enabling
+its strict receive set. Qualify schema errors, wrong-direction and
+wrong-phase messages, startup completion, cancel-before-ack, reordered
+and repeated fills, transport failure, and bounded teardown with
+offline provider fixtures and real local actors. Live order submission
+is outside this qualification.
+
+
+### Qualified dependency baseline and first review boundary
+
+The first boundary introduces `._schema`, local validation in `._codec`,
+directional streams and lifecycle checks in `._protocol`, and a pure
+`._state.OrderState` reducer. Existing endpoints keep their original
+`._messages` imports until their paired migration. This is infrastructure
+qualification, not a claim that paper or any live adapter is ported.
+
+The installed baseline is Python 3.13.14, msgspec 0.19.0, Trio 0.32.0,
+and Tractor `5c98ab1fb6e2cb7781aa75845c7f323a52853c46`. Its existing
+payload APIs support this boundary without a dependency update.
+All eight actor contract tests also pass against clean local Tractor
+`main` at `5cb600a84b6e9fcac68bad04a487b83640b3e5fb`.
+
+The first set has 65 schema/state/phase tests and eight real IPC tests.
+These tests cover protocol fixtures and the new reducer; adapter-level
+conformance, feed-driven paper clearing, abrupt transport EOF, and
+other spawn backends remain qualification work for subsequent ports.
+
+That local main adds `--ll` to Tractor's pytest plugin; Piker's current
+`tests/conftest.py` adds the same option. The ordinary downstream
+invocation therefore fails before collection. The isolated API check
+uses `--noconftest -p tractor._testing.pytest` on the same test file.
+Resolve that harness collision before upgrading Piker's dependency pin.
+Do not conflate this isolated API qualification with a whole-suite pass.
+
+Relevant unmerged Tractor work remains a separate qualification input:
+`pld_dec_refinements` (`6bd685ad`), `auto_codecs` (`45508252`),
+`wkt/context_fault_status` (`9547c2f9`), and
+`wkt/ipc_fault_regressions` (`fe9eb738`). The context-fault worktree has
+uncommitted transport-error handling changes; it is not a clean test
+baseline. The IPC-fault branch includes `192411c4` for transport EOF
+propagation. Typed Start arguments remain upstream work; validate them
+explicitly when migrating the actual clearing contexts.
+
+## 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,
+called `type`, with explicit protocol values such as `order`, `cancel`,
+`broker_ack`, `status`, `broker_fill`, `broker_error`, `position`, and
+`open_order`. Broker requests have distinct `broker_order` and
+`broker_cancel` tags so decoding cannot confuse the two directions. Do not rely on
+Python class names as wire tags.
+
+All newly formalized wire structs should use:
+
+- `tag='<stable value>'` and `tag_field='type'`;
+- `frozen=True` for wire snapshots;
+- `kw_only=True` where it improves constructor safety;
+- immutable tuple defaults; factories for mutable internal state;
+- finite `Literal` values for protocol enums;
+- `forbid_unknown_fields=True` from schema introduction; adopt the new
+  schema only when both endpoints can meet it.
+
+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 a required `Order`, never an unvalidated nested dictionary;
+  an uncorrelated backend error must not fabricate a client dialog;
+- `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 `broker_order` and `broker_cancel` protocol tags
+  independent of `action`;
+- `oid` always identifies the EMS dialog;
+- `BrokerdOrder.reqid` is `RequestId | None` because allocation policy
+  differs by backend; `BrokerdCancel.reqid` is required, and EMS retains
+  cancel intent until acknowledgement supplies that identity;
+- 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 acknowledgements. Status and fill events identify the
+order by `(account, reqid)` within their broker session; EMS retains that
+correlation across edits and terminal events. For an external order, the
+adapter creates a collision-safe stable `oid` from broker, account, and
+provider `reqid` before emitting `BrokerdOpenOrder`. Its embedded
+`Order.oid` carries that value. EMS registers it without allocating a
+reverse-message identity. Service errors may omit `oid`; a typed rejected
+request can identify a submission which never received a provider ID.
+
+`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:
+
+- `broker: str`;
+- `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`. Paper sessions announce exactly
+`('paper',)`. EMS readiness also requires at least one account.
+
+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
+
+Provider-native diagnostics remain inside adapters and their logs. The
+clearing wire schema has no raw-data extension point or `Any` field.
+Source backend, broker timestamp, symbol, account, order ID, request ID,
+and execution ID have explicit typed fields. Errors can carry a typed
+request. Position summaries contain no mutable accounting history.
+
+## 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 scheduling
+   can still expose ordering bugs. The first reducer bounds event count
+   and fails loudly on overflow without dropping retained executions.
+   Add wall-clock expiry in its asynchronous owner when porting EMS;
+   the pure reducer does not claim an age bound.
+5. A cancel received before ack marks the dialog `cancel_called`. After
+   replaying buffered events, the first ack sends one typed
+   `BrokerdCancel` if the dialog is still active.
+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
+
+Introduce the strict schema in `piker/clearing/_schema.py` alongside the
+legacy `._messages` imports. Qualify it independently before moving each
+producer/consumer pair together. Remove the legacy module after every
+endpoint and public consumer has migrated; do not add dictionary bridges.
+
+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`, and optional
+  identifier fields round-trip;
+- unknown tag, missing required field, wrong literal, wrong primitive type,
+  malformed nested struct, and duplicate/ambiguous tag fail;
+- unknown fields fail from the first use of the new schema;
+- 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 stays inside its adapter and logs.
+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. Verify every producer satisfies the already-strict wire schemas.
+
+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 typed request/event causes;
+- unknown/missing/wrong tags;
+- unknown fields rejected from schema introduction;
+- 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. add the refreshed dependency baseline, strict schemas, pure reducer,
+   lifecycle guards, and offline/actor protocol qualification;
+2. characterize and port paper together with its EMS consumer;
+3. port each live adapter and its paired EMS handling with provider
+   translation fixtures and execution-identity tests;
+4. migrate client/EMS and UI consumers together;
+5. enforce each migrated endpoint pair with directional limits and
+   lifecycle/race/failure qualification;
+6. remove legacy paths and document backend requirements.
+
+Every commit must leave deterministic tests green. The first boundary
+keeps legacy endpoints operational while independently qualifying the
+new protocol. Subsequent migrations move both ends together; no commit
+requires a raw-dictionary bridge into the strict schema.
+
+## Rollback Strategy
+
+Revert a backend port with its paired EMS changes. Once enforcement
+begins, include the matching decorator and receive limits in that
+rollback; 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, repair the producer or roll back the complete endpoint pair.
+Do not widen the schema to accept arbitrary dictionaries or unknown fields.
+
+## 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.
+
+
+Relevant issue/design threads:
+
+- https://github.com/goodboy/tractor/issues/196
+- https://github.com/goodboy/tractor/issues/365
+- https://github.com/goodboy/tractor/issues/514
+- https://github.com/goodboy/tractor/pull/376
+- https://github.com/pikers/piker/issues/514
+- https://github.com/pikers/piker/issues/297
+- https://github.com/pikers/piker/issues/466
diff --git a/tests/test_brokerd_protocol.py b/tests/test_brokerd_protocol.py
new file mode 100644
index 00000000..ac0bef53
--- /dev/null
+++ b/tests/test_brokerd_protocol.py
@@ -0,0 +1,214 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public
+# License as published by the Free Software Foundation, either
+# version 3 of the License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program. If not, see
+# <https://www.gnu.org/licenses/>.
+
+'''
+Offline lifecycle conformance for every future brokerd adapter.
+
+'''
+
+from decimal import Decimal
+
+import pytest
+
+from piker.clearing._codec import ProtocolError
+from piker.clearing._protocol import BrokerdPhase
+from piker.clearing._schema import (
+    BrokerdOpenOrder,
+    BrokerdOpenOrdersComplete,
+    BrokerdOrder,
+    BrokerdPosition,
+    BrokerdStartup,
+    EmsStartup,
+    Order,
+    Status,
+    validate_ems_startup,
+)
+
+
+@pytest.mark.parametrize('broker', ['ib', 'kraken', 'binance'])
+@pytest.mark.parametrize('mode', ['live', 'paper'])
+def test_startup_barrier(broker: str, mode: str) -> None:
+    '''
+    Every provider and paper mode shares one lifecycle contract.
+    Prohibit order submission before completion, allow an empty
+    snapshot, then reject a repeated completion or existing order.
+    These fixtures qualify the protocol, not the unmigrated adapters.
+
+    '''
+    account: str = 'paper' if mode == 'paper' else f'{broker}.test'
+    phase: BrokerdPhase = BrokerdPhase(
+        BrokerdStartup(
+            broker=broker,
+            mode=mode,
+            accounts=(account,),
+        )
+    )
+    request: BrokerdOrder = BrokerdOrder(
+        oid='one',
+        account=account,
+        symbol='btc.spot',
+        time_ns=1,
+        price=Decimal('10'),
+        size=1.0,
+        action='buy',
+    )
+    with pytest.raises(ProtocolError, match='before startup'):
+        phase.request(request)
+    complete: BrokerdOpenOrdersComplete = BrokerdOpenOrdersComplete(
+        broker=broker
+    )
+    assert phase.receive(complete) == complete
+    assert phase.request(request) == request
+    with pytest.raises(ProtocolError, match='Duplicate'):
+        phase.receive(complete)
+    with pytest.raises(ProtocolError, match='after startup'):
+        phase.receive(
+            BrokerdOpenOrder(
+                reqid=1,
+                time_ns=1,
+                broker=broker,
+                order=Order(
+                    oid='existing',
+                    account=account,
+                    symbol=f'btc.spot.{broker}',
+                    price=Decimal('10'),
+                    size=1.0,
+                    action='buy',
+                    exec_mode='live',
+                ),
+            )
+        )
+
+
+def test_paper_fallback_never_opens_live_stream() -> None:
+    '''
+    A backend without credentials hands control to a paper endpoint.
+    Its fallback startup cannot masquerade as an empty live snapshot:
+    even a syntactically valid completion event must be rejected.
+
+    '''
+    phase: BrokerdPhase = BrokerdPhase(
+        BrokerdStartup(
+            broker='ib',
+            mode='paper_fallback',
+        )
+    )
+    with pytest.raises(ProtocolError, match='Fallback'):
+        phase.receive(BrokerdOpenOrdersComplete(broker='ib'))
+    assert not phase.complete
+
+
+@pytest.mark.parametrize(
+    'changes',
+    [
+        {'broker': 'kraken'},
+        {'account': 'other'},
+        {'symbol': 'btc.spot.kraken'},
+    ],
+)
+def test_event_ownership(changes: dict[str, str]) -> None:
+    '''
+    A well-typed event may still belong to another session. Announce
+    one broker/account, then alter a position's owner or market
+    suffix. ``BrokerdPhase.receive()`` must reject it before EMS can
+    publish the position.
+
+    '''
+    phase: BrokerdPhase = BrokerdPhase(
+        BrokerdStartup(
+            broker='ib',
+            mode='live',
+            accounts=('ib.test',),
+        )
+    )
+    position: BrokerdPosition = BrokerdPosition(
+        broker='ib',
+        account='ib.test',
+        symbol='btc.spot.ib',
+        size=1.0,
+        avg_price=10.0,
+    )
+    with pytest.raises(ProtocolError):
+        phase.receive(position.copy(update=changes))
+
+
+@pytest.mark.parametrize(
+    'field,value',
+    [
+        ('oid', 'mismatch'),
+        ('resp', 'closed'),
+        ('resp', 'canceled'),
+    ],
+)
+def test_ems_snapshot_has_only_active_owned_dialogs(
+    field: str,
+    value: str,
+) -> None:
+    '''
+    Reconnect snapshots must not revive completed orders or associate
+    a request with another oid. Corrupt one field in a valid snapshot
+    and require semantic rejection after ordinary schema decoding.
+
+    '''
+    order: Order = Order(
+        oid='one',
+        account='paper',
+        symbol='btc.spot.ib',
+        price=Decimal('10'),
+        size=1.0,
+        action='buy',
+        exec_mode='live',
+    )
+    status: Status = Status(
+        time_ns=1, oid='one', resp='open', req=order
+    )
+    startup: EmsStartup = EmsStartup(
+        broker='ib',
+        accounts=('paper',),
+        positions=(),
+        dialogs=(status,),
+    )
+    assert validate_ems_startup(startup) == startup
+    bad: EmsStartup = startup.copy(
+        update={
+            'dialogs': (status.copy(update={field: value}),),
+        }
+    )
+    with pytest.raises(ProtocolError):
+        validate_ems_startup(bad)
+
+
+def test_ems_startup_requires_an_account() -> None:
+    '''
+    EMS readiness means a live or paper account can accept requests.
+    An empty snapshot is legal, but an empty account set cannot
+    provide that service. Reject it even with no positions or
+    dialogs.
+
+    '''
+    with pytest.raises(
+        ProtocolError,
+        match='requires an account',
+    ):
+        validate_ems_startup(
+            EmsStartup(
+                broker='ib',
+                accounts=(),
+                positions=(),
+                dialogs=(),
+            )
+        )
diff --git a/tests/test_ems_ipc.py b/tests/test_ems_ipc.py
new file mode 100644
index 00000000..a2456354
--- /dev/null
+++ b/tests/test_ems_ipc.py
@@ -0,0 +1,458 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public
+# License as published by the Free Software Foundation, either
+# version 3 of the License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program. If not, see
+# <https://www.gnu.org/licenses/>.
+
+'''
+Real local-actor qualification of clearing's directional contracts.
+
+No market-data service, credentials, or persistent account is opened.
+
+'''
+
+from decimal import Decimal
+
+import pytest
+import tractor
+from tractor.trionics import collapse_eg
+import trio
+
+from piker.clearing._codec import ProtocolError, validate
+from piker.clearing._protocol import (
+    BrokerdPhase,
+    Stream,
+    open_stream,
+)
+from piker.clearing._schema import (
+    BrokerdEvent,
+    BrokerdFill,
+    BrokerdOpenOrder,
+    BrokerdOpenOrdersComplete,
+    BrokerdOrder,
+    BrokerdOrderAck,
+    BrokerdPayload,
+    BrokerdRequest,
+    BrokerdStartup,
+    Order,
+    validate_startup,
+    BrokerdStatus,
+    Cancel,
+    EmsEvent,
+    EmsPayload,
+    EmsRequest,
+    EmsStartup,
+    Status,
+    validate_ems_startup,
+)
+from piker.clearing._state import Effects, OrderState
+
+
+@tractor.context(pld_spec=BrokerdPayload)
+async def fake_broker(
+    ctx: tractor.Context,
+    fault: str,
+    empty: bool,
+) -> None:
+    '''
+    Exchange typed broker traffic or inject a selected wire fault.
+
+    ``fault`` selects a provider exception, wrong direction, wrong
+    phase, or malformed execution. ``empty`` omits the existing order
+    while retaining the explicit completion barrier.
+
+    '''
+    startup: BrokerdStartup = BrokerdStartup(
+        broker='fixture',
+        mode='paper',
+        accounts=('paper',),
+    )
+    await ctx.started(validate_startup(startup))
+    stream: Stream[BrokerdRequest, BrokerdEvent]
+    async with open_stream(
+        ctx,
+        BrokerdRequest,
+        BrokerdEvent,
+    ) as stream:
+        if not empty:
+            await stream.send(
+                BrokerdOpenOrder(
+                    broker='fixture',
+                    reqid='existing-1',
+                    time_ns=1,
+                    order=Order(
+                        oid='fixture:paper:existing-1',
+                        account='paper',
+                        symbol='btc.spot.fixture',
+                        price=Decimal('12.25'),
+                        size=1.0,
+                        action='buy',
+                        exec_mode='live',
+                    ),
+                )
+            )
+        await stream.send(
+            BrokerdOpenOrdersComplete(
+                broker='fixture',
+            )
+        )
+        request: BrokerdOrder = await stream.receive()
+        if fault == 'peer_error':
+            raise RuntimeError('fixture provider failed')
+        if fault == 'wrong_direction':
+            # Deliberately bypass the local send guard to prove that
+            # the other actor's scoped wire decoder rejects this.
+            await stream.raw.send(request)
+        elif fault == 'wrong_phase':
+            await stream.raw.send(startup)
+        elif fault == 'malformed':
+            await stream.raw.send(
+                {
+                    'type': 'broker_fill',
+                    'time_ns': 1,
+                    'broker_time': 1.0,
+                    'reqid': 1,
+                    'account': 'paper',
+                    'broker': 'fixture',
+                    'action': 'buy',
+                    'size': 1.0,
+                    'price': 12.25,
+                }
+            )
+        else:
+            await stream.send(
+                BrokerdOrderAck(
+                    oid=request.oid,
+                    reqid=1,
+                    account='paper',
+                )
+            )
+            await stream.send(
+                BrokerdFill(
+                    time_ns=1,
+                    broker_time=1.0,
+                    reqid=1,
+                    account='paper',
+                    broker='fixture',
+                    fill_id='fill-1',
+                    action='buy',
+                    size=1.0,
+                    price=12.25,
+                )
+            )
+        # A malicious peer must not hold teardown open after its
+        # receiver rejects a payload. The parent cancels this sleep.
+        await trio.sleep_forever()
+
+
+async def exercise(fault: str = '', empty: bool = False) -> None:
+    '''
+    Own the fake broker actor tree through exchange and teardown.
+
+    ``fault`` and ``empty`` select ``fake_broker`` behavior. A Trio
+    deadline encloses all actor scopes; the test runner adds an
+    independent process timeout.
+
+    '''
+    # The deadline encloses the entire actor tree, including
+    # teardown.
+    # An outer process timeout remains necessary for shielded runtime
+    # failures; the run-tests harness supplies that second bound.
+    with trio.fail_after(15):
+        nursery: tractor.ActorNursery
+        async with (
+            collapse_eg(),
+            tractor.open_root_actor(
+                name='clearing-schema-test',
+                tpt_bind_addrs=[('127.0.0.1', 0)],
+            ),
+            tractor.open_nursery() as nursery,
+        ):
+            portal: tractor.Portal = await nursery.start_actor(
+                'fake-broker',
+                enable_modules=[__name__],
+            )
+            ctx: tractor.Context
+            startup: BrokerdStartup
+            async with portal.open_context(
+                fake_broker,
+                fault=fault,
+                empty=empty,
+            ) as (ctx, startup):
+                phase: BrokerdPhase = BrokerdPhase(startup)
+                stream: Stream[BrokerdEvent, BrokerdRequest]
+                async with open_stream(
+                    ctx,
+                    BrokerdEvent,
+                    BrokerdRequest,
+                ) as stream:
+                    if not empty:
+                        existing: BrokerdEvent = phase.receive(
+                            await stream.receive(),
+                        )
+                        assert isinstance(existing, BrokerdOpenOrder)
+                        assert isinstance(existing.order, Order)
+                        assert existing.order.price == Decimal(
+                            '12.25',
+                        )
+                    phase.receive(await stream.receive())
+                    assert phase.complete
+                    request: BrokerdOrder = BrokerdOrder(
+                        oid='client-1',
+                        account='paper',
+                        time_ns=1,
+                        symbol='btc.spot',
+                        price=Decimal('12.25'),
+                        size=1.0,
+                        action='buy',
+                    )
+                    await stream.send(phase.request(request))
+                    ack: BrokerdEvent = phase.receive(
+                        await stream.receive()
+                    )
+                    assert type(ack) is BrokerdOrderAck
+                    fill: BrokerdEvent = phase.receive(
+                        await stream.receive()
+                    )
+                    assert type(fill) is BrokerdFill
+                    assert fill.fill_id == 'fill-1'
+                    await ctx.cancel()
+            await nursery.cancel()
+
+
+@pytest.mark.parametrize('empty', [True, False])
+def test_typed_startup_and_stream(empty: bool) -> None:
+    '''
+    Typed startup must precede an explicit completion barrier, even
+    with no existing orders. Exercise both snapshots through a real
+    child process and exchange a request, ack and identified fill.
+    Nested Order and Decimal types must survive actor serialization.
+
+    '''
+    trio.run(exercise, '', empty)
+
+
+@pytest.mark.parametrize(
+    'fault',
+    [
+        'wrong_direction',
+        'wrong_phase',
+        'malformed',
+    ],
+)
+def test_wire_rejection_and_bounded_teardown(fault: str) -> None:
+    '''
+    A peer bypassing outbound checks must still fail at wire decode.
+    Send a request in the event direction, a startup in the stream,
+    or a fill without its execution ID. Require MsgTypeError and let
+    exercise's deadline cover cancellation of the sleeping producer.
+
+    '''
+    with pytest.raises(tractor.MsgTypeError):
+        trio.run(exercise, fault)
+
+
+def test_provider_failure_is_not_swallowed() -> None:
+    '''
+    Provider failure while a client waits for an ack must abort that
+    receive and terminate the actor tree. Raise inside the fake
+    broker and require its remote RuntimeError within the teardown
+    deadline.
+
+    '''
+    error: pytest.ExceptionInfo[tractor.RemoteActorError]
+    with pytest.raises(tractor.RemoteActorError) as error:
+        trio.run(exercise, 'peer_error')
+    assert error.value.boxed_type is RuntimeError
+
+
+@tractor.context(pld_spec=BrokerdPayload)
+async def wrong_final_result(
+    ctx: tractor.Context,
+) -> BrokerdStartup:
+    '''
+    Return a wire-valid startup in the forbidden final phase.
+
+    '''
+    startup: BrokerdStartup = BrokerdStartup(
+        broker='fixture',
+        mode='paper',
+        accounts=('paper',),
+    )
+    await ctx.started(startup)
+    return startup
+
+
+def test_final_result_has_its_own_phase() -> None:
+    '''
+    A lifecycle union permits startup objects but a finished broker
+    context returns only None. Receive a wire-valid wrong Return and
+    prove the phase validator rejects it after the child terminates.
+
+    '''
+
+    async def main() -> None:
+        '''
+        Receive a completed child result and reject its startup
+        payload.
+
+        '''
+        with trio.fail_after(15):
+            nursery: tractor.ActorNursery
+            async with (
+                collapse_eg(),
+                tractor.open_root_actor(
+                    tpt_bind_addrs=[('127.0.0.1', 0)],
+                ),
+                tractor.open_nursery() as nursery,
+            ):
+                portal: tractor.Portal = await nursery.start_actor(
+                    'wrong-return',
+                    enable_modules=[__name__],
+                )
+                ctx: tractor.Context
+                _: BrokerdStartup
+                async with portal.open_context(
+                    wrong_final_result,
+                ) as (ctx, _):
+                    result: BrokerdPayload = (
+                        await ctx.wait_for_result()
+                    )
+                    with pytest.raises(ProtocolError):
+                        validate(result, type(None))
+                await nursery.cancel()
+
+    trio.run(main)
+
+
+@tractor.context(pld_spec=EmsPayload)
+async def fake_ems(ctx: tractor.Context) -> None:
+    '''
+    Apply client cancellation before publishing the provider ack.
+
+    The receive order fixes the interleaving: ``OrderState.cancel()``
+    runs before ``OrderState.accept()`` handles the ack. Assert one
+    deferred cancellation and publish its terminal response.
+
+    '''
+    await ctx.started(
+        EmsStartup(
+            broker='fixture',
+            positions=(),
+            accounts=('paper',),
+            dialogs=(),
+        )
+    )
+    state: OrderState = OrderState()
+    stream: Stream[EmsRequest, EmsEvent]
+    async with open_stream(ctx, EmsRequest, EmsEvent) as stream:
+        order: Order = await stream.receive()
+        state.register(order)
+        cancel: Cancel = await stream.receive()
+        assert cancel.oid == order.oid
+        assert not state.cancel(cancel.oid).broker
+        effects: Effects = state.accept(
+            BrokerdOrderAck(
+                oid=order.oid,
+                account=order.account,
+                reqid=1,
+            )
+        )
+        assert len(effects.broker) == 1
+        assert effects.broker[0].oid == order.oid
+        effects = state.accept(
+            BrokerdStatus(
+                time_ns=1,
+                account=order.account,
+                broker='fixture',
+                reqid=1,
+                status='canceled',
+            )
+        )
+        await stream.send(effects.clients[0])
+        await trio.sleep_forever()
+
+
+def test_ems_cancel_before_ack_over_ipc() -> None:
+    '''
+    Client Order and Cancel must arrive as typed requests and produce
+    an immutable nested Status. Hold the fake provider ack until the
+    cancel reaches the remote EMS reducer, then require one deferred
+    broker cancel and a terminal client response through real IPC.
+
+    '''
+
+    async def main() -> None:
+        '''
+        Send ordered client intent and require a typed canceled
+        snapshot.
+
+        '''
+        with trio.fail_after(15):
+            nursery: tractor.ActorNursery
+            async with (
+                collapse_eg(),
+                tractor.open_root_actor(
+                    tpt_bind_addrs=[('127.0.0.1', 0)],
+                ),
+                tractor.open_nursery() as nursery,
+            ):
+                portal: tractor.Portal = await nursery.start_actor(
+                    'fake-ems',
+                    enable_modules=[__name__],
+                )
+                ctx: tractor.Context
+                startup: EmsStartup
+                async with portal.open_context(
+                    fake_ems,
+                ) as (ctx, startup):
+                    startup = validate_ems_startup(
+                        startup
+                    )
+                    assert startup.dialogs == ()
+                    stream: Stream[EmsEvent, EmsRequest]
+                    async with open_stream(
+                        ctx,
+                        EmsEvent,
+                        EmsRequest,
+                    ) as stream:
+                        order: Order = Order(
+                            oid='client-1',
+                            account='paper',
+                            symbol='btc.spot.fixture',
+                            price=Decimal('12.25'),
+                            size=1.0,
+                            action='buy',
+                            exec_mode='live',
+                        )
+                        await stream.send(order)
+                        await stream.send(
+                            Cancel(
+                                oid=order.oid,
+                                symbol=order.symbol,
+                            )
+                        )
+                        status: EmsEvent = await stream.receive()
+                        assert isinstance(status, Status)
+                        assert status.req == order
+                        assert status.resp == 'canceled'
+                        assert status.cancel_called
+                        assert isinstance(
+                            status.brokerd_event,
+                            BrokerdStatus,
+                        )
+                        await ctx.cancel()
+                await nursery.cancel()
+
+    trio.run(main)
diff --git a/tests/test_ems_messages.py b/tests/test_ems_messages.py
new file mode 100644
index 00000000..097909e9
--- /dev/null
+++ b/tests/test_ems_messages.py
@@ -0,0 +1,333 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public
+# License as published by the Free Software Foundation, either
+# version 3 of the License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program. If not, see
+# <https://www.gnu.org/licenses/>.
+
+'''
+Offline clearing schema and local API boundary conformance.
+
+'''
+
+from decimal import Decimal
+from types import UnionType
+
+import msgspec
+from msgspec import msgpack
+import pytest
+
+from piker.clearing._codec import (
+    Message,
+    ProtocolError,
+    decoder,
+    validate,
+)
+from piker.clearing._schema import (
+    BrokerdCancel,
+    BrokerdError,
+    BrokerdEvent,
+    BrokerdFill,
+    BrokerdOpenOrder,
+    BrokerdOpenOrdersComplete,
+    BrokerdOrder,
+    BrokerdOrderAck,
+    BrokerdPayload,
+    BrokerdPosition,
+    BrokerdRequest,
+    BrokerdStartup,
+    BrokerdStatus,
+    Cancel,
+    EmsEvent,
+    EmsPayload,
+    EmsRequest,
+    EmsStartup,
+    Order,
+    Status,
+    validate_startup,
+)
+
+
+ORDER: Order = Order(
+    oid='client-1',
+    symbol='btc.usdt.binance',
+    account='paper',
+    action='buy',
+    exec_mode='live',
+    price=Decimal('12.25'),
+    size=1.0,
+    brokers=('binance',),
+)
+REQUEST: BrokerdOrder = BrokerdOrder(
+    oid=ORDER.oid,
+    account=ORDER.account,
+    symbol='btc.usdt',
+    time_ns=1,
+    price=ORDER.price,
+    size=ORDER.size,
+    action='buy',
+)
+POSITION: BrokerdPosition = BrokerdPosition(
+    broker='binance',
+    account='paper',
+    symbol=ORDER.symbol,
+    size=1.0,
+    avg_price=12.25,
+)
+EVENTS: tuple[BrokerdEvent, ...] = (
+    BrokerdOrderAck(oid=ORDER.oid, reqid=7, account='paper'),
+    BrokerdStatus(
+        time_ns=1,
+        reqid=7,
+        status='open',
+        account='paper',
+        broker='binance',
+    ),
+    BrokerdFill(
+        time_ns=1,
+        broker_time=1.0,
+        reqid=7,
+        fill_id='exec-1',
+        account='paper',
+        broker='binance',
+        action='buy',
+        size=1.0,
+        price=12.25,
+    ),
+    BrokerdError(
+        account='paper',
+        broker='binance',
+        reason='rejected',
+        request=REQUEST,
+    ),
+    POSITION,
+    BrokerdOpenOrder(
+        reqid=7,
+        order=ORDER,
+        broker='binance',
+        time_ns=1,
+    ),
+    BrokerdOpenOrdersComplete(broker='binance'),
+)
+STATUS: Status = Status(
+    time_ns=1,
+    oid=ORDER.oid,
+    resp='fill',
+    req=ORDER,
+    brokerd_event=EVENTS[2],
+)
+STARTUP: BrokerdStartup = BrokerdStartup(
+    broker='binance',
+    mode='paper',
+    accounts=('paper',),
+    positions=(POSITION,),
+)
+
+
+event: BrokerdEvent
+
+
+@pytest.mark.parametrize(
+    'msg,spec',
+    [
+        (ORDER, EmsRequest),
+        (Cancel(oid=ORDER.oid, symbol=ORDER.symbol), EmsRequest),
+        (REQUEST, BrokerdRequest),
+        (
+            BrokerdCancel(
+                oid=ORDER.oid,
+                symbol=ORDER.symbol,
+                account='paper',
+                time_ns=1,
+                reqid=7,
+            ),
+            BrokerdRequest,
+        ),
+        *((event, BrokerdEvent) for event in EVENTS),
+        (STATUS, EmsEvent),
+        (STARTUP, BrokerdPayload),
+        (
+            EmsStartup(
+                broker='binance',
+                positions=(POSITION,),
+                accounts=('paper',),
+                dialogs=(STATUS,),
+            ),
+            EmsPayload,
+        ),
+    ],
+)
+def test_protocol_roundtrip(
+    msg: Message|None,
+    spec: type|UnionType,
+) -> None:
+    '''
+    Every clearing variant must round-trip through its real union,
+    including nested orders, fills, Decimal prices and startup
+    tuples. Decoding with the declared peer spec proves
+    discriminators are unique and fields are supported without hooks
+    or Any fallbacks.
+
+    '''
+    decoded: Message = decoder(spec).decode(msgpack.encode(msg))
+    assert type(decoded) is type(msg)
+    assert decoded == msg
+
+
+@pytest.mark.parametrize(
+    'msg,spec',
+    [
+        (ORDER, BrokerdRequest),
+        (REQUEST, BrokerdEvent),
+        (EVENTS[0], BrokerdRequest),
+        (STARTUP, BrokerdEvent),
+        (STATUS, EmsRequest),
+        (None, EmsRequest),
+    ],
+)
+def test_wrong_direction_and_phase(
+    msg: Message|None,
+    spec: type|UnionType,
+) -> None:
+    '''
+    A valid lifecycle message is invalid in another direction or
+    streaming phase. Test both the local send guard and wire decoder:
+    sharing a transport does not widen an endpoint's receive set.
+
+    '''
+    with pytest.raises(ProtocolError):
+        validate(msg, spec)
+    with pytest.raises(msgspec.ValidationError):
+        decoder(spec).decode(msgpack.encode(msg))
+
+
+@pytest.mark.parametrize(
+    'changes',
+    [
+        {'extra': 'unknown'},
+        {'action': 'cancel'},
+        {'size': '1.0'},
+        {'account': None},
+        {'price': []},
+    ],
+)
+def test_invalid_order_fields(changes: dict[str, object]) -> None:
+    '''
+    Foreign fields and incorrectly typed fields must fail at decode.
+    Modify a genuine encoded order so only the tested field changes;
+    successful parsing of the remaining fields cannot hide the error.
+
+    '''
+    raw: dict[str, object] = (
+        msgpack.decode(msgpack.encode(ORDER)) | changes
+    )
+    with pytest.raises(msgspec.ValidationError):
+        decoder(EmsRequest).decode(msgpack.encode(raw))
+
+
+def test_construction_is_not_validation() -> None:
+    '''
+    msgspec constructors accept an invalid status literal. The local
+    boundary must reject it before any transport send is attempted. A
+    structurally valid tagged dictionary is also rejected locally,
+    although its wire representation correctly decodes to a Struct.
+
+    '''
+    bad: BrokerdStatus = BrokerdStatus(
+        time_ns=1,
+        reqid=7,
+        status='filled',
+        account='paper',
+        broker='binance',
+    )
+    with pytest.raises(msgspec.ValidationError):
+        validate(bad, BrokerdEvent)
+    raw: dict[str, object] = msgpack.decode(msgpack.encode(ORDER))
+    with pytest.raises(ProtocolError):
+        validate(raw, EmsRequest)
+    assert decoder(EmsRequest).decode(msgpack.encode(raw)) == ORDER
+
+
+def test_immutable_snapshot_and_copy() -> None:
+    '''
+    Editing an order must not mutate a previously published snapshot.
+    Exercise the public copy method, frozen fields and nested request
+    validation so an invalid replacement cannot enter local history.
+
+    '''
+    replacement: Order = ORDER.copy(update={'price': Decimal('13')})
+    assert ORDER.price == Decimal('12.25')
+    assert replacement.price == Decimal('13')
+    with pytest.raises(AttributeError):
+        ORDER.price = Decimal('14')
+    with pytest.raises(msgspec.ValidationError):
+        STATUS.copy(
+            update={
+                'req': msgspec.structs.replace(ORDER, size='bad'),
+            }
+        )
+
+
+@pytest.mark.parametrize(
+    'changes',
+    [
+        {'accounts': ('paper', 'paper')},
+        {'accounts': (), 'positions': ()},
+        {'mode': 'live', 'accounts': (), 'positions': ()},
+        {'accounts': ('other',), 'positions': ()},
+        {'accounts': ('other',)},
+        {'broker': 'kraken'},
+        {'positions': (POSITION, POSITION)},
+        {'mode': 'paper_fallback'},
+    ],
+)
+def test_startup_ownership(changes: dict[str, object]) -> None:
+    '''
+    A typed startup can still claim another backend's positions or
+    duplicate an account. Alter one invariant at a time and verify
+    the semantic validator rejects it before EMS publishes the state.
+
+    '''
+    with pytest.raises(ProtocolError):
+        validate_startup(STARTUP.copy(update=changes))
+    assert validate_startup(STARTUP) == STARTUP
+
+
+@pytest.mark.parametrize(
+    'bad',
+    [
+        {'req': msgpack.decode(msgpack.encode(ORDER))},
+        {
+            'req': msgspec.structs.replace(
+                ORDER,
+                price=Decimal('NaN'),
+            ),
+        },
+        {'req': msgspec.structs.replace(ORDER, price='12.25')},
+        {'req': msgspec.structs.replace(ORDER, brokers=['binance'])},
+        {'req': msgspec.structs.replace(ORDER, size=float('inf'))},
+    ],
+)
+def test_local_nested_objects_are_typed(
+    bad: dict[str, object],
+) -> None:
+    '''
+    A tagged nested map can encode identically to an Order and a
+    string can encode identically to Decimal. The local API must
+    still require typed objects, immutable tuples and finite prices.
+    Replace nested fields without constructor validation and require
+    rejection before the snapshot can enter a queue or order history.
+
+    '''
+    with pytest.raises(ProtocolError):
+        validate(msgspec.structs.replace(STATUS, **bad), EmsEvent)
diff --git a/tests/test_ems_state.py b/tests/test_ems_state.py
new file mode 100644
index 00000000..96ffd809
--- /dev/null
+++ b/tests/test_ems_state.py
@@ -0,0 +1,332 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public
+# License as published by the Free Software Foundation, either
+# version 3 of the License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program. If not, see
+# <https://www.gnu.org/licenses/>.
+
+'''
+Deterministic order-event sequencing without feeds or credentials.
+
+'''
+
+from decimal import Decimal
+
+import pytest
+
+from piker.clearing._codec import ProtocolError
+from piker.clearing._schema import (
+    BrokerdCancel,
+    BrokerdError,
+    BrokerdFill,
+    BrokerdOrderAck,
+    BrokerdStatus,
+    Order,
+    Status,
+)
+from piker.clearing._state import Effects, OrderState
+
+
+def order(
+    oid: str = 'one',
+    account: str = 'paper',
+) -> Order:
+    '''
+    Build live client intent with controllable dialog/account
+    identity.
+
+    '''
+    return Order(
+        oid=oid,
+        account=account,
+        symbol='btc.usdt.binance',
+        price=Decimal('10'),
+        size=1.0,
+        action='buy',
+        exec_mode='live',
+    )
+
+
+def ack(
+    oid: str = 'one',
+    account: str = 'paper',
+    reqid: int|str = 1,
+) -> BrokerdOrderAck:
+    '''
+    Build the acknowledgement establishing an account/provider ID.
+
+    '''
+    return BrokerdOrderAck(oid=oid, account=account, reqid=reqid)
+
+
+def fill(
+    fill_id: str = 'fill-1',
+    account: str = 'paper',
+    reqid: int|str = 1,
+) -> BrokerdFill:
+    '''
+    Build an execution whose delivery may precede its
+    acknowledgement.
+
+    '''
+    return BrokerdFill(
+        time_ns=1,
+        broker_time=1.0,
+        account=account,
+        broker='binance',
+        reqid=reqid,
+        fill_id=fill_id,
+        size=0.5,
+        price=10.0,
+        action='buy',
+    )
+
+
+def closed() -> BrokerdStatus:
+    '''
+    Build a terminal provider status for the default test order.
+
+    '''
+    return BrokerdStatus(
+        time_ns=1,
+        reqid=1,
+        account='paper',
+        broker='binance',
+        status='closed',
+    )
+
+
+def test_cancel_before_ack_once() -> None:
+    '''
+    A cancel can reach EMS while broker submission is suspended.
+    Register intent, cancel twice, then repeat the ack. Only the
+    first ack may produce a cancel, carrying the original account and
+    oid. The earlier published snapshot must remain unchanged.
+
+    '''
+    state: OrderState = OrderState()
+    initial: Status = state.register(order())
+    assert not state.cancel('one').broker
+    assert not state.cancel('one').broker
+    effects: Effects = state.accept(ack())
+    assert len(effects.broker) == 1
+    cancel: BrokerdCancel = effects.broker[0]
+    assert (cancel.oid, cancel.reqid, cancel.account) == (
+        'one',
+        1,
+        'paper',
+    )
+    assert not state.accept(ack()).broker
+    assert initial.cancel_called is False
+
+
+@pytest.mark.parametrize('terminal_first', [True, False])
+def test_early_duplicate_and_late_fills(
+    terminal_first: bool,
+) -> None:
+    '''
+    Native callbacks may precede an ack and a final fill may follow
+    closed. Buffer both event orders, correlate with the ack, then
+    repeat the fill and inject a distinct late execution. Each fill
+    must be emitted once; neither may reopen the completed dialog.
+
+    '''
+    state: OrderState = OrderState()
+    state.register(order())
+    events: list[BrokerdStatus|BrokerdFill] = (
+        [closed(), fill()] if terminal_first else [fill(), closed()]
+    )
+    event: BrokerdStatus|BrokerdFill
+    for event in events:
+        assert not state.accept(event).clients
+    effects: Effects = state.accept(ack())
+    responses: list[str] = []
+    msg: Status
+    for msg in effects.clients:
+        responses.append(msg.resp)
+    assert responses == (
+        ['closed', 'fill'] if terminal_first else ['fill', 'closed']
+    )
+    assert not state.accept(fill()).clients
+    assert state.accept(fill('fill-2')).clients[0].resp == 'fill'
+    assert not state.active
+    assert state.terminal['one'].status.resp == 'closed'
+
+
+def test_closed_before_ack_cancels_no_finished_order() -> None:
+    '''
+    An order can finish while its ack is delayed and cancel is
+    queued. Replay the pre-ack terminal status before generating
+    effects so the delayed acknowledgement cannot cancel an already
+    closed order.
+
+    '''
+    state: OrderState = OrderState()
+    state.register(order())
+    state.cancel('one')
+    state.accept(closed())
+    assert not state.accept(ack()).broker
+
+
+def test_accounts_partition_request_ids() -> None:
+    '''
+    Separate broker sessions may both allocate request ID one.
+    Register and acknowledge two accounts, then route their fills.
+    Correlation must retain account ownership even when IDs collide.
+
+    '''
+    state: OrderState = OrderState()
+    oid: str
+    account: str
+    for oid, account in [('one', 'paper'), ('two', 'other')]:
+        state.register(order(oid, account))
+        state.accept(ack(oid, account))
+        effect: Effects = state.accept(fill(account=account))
+        assert effect.clients[0].oid == oid
+
+
+def test_pre_ack_capacity_is_bounded() -> None:
+    '''
+    A backend that never acknowledges must not grow EMS memory
+    forever. Exceed a deliberately tiny buffer and assert a protocol
+    failure preserves the already buffered execution instead of
+    dropping it.
+
+    '''
+    state: OrderState = OrderState(max_early=1)
+    state.accept(fill())
+    with pytest.raises(ProtocolError, match='capacity'):
+        state.accept(fill('fill-2'))
+    assert state.early[('paper', 1)] == [fill()]
+
+
+def test_tombstone_eviction_removes_correlation() -> None:
+    '''
+    Terminal dialogs have finite retention. Complete two orders with
+    a one-entry limit; evicting the first must also remove every old
+    request ID so a later event cannot target a missing dialog.
+
+    '''
+    state: OrderState = OrderState(max_terminal=1)
+    oid: str
+    reqid: int
+    for oid, reqid in [('one', 1), ('two', 2)]:
+        state.register(order(oid))
+        state.accept(ack(oid, reqid=reqid))
+        state.accept(closed().copy(update={'reqid': reqid}))
+    assert tuple(state.terminal) == ('two',)
+    assert state.reqids == {('paper', 2): 'two'}
+
+
+def test_edit_before_ack_cannot_duplicate_submission() -> None:
+    '''
+    An edit with no known backend request ID could place a second
+    order. Hold the ack and try an edit; rejection must leave the
+    original request intact for subsequent acknowledgement and
+    cancellation.
+
+    '''
+    state: OrderState = OrderState()
+    initial: Status = state.register(order())
+    with pytest.raises(
+        ProtocolError,
+        match='before acknowledgement',
+    ):
+        state.register(order().copy(update={'price': Decimal('11')}))
+    assert state.active['one'] is initial
+
+
+def test_edit_ack_keeps_late_fills_on_original_dialog() -> None:
+    '''
+    An edit may replace the provider request ID while executions for
+    the old ID remain in flight. A delayed duplicate ack must not
+    revert the cancel target, and either ID must still route fills to
+    the same EMS order. Cancel once per accepted request identity.
+
+    '''
+    state: OrderState = OrderState()
+    state.register(order())
+    state.accept(ack())
+    state.register(order().copy(update={'price': Decimal('11')}))
+    state.accept(ack(reqid=2))
+    state.accept(ack(reqid=1))
+    assert state.active['one'].reqid == 2
+    assert state.cancel('one').broker[0].reqid == 2
+    assert not state.cancel('one').broker
+    assert state.accept(fill(reqid=1)).clients[0].oid == 'one'
+    effects: Effects = state.accept(fill('fill-2', reqid=2))
+    assert effects.clients[0].oid == 'one'
+    assert state.active['one'].reqid == 2
+    state.accept(closed())
+    assert 'one' in state.active
+
+
+def test_startup_collision_does_not_register_second_order() -> None:
+    '''
+    Two existing orders cannot claim one account/request ID pair.
+    Correlate a normal order, then inject a startup order with the
+    same provider ID. Rejection leaves ``OrderState.active`` and
+    ``OrderState.dialogs`` free of the rejected second order.
+
+    '''
+    from piker.clearing._schema import BrokerdOpenOrder
+
+    state: OrderState = OrderState()
+    state.register(order())
+    state.accept(ack())
+    with pytest.raises(ProtocolError, match='ownership'):
+        state.accept(
+            BrokerdOpenOrder(
+                reqid=1,
+                order=order('two'),
+                broker='binance',
+                time_ns=1,
+            )
+        )
+    assert tuple(state.dialogs) == ('one',)
+    assert tuple(state.active) == ('one',)
+
+
+def test_submission_and_edit_rejections() -> None:
+    '''
+    A new submission error has no future ack to unblock cancellation.
+    Its typed request identifies a terminal rejection; an edit error
+    must retain the original live control ID so it can still cancel.
+    Exercise both with identical reasons to prove request semantics,
+    instead of parsing provider error strings, choose the transition.
+
+    '''
+    from piker.clearing._state import broker_request
+
+    state: OrderState = OrderState()
+    state.register(order())
+    error: BrokerdError = BrokerdError(
+        account='paper',
+        broker='binance',
+        reason='rejected',
+        request=broker_request(order()),
+    )
+    assert state.accept(error).clients[0].resp == 'error'
+    assert not state.active
+    assert 'one' in state.terminal
+
+    state = OrderState()
+    state.register(order())
+    state.accept(ack())
+    error = error.copy(
+        update={
+            'request': broker_request(order(), 1),
+        }
+    )
+    state.accept(error)
+    assert state.cancel('one').broker[0].reqid == 1