Add strict clearing schemas and `OrderState`

Introduce frozen, tagged `msgspec.Struct` payloads alongside the
legacy clearing protocol. Separate startup, request, event and
final result phases, with strict directional Tractor receive sets
and local validation before typed objects enter state or transport.

The pure reducer preserves cancel-before-ack intent, correlates
account/request IDs across edits, deduplicates identified fills and
retains bounded terminal dialogs for late events. Snapshot copies
cannot mutate previously published requests or statuses.

Also,
- require owned accounts, positions and an open-order completion
  barrier before a backend session accepts requests;
- refresh the migration plan against current Piker and Tractor
  branches, preserving the installed dependency pin;
- add 65 offline schema/state/phase tests and eight real actor
  tests for typed traffic, rejection, race handling and bounded
  teardown.
- include AGPL headers and apply the Python style guide across
  protocol code and qualification fixtures.

Ruff and all 73 targeted tests pass on installed Tractor
`5c98ab1f`. Eight IPC tests also pass against local main `5cb600a8`
with its pytest plugin isolated from Piker's duplicate `--ll`
option. Production paper/live adapters and EMS/UI consumers migrate
in subsequent paired changes; no existing endpoint switches here.

(this patch was generated in some part by `codex` using
`gpt-6` (`openai`))

ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.md

@@ -0,0 +1,45 @@
+---
+model: gpt-6
+service: codex
+timestamp: 2026-09-24T23:24:05.525363+00:00
+git_ref: 441358e2
+scope: code
+substantive: true
+raw_file: 20260924T232405Z_441358e2_schedule_coverage_prompt_io.raw.md
+---
+
+## Prompt
+
+User: "do it", approving the proposed bounded wider-window retry
+for incomplete schedule coverage and its MNQ regression test.
+Previous live qualification showed a one-day request began after
+the gap's left edge; a wider query covered it and excluded the
+one-hour venue closure.
+
+## Response summary
+
+Added bounded backwards widening under one timeout, preserving actual
+coverage and fixed end bounds. Added captured-response allocation,
+unknown-coverage, and retry-timeout tests, and updated API documentation.
+39 selected tests and Ruff passed. Live data was not changed.
+
+## Files changed
+
+- piker/brokers/ib/api.py — bounded schedule coverage retries.
+- tests/providers/test_ib_schedule.py — captured MNQ and retry regressions.
+- docs/history_segment_backfill.rst — retry contract and limits.
+
+## Human edits
+
+The user established schedule-driven allocation, explicit unknown
+coverage, and avoiding reindexing as the design constraints, supplied
+the MNQ gap, and authorized this focused coverage fix. No direct human
+source edits occurred during this patch; review is pending.
+
+## Follow-up review
+
+The user requested factoring and documenting the repeated 86400
+conversion. Added piker.calc.SECONDS_PER_DAY, documented elapsed
+versus calendar/session days once, and reused it in the IB query
+and reservation span check. Named the client retry limit locally.
+Schedule tests: 11 passed; Ruff passed after this adjustment.

ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.raw.md

@@ -0,0 +1,66 @@
+---
+model: gpt-6
+service: codex
+timestamp: 2026-09-24T23:24:05.525363+00:00
+git_ref: 441358e2
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- piker/brokers/ib/api.py`
+
+> `git diff HEAD~1..HEAD -- tests/providers/test_ib_schedule.py`
+
+> `git diff HEAD~1..HEAD -- docs/history_segment_backfill.rst`
+
+Client.history_schedule() widens incomplete left coverage by doubling
+numDays up to 31 while preserving endDateTime and actual returned
+coverage. All attempts share an asyncio timeout. Missing right coverage
+returns unchanged because extending the lookback cannot repair it.
+Tests replay the captured MNQ schedules through the real client and
+reservation allocator, check bounded incomplete coverage and retry
+cancellation under one timeout. Documentation describes these limits.
+
+## User-facing output
+
+The captured MNQ case now passes through the real client method
+and allocator: the request widens from one day to two, then
+reserves 51,824 slots while leaving the venue closure compressed.
+
+The tests also cover permanently incomplete replies and a timeout
+during the retry. I’m running the related regression suites now.
+
+All 39 selected tests passed, and Ruff passed. The reservation
+suite needed a retry outside the sandbox because Trio’s local
+wakeup socket was blocked.
+
+The fix and regression tests are ready for review. I’m recording
+the prompt provenance; live SHM and parquet remain unchanged.
+
+## Validation commands and results
+
+Worktree-local py313 in the existing Nix environment; import piker
+resolved inside this worktree.
+
+- py313/bin/python -m pytest -p no:xonsh -q -x --tb=short --no-header tests/providers/test_ib_schedule.py tests/providers/test_history_reservations.py
+  Sandbox: 11 passed then PermissionError creating Trio wakeup socket.
+  Escalated rerun: 22 passed in 1.02s; fake IB, disposable test SHM.
+- py313/bin/python -m pytest -p no:xonsh -q -x --tb=short --no-header tests/test_ib_history.py
+  17 passed in 0.69s.
+- ruff check piker/brokers/ib/api.py tests/providers/test_ib_schedule.py
+  All checks passed (existing deprecated configuration warning).
+- git diff --check: passed.
+
+No live service restart, live history repair, staging, or commit.
+
+## Follow-up user review and generated changes
+
+User: "feels like this same magic # from the prior commit should be
+factored and commented once? `num_days: int = max(1, ceil(seconds /
+86400))` other then that all looks decent"
+
+> `git diff HEAD~1..HEAD -- piker/calc.py piker/tsp/_reserve.py`
+
+Added SECONDS_PER_DAY for both consumers, with one comment explaining
+fixed elapsed days versus calendar days and trading sessions. Named
+the client's 31-day retry limit locally. After this change the
+schedule suite passed 11 tests and Ruff passed all four Python files.

docs/history_segment_backfill.rst

@@ -21,7 +21,13 @@ Schedule queries
 for UTC epoch bounds. The response includes actual coverage, timezone,
 regular-hours policy, and half-open trading sessions. This wraps
 ``Client.history_schedule()`` through the existing asyncio method proxy.
-The first pass limits a query to 31 days and times out after 10 seconds.
+The first pass limits a query to 31 days. If IB returns coverage
+starting after the requested left edge, double the lookback up to
+31 days while retaining the requested end timestamp. All attempts
+share one 10-second timeout. A missing right edge cannot be repaired
+by widening backwards; return the actual partial coverage so the
+allocator leaves that gap unresolved. The same applies when the
+31-day lookback still leaves the left edge uncovered.
 Historical queries must match bars' ``useRTH=False`` selection.
 
 Do not extrapolate sessions outside response coverage. Empty schedules,

piker/brokers/ib/api.py

@@ -363,10 +363,13 @@ class Client:
 
         Return UTC coverage separately from sessions. Missing coverage
         is unknown, never evidence that the venue was closed.
-        Match bars()'s extended-hours policy by default.
+        Match bars()'s extended-hours policy by default. Retry missing
+        left coverage with a wider lookback, up to 31 days, sharing
+        `timeout_s` across schedule requests. Keep the end fixed.
 
         '''
         from math import ceil
+        from piker.calc import SECONDS_PER_DAY
         from .venues import normalize_schedule
 
         if start_dt.tzinfo is None or end_dt.tzinfo is None:
@@ -374,23 +377,43 @@ class Client:
         seconds: float = (
             end_dt.astimezone(UTC) - start_dt.astimezone(UTC)
         ).total_seconds()
-        if not 0 < seconds <= 31 * 86400:
+        # Bound schedule lookback and retries to one elapsed month.
+        max_schedule_days: int = 31
+        if not 0 < seconds <= max_schedule_days * SECONDS_PER_DAY:
             raise ValueError('Schedule range must be within 31 days')
         contract: Contract = (await self.find_contracts(fqme))[0]
-        schedule = await asyncio.wait_for(
-            self.ib.reqHistoricalScheduleAsync(
-                contract,
-                numDays=max(1, ceil(seconds / 86400)),
-                endDateTime=end_dt.astimezone(UTC).strftime(
-                    '%Y%m%d-%H:%M:%S'
-                ),
-                useRTH=use_rth,
-            ),
-            timeout=timeout_s,
+        num_days: int = max(1, ceil(seconds / SECONDS_PER_DAY))
+        end_stamp: str = end_dt.astimezone(UTC).strftime(
+            '%Y%m%d-%H:%M:%S'
         )
+        # IB can return a single dated session for a one-day query,
+        # even when the requested interval starts in the preceding
+        # session. Widen the lookback, keeping the requested end fixed.
+        # Doubling bounds request count; 31 days caps provider work.
+        # All attempts share one timeout, rather than renewing it.
         # TODO: explicit cancellation of the underlying IB request
         # requires a request-id exposing ib_async schedule wrapper.
-        return normalize_schedule(schedule, use_rth)
+        async with asyncio.timeout(timeout_s):
+            while True:
+                schedule = await self.ib.reqHistoricalScheduleAsync(
+                    contract,
+                    numDays=num_days,
+                    endDateTime=end_stamp,
+                    useRTH=use_rth,
+                )
+                coverage: dict = normalize_schedule(schedule, use_rth)
+                if (
+                    coverage['start'] <= start_dt.timestamp()
+                    or
+                    coverage['end'] < end_dt.timestamp()
+                    or
+                    num_days == max_schedule_days
+                ):
+                    # Widening backwards cannot fix missing coverage
+                    # at the right edge. Preserve actual coverage for
+                    # the allocator to reject any unresolved interval.
+                    return coverage
+                num_days = min(max_schedule_days, num_days * 2)
 
     async def bars(
         self,

piker/calc.py

@@ -23,6 +23,11 @@ import itertools
 from bidict import bidict
 
 
+# Fixed elapsed-day conversion for timestamp arithmetic. Calendar
+# days may span 23 or 25 hours across DST; trading sessions vary too.
+SECONDS_PER_DAY: int = 24 * 60 * 60
+
+
 _mag2suffix = bidict({3: 'k', 6: 'M', 9: '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')

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

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)

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,
+    )

piker/tsp/_reserve.py

@@ -25,6 +25,7 @@ import math
 
 import numpy as np
 
+from piker.calc import SECONDS_PER_DAY
 from piker.log import get_logger
 
 log = get_logger(__name__)
@@ -102,10 +103,9 @@ async def reserve_history_gaps(
     inserts: dict[int, np.ndarray] = {}
     ranges: list[tuple[float, float]] = []
     budget: int = max_rows
-    # Match IB Client.history_schedule()'s first-pass 31-day bound;
-    # 86400 converts elapsed days to seconds, not local venue days.
+    # Match IB Client.history_schedule()'s first-pass 31-day bound.
     # TODO: advertise query limits through backend history config.
-    max_schedule_span_s: int = 31 * 86400
+    max_schedule_span_s: int = 31 * SECONDS_PER_DAY
     for offset in reversed(gaps[-newest_gaps:]):
         left: float = float(times[offset])
         right: float = float(times[offset + 1])

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

tests/providers/test_ib_schedule.py

@@ -99,11 +99,11 @@ def test_schedule_request_uses_extended_hours_and_utc():
     )
     out = asyncio.run(Client.history_schedule(
         client, 'mnq.cme.20261218',
-        datetime(2026, 9, 23, 0, tzinfo=UTC),
+        datetime(2026, 9, 23, 4, tzinfo=UTC),
         datetime(2026, 9, 24, 1, tzinfo=UTC),
     ))
     assert calls == [('contract', {
-        'numDays': 2, 'endDateTime': '20260924-01:00:00',
+        'numDays': 1, 'endDateTime': '20260924-01:00:00',
         'useRTH': False,
     })]
     assert out['timezone'] == 'America/New_York'
@@ -173,3 +173,191 @@ def test_schedule_duration_counts_elapsed_time_across_dst():
         datetime(2026, 11, 1, tzinfo=zone),
         datetime(2026, 11, 2, tzinfo=zone),
     ))
+
+
+def mnq_schedule(wider: bool = False) -> SimpleNamespace:
+    '''
+    Reproduce the dated US/Central sessions captured from live IB.
+
+    '''
+    starts: list[str] = ['20260923-17:00:00']
+    ends: list[str] = ['20260924-16:00:00']
+    if wider:
+        starts.insert(0, '20260922-17:00:00')
+        ends.insert(0, '20260923-16:00:00')
+    sessions: list[SimpleNamespace] = []
+    start: str
+    end: str
+    for start, end in zip(starts, ends):
+        sessions.append(SimpleNamespace(
+            startDateTime=start,
+            endDateTime=end,
+        ))
+    return SimpleNamespace(
+        timeZone='US/Central',
+        startDateTime=starts[0],
+        endDateTime=ends[-1],
+        sessions=sessions,
+    )
+
+
+def test_schedule_retry_reserves_mnq_gap_outside_closure():
+    '''
+    One elapsed day returned only MNQ's newest trading session.
+
+    Its coverage began after the left gap margin, so the allocator
+    could not reserve any slots. Replay both captured IB responses
+    through Client.history_schedule() and reserve_history_gaps().
+    Assert that widening keeps the end fixed, reserves both trading
+    portions, and leaves the one-hour maintenance closure compressed.
+    This uses fake transport and arrays, with no live SHM or storage.
+
+    '''
+    import numpy as np
+    from piker.tsp._reserve import reserve_history_gaps
+
+    calls: list[dict] = []
+
+    async def find_contracts(fqme: str) -> list[str]:
+        return ['contract']
+
+    async def schedule(contract: str, **kwargs) -> SimpleNamespace:
+        calls.append(kwargs)
+        return mnq_schedule(wider=len(calls) > 1)
+
+    client = SimpleNamespace(
+        find_contracts=find_contracts,
+        ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule),
+    )
+
+    async def query(left: float, right: float) -> dict:
+        return await Client.history_schedule(
+            client, 'mnq.cme.20261218',
+            datetime.fromtimestamp(left, UTC),
+            datetime.fromtimestamp(right, UTC),
+        )
+
+    frame: np.ndarray = np.zeros(2, dtype=[
+        ('time', 'f8'),
+        ('open', 'f8'),
+        ('high', 'f8'),
+        ('low', 'f8'),
+        ('close', 'f8'),
+    ])
+    frame['time'] = [1790193489, 1790248914]
+    frame['close'] = [30755.75, 30478.5]
+    result: np.ndarray
+    ranges: list[tuple[float, float]]
+    result, ranges = asyncio.run(reserve_history_gaps(
+        frame, 1, query, max_rows=51826,
+    ))
+    assert calls == [
+        {
+            'numDays': days,
+            'endDateTime': '20260924-11:21:54',
+            'useRTH': False,
+        }
+        for days in (1, 2)
+    ]
+    assert len(result) == 51826
+    assert ranges == [
+        (1790193490, 1790197200),
+        (1790200800, 1790248914),
+    ]
+    assert not np.any(
+        (result['time'] >= 1790197200)
+        & (result['time'] < 1790200800)
+    )
+    np.testing.assert_array_equal(result[[0, -1]], frame)
+
+
+@pytest.mark.parametrize('missing_right', [False, True])
+def test_schedule_retry_preserves_unknown_coverage(
+    missing_right: bool,
+):
+    '''
+    Widening must not turn a partial reply into claimed coverage.
+
+    Return the same incomplete schedule on every fake request.
+    Missing left coverage exhausts the bounded lookback; missing
+    right coverage stops immediately because an earlier start cannot
+    repair it. The unchanged response retains the unknown interval
+    so downstream reservation still rejects it.
+
+    '''
+    calls: list[int] = []
+
+    async def find_contracts(fqme: str) -> list[str]:
+        return ['contract']
+
+    async def schedule(contract: str, **kwargs) -> SimpleNamespace:
+        calls.append(kwargs['numDays'])
+        return mnq_schedule()
+
+    client = SimpleNamespace(
+        find_contracts=find_contracts,
+        ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule),
+    )
+    end: datetime = datetime(2026, 9, 24, 11, tzinfo=UTC)
+    if missing_right:
+        end = datetime(2026, 9, 24, 23, tzinfo=UTC)
+    out: dict = asyncio.run(Client.history_schedule(
+        client, 'mnq.cme.20261218',
+        datetime(2026, 9, 23, 19, tzinfo=UTC), end,
+    ))
+    assert out == normalize_schedule(mnq_schedule(), False)
+    assert calls == ([2] if missing_right else [1, 2, 4, 8, 16, 31])
+
+
+def test_schedule_retry_shares_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    '''
+    A retry must not renew the caller's total schedule wait budget.
+
+    Record timeout scopes while the first fake reply lacks coverage
+    and the second request waits forever. The shared deadline must
+    cancel that second request and propagate TimeoutError, with no
+    partial result substituted for failure. No sleep ordering is
+    required: only the blocked second request yields to the deadline.
+
+    '''
+    timeouts: list[float] = []
+    calls: int = 0
+    cancelled: bool = False
+    original_timeout = asyncio.timeout
+
+    def timeout(delay: float):
+        timeouts.append(delay)
+        return original_timeout(delay)
+
+    monkeypatch.setattr(asyncio, 'timeout', timeout)
+
+    async def find_contracts(fqme: str) -> list[str]:
+        return ['contract']
+
+    async def schedule(contract: str, **kwargs) -> SimpleNamespace:
+        nonlocal calls, cancelled
+        calls += 1
+        if calls == 1:
+            return mnq_schedule()
+        try:
+            await asyncio.Future()
+        except asyncio.CancelledError:
+            cancelled = True
+            raise
+
+    client = SimpleNamespace(
+        find_contracts=find_contracts,
+        ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule),
+    )
+    with pytest.raises(TimeoutError):
+        asyncio.run(Client.history_schedule(
+            client, 'mnq.cme.20261218',
+            datetime(2026, 9, 23, 19, tzinfo=UTC),
+            datetime(2026, 9, 24, 11, tzinfo=UTC),
+            timeout_s=0.01,
+        ))
+    assert timeouts == [0.01]
+    assert calls == 2
+    assert cancelled

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=(),
+            )
+        )

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)

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)

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