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`))
Retry incomplete IB schedule coverage
2026-10-05 04:00:45 -04:00 · 2026-09-24 20:20:20 -04:00
16 changed files with 3778 additions and 19 deletions
--- a/ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.md
+++ b/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.
--- a/ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.raw.md
+++ b/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.
--- a/docs/history_segment_backfill.rst
+++ b/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,
--- a/piker/brokers/ib/api.py
+++ b/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(
+        num_days: int = max(1, ceil(seconds / SECONDS_PER_DAY))
-            self.ib.reqHistoricalScheduleAsync(
+        end_stamp: str = end_dt.astimezone(UTC).strftime(
                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,
        )
        # 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,
--- a/piker/calc.py
+++ b/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'})
--- a/piker/clearing/_codec.py
+++ 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')
--- a/piker/clearing/_protocol.py
+++ b/piker/clearing/_protocol.py
@ -0,0 +1,210 @@
 # piker: trading gear for hackers
 # Copyright (C) Tyler Goodlet (in stewardship for pikers)
 # This program is free software: you can redistribute it and/or
 # modify it under the terms of the GNU Affero General Public
 # License as published by the Free Software Foundation, either
 # version 3 of the License, or (at your option) any later version.
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 # GNU Affero General Public License for more details.
 # You should have received a copy of the GNU Affero General Public
 # License along with this program. If not, see
 # <https://www.gnu.org/licenses/>.
 '''
 Directional clearing streams over Tractor's typed payload decoder.
 The context decorator declares the whole lifecycle. Each stream
 narrows reception to its peer's messages; startup records and return
 values cannot appear as streaming events. Phase barriers are checked
 by the EMS separately from this direction check.
 '''
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager as acm
 from types import UnionType
 from typing import Generic, Self, TypeVar
 import tractor
 import trio
 from piker.types import Struct
 from ._codec import ProtocolError, validate
 from ._schema import (
    BrokerdEvent,
    BrokerdOpenOrder,
    BrokerdOpenOrdersComplete,
    BrokerdPosition,
    BrokerdRequest,
    BrokerdStartup,
    validate_startup,
 )
 Rx = TypeVar('Rx')
 Tx = TypeVar('Tx')
 class Stream(Generic[Rx, Tx]):
    '''
    Validate outbound objects before the actor transport sees them.
    Tractor checks the wire payload on reception. The additional
    local check also protects memory-channel test doubles and
    disallows maps at the Python API boundary.
    '''
    def __init__(
        self,
        stream: tractor.MsgStream,
        incoming: type[Rx]|UnionType,
        outgoing: type[Tx]|UnionType,
    ) -> None:
        '''
        Bind the raw stream and its directional payload specs.
        '''
        self.raw: tractor.MsgStream = stream
        self.incoming: type[Rx]|UnionType = incoming
        self.outgoing: type[Tx]|UnionType = outgoing
    @property
    def _closed(self) -> bool:
        '''
        Expose the underlying stream closure state.
        '''
        return self.raw._closed
    @property
    def chan(self) -> tractor.Channel:
        '''
        Expose the Tractor channel used by this stream.
        '''
        return self.raw.chan
    async def send(self, msg: Tx) -> None:
        '''
        Validate and detach an outgoing object before IPC.
        '''
        await self.raw.send(validate(msg, self.outgoing))
    async def receive(self) -> Rx:
        '''
        Receive and validate an object against the incoming spec.
        '''
        return validate(await self.raw.receive(), self.incoming)
    def __aiter__(self) -> Self:
        '''
        Iterate the same directional receiving stream.
        '''
        return self
    async def __anext__(self) -> Rx:
        '''
        Translate normal stream completion into iterator completion.
        '''
        try:
            return await self.receive()
        except trio.EndOfChannel:
            raise StopAsyncIteration from None
@acm
 async def open_stream(
    ctx: tractor.Context,
    incoming: type[Rx]|UnionType,
    outgoing: type[Tx]|UnionType,
 ) -> AsyncIterator[Stream[Rx, Tx]]:
    '''
    Give one receiving task ownership of the directional decoder.
    '''
    with ctx.pld_rx.limit_plds(spec=incoming):
        raw: tractor.MsgStream
        async with ctx.open_stream() as raw:
            yield Stream(raw, incoming, outgoing)
 class BrokerdPhase(Struct):
    '''
    Require one startup barrier before accepting outgoing requests.
    ``BrokerdPhase.complete`` belongs to the receiving task. Existing
    orders are legal only before that barrier, including an empty
    snapshot. Account and broker ownership are checked independently
    of decoding.
    '''
    startup: BrokerdStartup
    complete: bool = False
    def __post_init__(self) -> None:
        '''
        Validate the startup snapshot before accepting events.
        '''
        self.startup = validate_startup(self.startup)
    def receive(self, msg: BrokerdEvent) -> BrokerdEvent:
        '''
        Check event ownership and advance the startup barrier once.
        '''
        msg = validate(msg, BrokerdEvent)
        if self.startup.mode == 'paper_fallback':
            raise ProtocolError('Fallback has no live event stream')
        broker: str = self.startup.broker
        if getattr(msg, 'broker', broker) != broker:
            raise ProtocolError('Event changed broker ownership')
        if isinstance(msg, BrokerdOpenOrdersComplete):
            if self.complete:
                raise ProtocolError('Duplicate startup barrier')
            self.complete = True
            return msg
        if isinstance(msg, BrokerdOpenOrder):
            if self.complete:
                raise ProtocolError('Existing order after startup')
            account: str = msg.order.account
            symbol: str = msg.order.symbol
            if not symbol.endswith(f'.{broker}'):
                raise ProtocolError('Wrong existing-order broker')
        else:
            account = msg.account
        if account not in self.startup.accounts:
            raise ProtocolError('Event has an unannounced account')
        if isinstance(msg, BrokerdPosition) and not (
            msg.symbol.endswith(f'.{broker}')
        ):
            raise ProtocolError('Position has wrong broker')
        return msg
    def request(self, msg: BrokerdRequest) -> BrokerdRequest:
        '''
        Require startup completion and an announced request account.
        '''
        msg = validate(msg, BrokerdRequest)
        if not self.complete:
            raise ProtocolError('Request before startup barrier')
        if msg.account not in self.startup.accounts:
            raise ProtocolError('Request has an unannounced account')
        return msg
--- a/piker/clearing/_schema.py
+++ b/piker/clearing/_schema.py
@ -0,0 +1,358 @@
 # piker: trading gear for hackers
 # Copyright (C) Tyler Goodlet (in stewardship for pikers)
 # This program is free software: you can redistribute it and/or
 # modify it under the terms of the GNU Affero General Public
 # License as published by the Free Software Foundation, either
 # version 3 of the License, or (at your option) any later version.
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 # GNU Affero General Public License for more details.
 # You should have received a copy of the GNU Affero General Public
 # License along with this program. If not, see
 # <https://www.gnu.org/licenses/>.
 '''
 Strict clearing protocol for the staged EMS migration.
 The legacy endpoints still use ``._messages`` until their producers
 and consumers move together. New endpoints import these schemas;
 there is deliberately no dictionary compatibility decoder.
 ``oid`` belongs to the client/EMS dialog, ``reqid`` to a provider's
 account, and ``fill_id`` to an execution. Position summaries contain
 no ledger history. Provider-native records stay inside adapters.
 '''
 from decimal import Decimal
 from typing import Annotated, Literal
 from msgspec import Meta
 from ._codec import Message, ProtocolError, validate
 Identity = Annotated[str, Meta(min_length=1)]
 RequestId = int|Identity
 TradeSide = Literal['buy', 'sell']
 OrderResponse = Literal[
    'pending',
    'open',
    'dark_open',
    'triggered',
    'closed',
    'fill',
    'canceled',
    'error',
 ]
 class Order(Message, tag='order', kw_only=True):
    '''
    Client intent for a live order, dark trigger, or alert.
    '''
    oid: Identity
    symbol: Identity
    account: Identity
    action: Literal['buy', 'sell', 'alert']
    exec_mode: Literal['dark', 'live']
    price: Decimal
    size: float  # signed: negative sells, positive buys
    brokers: tuple[Identity, ...] = ()
 class Cancel(Message, tag='cancel', kw_only=True):
    '''
    Client intent to cancel the dialog identified by ``oid``.
    '''
    oid: Identity
    symbol: Identity
 class BrokerdOrder(Message, tag='broker_order', kw_only=True):
    '''
    Submit or edit a provider order for an EMS dialog.
    ``BrokerdOrder.reqid`` is absent on a new submission and names
    the current provider order on an edit. ``BrokerdOrder.symbol``
    contains the market endpoint without its broker suffix.
    '''
    oid: Identity
    account: Identity
    time_ns: int
    symbol: Identity  # broker suffix removed at the EMS boundary
    price: Decimal
    size: float
    action: TradeSide
    reqid: RequestId|None = None
 class BrokerdCancel(Message, tag='broker_cancel', kw_only=True):
    '''
    Cancel an acknowledged provider order.
    ``BrokerdCancel.reqid`` is required; EMS retains cancellation
    intent until an acknowledgement establishes that provider ID.
    '''
    oid: Identity
    account: Identity
    time_ns: int
    symbol: Identity
    reqid: RequestId
 class BrokerdOrderAck(Message, tag='broker_ack', kw_only=True):
    '''
    Associate an EMS dialog with its account/provider ID.
    '''
    oid: Identity
    account: Identity
    reqid: RequestId
 class BrokerdStatus(Message, tag='broker_status', kw_only=True):
    '''
    Report provider control state independently of executions.
    '''
    time_ns: int
    reqid: RequestId
    account: Identity
    broker: Identity
    status: Literal['pending', 'open', 'canceled', 'closed']
    filled: float = 0.0
    remaining: float = 0.0
    reason: str = ''
 class BrokerdFill(Message, tag='broker_fill', kw_only=True):
    '''
    One execution, independently identified from its order.
    ``BrokerdFill.broker_time`` is Unix seconds for chart placement.
    Adapters normalize native clocks and execution IDs before
    sending. Repeated delivery retains ``BrokerdFill.fill_id``.
    '''
    time_ns: int
    broker_time: float
    reqid: RequestId
    fill_id: Identity
    account: Identity
    broker: Identity
    action: TradeSide
    size: float
    price: float
 class BrokerdError(Message, tag='broker_error', kw_only=True):
    '''
    Report a provider rejection with optional typed request context.
    '''
    account: Identity
    broker: Identity
    reason: Identity
    oid: Identity|None = None
    reqid: RequestId|None = None
    symbol: Identity|None = None
    request: BrokerdOrder|BrokerdCancel|None = None
 class BrokerdPosition(Message, tag='position', kw_only=True):
    '''
    Account projection; never a durable accounting transaction.
    '''
    broker: Identity
    account: Identity
    symbol: Identity  # full FQME
    size: float
    avg_price: float
    currency: str = ''
    bs_mktid: int|str|None = None
 class BrokerdOpenOrder(Message, tag='open_order', kw_only=True):
    '''
    An existing provider order emitted before the startup barrier.
    ``BrokerdOpenOrder.order.oid`` names the provider and account.
    Its symbol is a full FQME. ``BrokerdOpenOrder.reqid`` retains the
    provider's control identity.
    '''
    reqid: RequestId
    order: Order
    broker: Identity
    time_ns: int
 class BrokerdOpenOrdersComplete(
    Message,
    tag='open_orders_complete',
    kw_only=True,
 ):
    '''
    All existing orders have been emitted, including an empty set.
    '''
    broker: Identity
 BrokerdRequest = BrokerdOrder|BrokerdCancel
 BrokerdEvent = (
    BrokerdOrderAck
    |BrokerdStatus
    |BrokerdFill
    |BrokerdError
    |BrokerdPosition
    |BrokerdOpenOrder
    |BrokerdOpenOrdersComplete
 )
 class Status(Message, tag='status', kw_only=True):
    '''
    Immutable EMS snapshot with a typed cause for each transition.
    Mutable cancel and deduplication state belongs to ``Dialog``.
    Request and event causes occupy separate fields, so neither needs
    a catch-all dictionary or a recursive copy of provider history.
    '''
    time_ns: int
    oid: Identity
    resp: OrderResponse
    req: Order
    reqid: RequestId|None = None
    src: str|None = None
    cancel_called: bool = False
    brokerd_request: BrokerdRequest|None = None
    brokerd_event: BrokerdEvent|None = None
 class BrokerdStartup(Message, tag='broker_startup', kw_only=True):
    '''
    Announce owned accounts and positions, or delegate to paper.
    '''
    broker: Identity
    mode: Literal['live', 'paper', 'paper_fallback']
    accounts: tuple[Identity, ...] = ()
    positions: tuple[BrokerdPosition, ...] = ()
 class EmsStartup(Message, tag='ems_startup', kw_only=True):
    '''
    Publish owned positions and active dialogs when EMS is ready.
    '''
    broker: Identity
    positions: tuple[BrokerdPosition, ...]
    accounts: tuple[Identity, ...]
    dialogs: tuple[Status, ...]
 EmsRequest = Order|Cancel
 EmsEvent = Status|BrokerdPosition
 EmsPayload = EmsStartup|EmsRequest|EmsEvent|None
 BrokerdPayload = (
    BrokerdStartup
    |BrokerdRequest
    |BrokerdEvent
    |None
 )
 def validate_startup(msg: BrokerdStartup) -> BrokerdStartup:
    '''
    Validate account ownership before publishing a backend snapshot.
    '''
    msg = validate(msg, BrokerdStartup)
    _validate_positions(msg.broker, msg.accounts, msg.positions)
    if msg.mode == 'paper_fallback':
        if msg.accounts or msg.positions:
            raise ProtocolError('Fallback cannot publish live state')
    elif not msg.accounts:
        raise ProtocolError('Active startup requires an account')
    elif msg.mode == 'paper' and msg.accounts != ('paper',):
        raise ProtocolError(
            'Paper startup requires the paper account'
        )
    return msg
 def validate_ems_startup(msg: EmsStartup) -> EmsStartup:
    '''
    Validate account, position, and active-dialog ownership.
    '''
    msg = validate(msg, EmsStartup)
    if not msg.accounts:
        raise ProtocolError('EMS startup requires an account')
    broker: str = msg.broker
    _validate_positions(broker, msg.accounts, msg.positions)
    oids: set[str] = set()
    status: Status
    for status in msg.dialogs:
        order: Order = status.req
        if (
            status.oid != order.oid
            or status.oid in oids
            or order.account not in msg.accounts
            or not order.symbol.endswith(f'.{broker}')
            or status.resp in ('closed', 'canceled')
        ):
            raise ProtocolError('Invalid startup dialog ownership')
        oids.add(status.oid)
    return msg
 def _validate_positions(
    broker: str,
    accounts: tuple[str, ...],
    positions: tuple[BrokerdPosition, ...],
 ) -> None:
    '''
    Reject duplicate accounts or positions outside their owner.
    '''
    if len(set(accounts)) != len(accounts):
        raise ProtocolError('Duplicate startup accounts')
    keys: set[tuple[str, str]] = set()
    pos: BrokerdPosition
    for pos in positions:
        key: tuple[str, str] = (pos.account, pos.symbol)
        if (
            pos.broker != broker
            or pos.account not in accounts
            or not pos.symbol.endswith(f'.{broker}')
            or key in keys
        ):
            raise ProtocolError('Invalid startup position ownership')
        keys.add(key)
--- a/piker/clearing/_state.py
+++ b/piker/clearing/_state.py
@ -0,0 +1,470 @@
 # piker: trading gear for hackers
 # Copyright (C) Tyler Goodlet (in stewardship for pikers)
 # This program is free software: you can redistribute it and/or
 # modify it under the terms of the GNU Affero General Public
 # License as published by the Free Software Foundation, either
 # version 3 of the License, or (at your option) any later version.
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 # GNU Affero General Public License for more details.
 # You should have received a copy of the GNU Affero General Public
 # License along with this program. If not, see
 # <https://www.gnu.org/licenses/>.
 '''
 EMS order-dialog transitions, independent of actor I/O.
 ``Account`` owns accounting. ``OrderState`` only correlates order
 control IDs and produces immutable client snapshots and broker
 requests. Events can precede an ack or follow a terminal status;
 finite buffers retain those events without reopening finished orders.
 '''
 from collections import OrderedDict
 from time import time_ns
 from typing import Literal
 from msgspec import field
 from msgspec.structs import replace
 from piker.types import Struct
 from ._codec import ProtocolError, validate
 from ._schema import (
    BrokerdCancel,
    BrokerdError,
    BrokerdEvent,
    BrokerdFill,
    BrokerdOpenOrder,
    BrokerdOrder,
    BrokerdOrderAck,
    BrokerdRequest,
    BrokerdStatus,
    Order,
    Status,
 )
 class Effects(Struct, frozen=True):
    '''
    Return immutable broker requests and client status snapshots.
    '''
    broker: tuple[BrokerdRequest, ...] = ()
    clients: tuple[Status, ...] = ()
 class Dialog(Struct):
    '''
    Retain mutable cancellation and fill state for one EMS dialog.
    '''
    status: Status
    cancel_sent: set[int|str] = field(default_factory=set)
    fills: set[str] = field(default_factory=set)
    terminal: bool = False
 class OrderState(Struct):
    '''
    Own active dialogs, bounded early events and terminal tombstones.
    ``OrderState.reqids`` includes the account in each key: separate
    broker sessions can allocate the same numeric order ID. Old
    request IDs remain correlated until their tombstone is evicted
    from ``OrderState.terminal``, so a late execution following an
    edit still belongs to its original order. Capacity exhaustion
    raises without discarding early fills.
    '''
    dialogs: dict[str, Dialog] = field(default_factory=dict)
    active: dict[str, Status] = field(default_factory=dict)
    reqids: dict[tuple[str, int|str], str] = field(
        default_factory=dict,
    )
    early: dict[tuple[str, int|str], list[BrokerdEvent]] = field(
        default_factory=dict,
    )
    terminal: OrderedDict[str, Dialog] = field(
        default_factory=OrderedDict,
    )
    max_early: int = 256
    max_terminal: int = 1024
    def register(
        self,
        order: Order,
        resp: Literal['pending', 'open', 'dark_open'] = 'pending',
    ) -> Status:
        '''
        Register intent or replace an acknowledged active order.
        Keep ``OrderState.dialogs`` and ``OrderState.active``
        coherent while rejecting changes to the original order
        ownership.
        '''
        order = validate(order, Order)
        oid: str = order.oid
        prior: Dialog|None = self.dialogs.get(oid)
        if oid in self.terminal:
            raise ProtocolError('Cannot reuse a completed order ID')
        reqid: int|str|None = None
        if prior:
            previous: Order = prior.status.req
            if (
                previous.symbol != order.symbol
                or previous.account != order.account
                or previous.exec_mode != order.exec_mode
            ):
                raise ProtocolError('Edit changed order ownership')
            reqid = prior.status.reqid
            if prior.status.cancel_called:
                raise ProtocolError('Cannot edit a canceling order')
            if reqid is None and prior.status.resp != 'dark_open':
                raise ProtocolError(
                    'Cannot edit before acknowledgement',
                )
        status: Status = validate(
            Status(
                time_ns=time_ns(),
                oid=oid,
                req=order,
                reqid=reqid,
                resp=resp,
            ),
            Status,
        )
        if prior:
            prior.status = status
        else:
            self.dialogs[oid] = Dialog(status=status)
        self.active[oid] = status
        return status
    def _publish(self, status: Status) -> None:
        '''
        Refresh ``Dialog.status`` and ``OrderState.active``.
        '''
        dialog: Dialog = self.dialogs[status.oid]
        dialog.status = status
        if not dialog.terminal:
            self.active[status.oid] = status
    def _finish(self, status: Status) -> None:
        '''
        Move a completed dialog into ``OrderState.terminal``.
        Evict its entries from ``OrderState.reqids`` when terminal
        retention exceeds ``OrderState.max_terminal``.
        '''
        oid: str = status.oid
        dialog: Dialog = self.dialogs.pop(oid)
        dialog.status = status
        dialog.terminal = True
        self.active.pop(oid, None)
        self.terminal[oid] = dialog
        while len(self.terminal) > self.max_terminal:
            old_oid: str
            _: Dialog
            old_oid, _ = self.terminal.popitem(last=False)
            key: tuple[str, int|str]
            for key in tuple(self.reqids):
                if self.reqids[key] == old_oid:
                    del self.reqids[key]
    def _cancel(self, dialog: Dialog) -> tuple[BrokerdCancel, ...]:
        '''
        Emit a cancel once per ID recorded in ``Dialog.cancel_sent``.
        '''
        status: Status = dialog.status
        reqid: int|str|None = status.reqid
        if (
            reqid is None
            or reqid in dialog.cancel_sent
            or dialog.terminal
        ):
            return ()
        dialog.cancel_sent.add(reqid)
        order: Order = status.req
        return (
            BrokerdCancel(
                oid=status.oid,
                reqid=reqid,
                account=order.account,
                symbol=order.symbol,
                time_ns=time_ns(),
            ),
        )
    def cancel(self, oid: str) -> Effects:
        '''
        Retain cancel intent or finish an unsubmitted dark order.
        '''
        dialog: Dialog|None = self.dialogs.get(oid)
        if dialog is None:
            return Effects()
        status: Status = replace(
            dialog.status,
            cancel_called=True,
            time_ns=time_ns(),
        )
        if status.resp == 'dark_open':
            status = replace(status, resp='canceled')
            self._finish(status)
            return Effects(clients=(status,))
        self._publish(status)
        return Effects(broker=self._cancel(dialog))
    def _correlate(
        self,
        account: str,
        reqid: int|str,
        oid: str,
    ) -> None:
        '''
        Bind account/provider identity in ``OrderState.reqids`` once.
        '''
        key: tuple[str, int|str] = (account, reqid)
        previous: str|None = self.reqids.get(key)
        if previous is not None and previous != oid:
            raise ProtocolError('Request ID changed ownership')
        self.reqids[key] = oid
    def accept(self, event: BrokerdEvent) -> Effects:
        '''
        Apply one broker event and return immutable transition
        effects.
        ``OrderState.early`` retains uncorrelated events until an ack
        or startup order establishes their ``OrderState.reqids``
        entry. ``Dialog.fills`` deduplicates executions without
        reviving a completed dialog in ``OrderState.active``.
        '''
        event = validate(event, BrokerdEvent)
        if isinstance(event, BrokerdOpenOrder):
            order: Order = event.order
            if (
                order.oid in self.dialogs
                or order.oid in self.terminal
            ):
                raise ProtocolError('Duplicate startup order')
            key: tuple[str, int|str] = (order.account, event.reqid)
            if key in self.reqids:
                raise ProtocolError('Request ID changed ownership')
            broker: str = event.broker
            if not order.symbol.endswith(f'.{broker}'):
                raise ProtocolError('Wrong existing-order broker')
            status: Status = self.register(order, resp='open')
            status = replace(
                status,
                reqid=event.reqid,
                src=event.broker,
                time_ns=event.time_ns,
                brokerd_event=event,
            )
            self._publish(status)
            self._correlate(order.account, event.reqid, order.oid)
            return self._replay(
                order.account,
                event.reqid,
                (status,),
            )
        if isinstance(event, BrokerdOrderAck):
            oid: str|None = event.oid
            dialog: Dialog|None = (
                self.dialogs.get(oid) or self.terminal.get(oid)
            )
            if dialog is None:
                raise ProtocolError('Ack for an unknown order')
            if event.account != dialog.status.req.account:
                raise ProtocolError('Ack changed account ownership')
            key = (event.account, event.reqid)
            known: bool = key in self.reqids
            self._correlate(event.account, event.reqid, oid)
            if known and dialog.status.reqid != event.reqid:
                # A delayed duplicate ack for a replaced request ID
                # cannot revert the current cancel target.
                return Effects()
            if not dialog.terminal:
                self._publish(
                    replace(
                        dialog.status,
                        reqid=event.reqid,
                        time_ns=time_ns(),
                    )
                )
            replayed: Effects = self._replay(
                event.account,
                event.reqid,
            )
            # Replay a terminal event first: an order already closed
            # before its ack no longer needs a deferred cancel.
            cancels: tuple[BrokerdCancel, ...] = (
                self._cancel(dialog)
                if dialog.status.cancel_called
                else ()
            )
            return Effects(
                broker=replayed.broker + cancels,
                clients=replayed.clients,
            )
        if not isinstance(
            event,
            (BrokerdStatus, BrokerdFill, BrokerdError),
        ):
            raise ProtocolError('Not an order-dialog event')
        account: str = event.account
        broker: str = event.broker
        reqid: int|str|None = event.reqid
        oid = self.reqids.get((account, reqid))
        if isinstance(event, BrokerdError):
            error_oid: str|None = event.oid
            request: BrokerdRequest|None = event.request
            if request:
                if (
                    request.account != account
                    or error_oid not in (None, request.oid)
                ):
                    raise ProtocolError(
                        'Error changed request ownership',
                    )
                error_oid = request.oid
            if oid and error_oid and oid != error_oid:
                raise ProtocolError('Error changed order ownership')
            oid = error_oid or oid
        if oid is None:
            if reqid is None:
                raise ProtocolError('Event has no correlation ID')
            count: int = sum(map(len, self.early.values()))
            if count >= self.max_early:
                raise ProtocolError('Pre-ack capacity exceeded')
            self.early.setdefault((account, reqid), []).append(event)
            return Effects()
        dialog = self.dialogs.get(oid) or self.terminal.get(oid)
        if dialog is None:
            raise ProtocolError('Missing correlated dialog')
        if account != dialog.status.req.account:
            raise ProtocolError('Event changed account ownership')
        if not dialog.status.req.symbol.endswith(f'.{broker}'):
            raise ProtocolError('Event changed broker ownership')
        status = replace(
            dialog.status,
            time_ns=time_ns(),
            brokerd_event=event,
            src=event.broker,
        )
        if isinstance(event, BrokerdFill):
            if not event.fill_id:
                raise ProtocolError('Fill has no execution identity')
            if event.fill_id in dialog.fills:
                return Effects()
            dialog.fills.add(event.fill_id)
            # The event retains its execution's reqid. The snapshot
            # keeps the current control ID after a broker-side edit.
            status = replace(status, resp='fill')
            # A late fill is observable but never reopens its dialog.
            if not dialog.terminal:
                self._publish(status)
            return Effects(clients=(status,))
        if dialog.terminal:
            return Effects()
        if (
            isinstance(event, BrokerdStatus)
            and reqid != dialog.status.reqid
        ):
            # Replacing a provider order may cancel the old ID. Its
            # delayed status cannot close the replacement dialog.
            return Effects()
        if isinstance(event, BrokerdError):
            status = replace(status, resp='error')
            if (
                isinstance(event.request, BrokerdOrder)
                and event.request.reqid is None
                and dialog.status.reqid is None
            ):
                # A rejected new submission cannot receive a useful
                # ack. Retain a tombstone instead of a pending order.
                self._finish(status)
            else:
                self._publish(status)
        else:
            status = replace(
                status, resp=event.status, reqid=reqid
            )
            if event.status in ('closed', 'canceled'):
                self._finish(status)
            else:
                self._publish(status)
        return Effects(clients=(status,))
    def _replay(
        self,
        account: str,
        reqid: int|str,
        clients: tuple[Status, ...] = (),
    ) -> Effects:
        '''
        Drain correlated events from ``OrderState.early``.
        '''
        requests: tuple[BrokerdRequest, ...] = ()
        event: BrokerdEvent
        for event in self.early.pop((account, reqid), ()):
            effects: Effects = self.accept(event)
            requests += effects.broker
            clients += effects.clients
        return Effects(broker=requests, clients=clients)
 def broker_request(
    order: Order,
    reqid: int|str|None = None,
 ) -> BrokerdOrder:
    '''
    Remove exactly the final broker suffix for the provider API.
    '''
    order = validate(order, Order)
    symbol: str
    _: str
    broker: str
    (
        symbol,
        _,
        broker,
    ) = order.symbol.rpartition('.')
    if not symbol or not broker:
        raise ProtocolError('Order requires a full FQME')
    return validate(
        BrokerdOrder(
            oid=order.oid,
            reqid=reqid,
            symbol=symbol,
            time_ns=time_ns(),
            account=order.account,
            action=order.action,
            price=order.price,
            size=order.size,
        ),
        BrokerdOrder,
    )
--- a/piker/tsp/_reserve.py
+++ b/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;
+    # Match IB Client.history_schedule()'s first-pass 31-day bound.
    # 86400 converts elapsed days to seconds, not local venue days.
    # 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])
--- a/plans/opencode/typed-ems-messaging.md
+++ b/plans/opencode/typed-ems-messaging.md
@ -0,0 +1,908 @@
 # Typed EMS and Brokerd Messaging
 ## October 2026 implementation baseline
 This revision carries the August 28 design into the September Piker
 baseline (`eed58566`). Preserve IB execution-ID accounting (`5eb4fc9d`)
 and socket-disconnect propagation (`efa72980`) while changing payloads.
 The installed Tractor is `5c98ab1f`; qualification must distinguish
 that April revision from local `main` (`5cb600a8`) and the September
 context-fault development stack. Do not test against a dirty upstream
 checkout without recording its changes.
 The complete contract includes local client queues, EMS startup and
 service readiness, both order-control streams, quote normalization for
 dark/paper execution, and typed internal trigger/dialog records.
 Provider-native APIs and persistent ledger formats retain their own
 normalization boundaries. Runtime handles are not wire messages.
 Use explicit Struct validation before local state mutation or sending;
 annotations and Struct construction do not validate field values.
 Receive decoding proves wire shape, not the sender's Python type: a
 correctly tagged mapping has the same encoding as a Struct.
 Endpoint-wide unions need separate startup, receive-direction, and
 final-result checks. Validate RPC arguments explicitly until Tractor
 issue #514 supplies signature-derived `Start` contracts. Never admit
 `Any` or raw diagnostic mappings into the clearing wire schema.
 Frozen snapshots must use replacement rather than the inherited
 `Struct.copy(update=...)`, which mutates its source. Keep mutable EMS
 dialog state separate from published messages and retain execution IDs
 for fill deduplication. Position messages remain explicit summaries;
 they must never become synthetic entries in durable trade accounting.
 Port each boundary's producers and consumers together before enabling
 its strict receive set. Qualify schema errors, wrong-direction and
 wrong-phase messages, startup completion, cancel-before-ack, reordered
 and repeated fills, transport failure, and bounded teardown with
 offline provider fixtures and real local actors. Live order submission
 is outside this qualification.
 ### Qualified dependency baseline and first review boundary
 The first boundary introduces `._schema`, local validation in `._codec`,
 directional streams and lifecycle checks in `._protocol`, and a pure
 `._state.OrderState` reducer. Existing endpoints keep their original
 `._messages` imports until their paired migration. This is infrastructure
 qualification, not a claim that paper or any live adapter is ported.
 The installed baseline is Python 3.13.14, msgspec 0.19.0, Trio 0.32.0,
 and Tractor `5c98ab1fb6e2cb7781aa75845c7f323a52853c46`. Its existing
 payload APIs support this boundary without a dependency update.
 All eight actor contract tests also pass against clean local Tractor
 `main` at `5cb600a84b6e9fcac68bad04a487b83640b3e5fb`.
 The first set has 65 schema/state/phase tests and eight real IPC tests.
 These tests cover protocol fixtures and the new reducer; adapter-level
 conformance, feed-driven paper clearing, abrupt transport EOF, and
 other spawn backends remain qualification work for subsequent ports.
 That local main adds `--ll` to Tractor's pytest plugin; Piker's current
 `tests/conftest.py` adds the same option. The ordinary downstream
 invocation therefore fails before collection. The isolated API check
 uses `--noconftest -p tractor._testing.pytest` on the same test file.
 Resolve that harness collision before upgrading Piker's dependency pin.
 Do not conflate this isolated API qualification with a whole-suite pass.
 Relevant unmerged Tractor work remains a separate qualification input:
 `pld_dec_refinements` (`6bd685ad`), `auto_codecs` (`45508252`),
 `wkt/context_fault_status` (`9547c2f9`), and
 `wkt/ipc_fault_regressions` (`fe9eb738`). The context-fault worktree has
 uncommitted transport-error handling changes; it is not a clean test
 baseline. The IPC-fault branch includes `192411c4` for transport EOF
 propagation. Typed Start arguments remain upstream work; validate them
 explicitly when migrating the actual clearing contexts.
 ## Purpose
 Replace EMS's dictionary-shaped actor traffic with explicit, tagged
 `msgspec.Struct` protocols and modern Tractor payload enforcement on both
 actor boundaries:
 ```text
 order client <-> emsd <-> brokerd.<backend> / paperboi.<backend>
 ```
 The broker boundary is part of the formal protocol, not an implementation
 detail behind the EMS endpoint. Paper, Binance, Kraken, and IB must all emit
 the same startup type, consume the same request union, and produce the same
 event union before strict Tractor payload limits are enabled.
 This plan deliberately normalizes schemas and backend behavior before adding
 `pld_spec`. Enabling transport validation first would only turn existing
 status drift, startup-shape drift, and raw dictionary reconstruction into
 hard actor failures without first defining the intended behavior.
 ## Scope
 This work includes:
 - typed client-to-EMS requests and EMS-to-client events;
 - typed EMS-to-broker requests and broker-to-EMS events;
 - one typed broker startup payload shared by live and paper endpoints;
 - one typed EMS startup payload;
 - stable tagged unions suitable for native `msgspec` decoding;
 - normalized status, identifier, startup, and metadata semantics;
 - per-direction `Context.pld_rx.limit_plds()` enforcement;
 - backend-neutral conformance tests for paper, Binance, Kraken, and IB;
 - deterministic actor-free tests and real Tractor actor/IPC tests;
 - bounded handling of acknowledgement, status, cancellation, and teardown
  races;
 - removal of raw dictionary dispatch after all producers have migrated.
 This work does not include:
 - typing provider-native websocket or REST payloads inside each backend;
 - changing the datad/brokerd actor split;
 - spawning brokerd from any location other than `open_brokerd_dialog()`;
 - redesigning accounting or replacing `BrokerdPosition` with the full
  accounting `Position` type in the same patch series;
 - live order submission in CI;
 - preserving undocumented dictionary IPC between actors running different
  Piker revisions. The actor tree is treated as a same-checkout deployment.
 ## Current Contract Inventory
 ### Client and EMS boundary
 | Location | Startup payload | Stream received | Stream sent | Current issue |
 | --- | --- | --- | --- | --- |
 | `_emsd_main()` | tuple of positions map, accounts list, dialogs map | dictionaries reconstructed as `Order` or `Cancel` | `Status` and `BrokerdPosition`, often decoded as dictionaries by clients | no payload spec, tuple startup, raw matching, loose nested fields |
 | `open_ems()` / `OrderClient` | destructures tuple | `Status` and `BrokerdPosition` dictionaries | annotated as `Order | dict`, but implementation assumes struct attributes | public shape and wire shape are conflated |
 ### EMS and broker boundary
 | Endpoint | Startup payload | Stream received by backend | Stream sent by backend | Current issue |
 | --- | --- | --- | --- | --- |
 | paper `open_trade_dialog()` | `(list[BrokerdPosition], ['paper'])` | raw `BrokerdOrder` / `BrokerdCancel` dictionaries | ack, status, fill, error, position structs | raw dispatch and network-dependent simulator setup |
 | Binance `open_trade_dialog()` | `'paper'` sentinel or `(positions, accounts)` | raw request dictionaries | structs plus client-side `Status` for existing orders | mixed protocol layers and `fill` status drift |
 | Kraken `open_trade_dialog()` | `'paper'` sentinel or `(positions, accounts)` | raw request dictionaries | structs plus client-side `Status` for existing orders | `filled` status drift and race workarounds |
 | IB `open_trade_dialog()` | `(positions, tuple(accounts))` | raw request dictionaries | structs plus client-side `Status` for existing orders | status can race ack and internal event types leak into sequencing |
 The shared broker boundary already exists operationally: all four endpoints
 are long-lived Tractor contexts opened by `open_brokerd_dialog()`. The upgrade
 formalizes that de facto interface instead of allowing each backend to define
 its own wire dialect.
 ## Target Protocol
 ### Tagging and decoding policy
 Every union member must have an explicit stable tag. Use one common tag field,
 called `type`, with explicit protocol values such as `order`, `cancel`,
 `broker_ack`, `status`, `broker_fill`, `broker_error`, `position`, and
 `open_order`. Broker requests have distinct `broker_order` and
 `broker_cancel` tags so decoding cannot confuse the two directions. Do not rely on
 Python class names as wire tags.
 All newly formalized wire structs should use:
 - `tag='<stable value>'` and `tag_field='type'`;
 - `frozen=True` for wire snapshots;
 - `kw_only=True` where it improves constructor safety;
 - immutable tuple defaults; factories for mutable internal state;
 - finite `Literal` values for protocol enums;
 - `forbid_unknown_fields=True` from schema introduction; adopt the new
  schema only when both endpoints can meet it.
 The tag is the dispatch discriminator. Fields such as `action`, `status`, and
 `resp` retain only domain meaning and are not overloaded as message type tags.
 Mutable EMS dialog bookkeeping must use a separate internal state type. Do not
 make a wire struct mutable merely because current code mutates and re-sends the
 same `Status` or `Order` instance.
 Define shared scalar aliases and use them consistently:
 ```python
 RequestId = int | str
 FillId = str
 ```
 ### Client-to-EMS requests
 Define:
 ```python
 EmsRequest = Order | Cancel
 ```
 Normalize the request structs as follows:
 - `Order.action` is exactly `buy | sell | alert`.
 - `Order.exec_mode` is exactly `dark | live`.
 - `Order.oid`, `symbol`, and `account` remain required strings.
 - `Order.price` remains `Decimal`; tests must prove transport round-tripping.
 - `Order.size` remains signed `float` for this migration. Resolving the
  duplicated side/action representation is a later domain change.
 - `Cancel` has its own tag; `action='cancel'` can be removed once no local UI
  matching depends on it.
 - Actor IPC accepts structs only. Any user-facing convenience constructor for
  mappings must validate and construct a struct before entering the stream.
 ### EMS-to-client events
 Define:
 ```python
 EmsEvent = Status | BrokerdPosition
 ```
 Retain `BrokerdPosition` as the wire-level position summary for this series,
 but document it as shared by the broker and client event protocols. A later
 accounting patch may rename it to `PositionSummary` without coupling that
 change to IPC enforcement.
 Normalize `Status` so that:
 - `resp` is exactly `pending | open | dark_open | triggered | fill | closed |
  canceled | error`;
 - `req` is a required `Order`, never an unvalidated nested dictionary;
  an uncorrelated backend error must not fabricate a client dialog;
 - `brokerd_request` is `BrokerdRequest | None` and records the command that
  caused the status transition;
 - `brokerd_event` is `BrokerdEvent | None` and records the backend response;
 - the current mixed-purpose `brokerd_msg` field is removed after callers have
  migrated; neither replacement accepts an untyped dictionary;
 - an error is represented by `Status(resp='error', ...)` or by a distinct
  tagged client event, not the current incomplete `Error(Status)` subclass;
 - internal mutable dialog state is not the same object sent repeatedly over
  the wire. Build a status snapshot when broadcasting so prior history and
  multi-client delivery cannot be changed by later in-place mutation.
 ### EMS-to-broker requests
 Define:
 ```python
 BrokerdRequest = BrokerdOrder | BrokerdCancel
 ```
 Normalize both variants so that:
 - they have explicit `broker_order` and `broker_cancel` protocol tags
  independent of `action`;
 - `oid` always identifies the EMS dialog;
 - `BrokerdOrder.reqid` is `RequestId | None` because allocation policy
  differs by backend; `BrokerdCancel.reqid` is required, and EMS retains
  cancel intent until acknowledgement supplies that identity;
 - account and symbol validation failures produce a typed `BrokerdError`
  rather than logging and continuing silently;
 - updates remain `BrokerdOrder` with an existing `reqid`; a separate modify
  variant is deferred unless conformance tests prove the current distinction
  ambiguous.
 ### Broker-to-EMS events
 Define:
 ```python
 BrokerdEvent = (
    BrokerdOrderAck
    | BrokerdStatus
    | BrokerdFill
    | BrokerdError
    | BrokerdPosition
    | BrokerdOpenOrder
    | BrokerdOpenOrdersComplete
 )
 ```
 `BrokerdOpenOrder` replaces the current backend-to-EMS use of client-facing
 `Status`. It carries the provider request id, account, source backend, and a
 fully typed embedded `Order`. EMS alone translates this event into a client
 `Status` and registers it in the dialog book.
 Require `oid` on acknowledgements. Status and fill events identify the
 order by `(account, reqid)` within their broker session; EMS retains that
 correlation across edits and terminal events. For an external order, the
 adapter creates a collision-safe stable `oid` from broker, account, and
 provider `reqid` before emitting `BrokerdOpenOrder`. Its embedded
 `Order.oid` carries that value. EMS registers it without allocating a
 reverse-message identity. Service errors may omit `oid`; a typed rejected
 request can identify a submission which never received a provider ID.
 `BrokerdFill` has a required `fill_id: FillId`. Use the provider execution id
 or provider event sequence. The four current implementations must use Binance
 trade id, Kraken transaction id, IB execution id, and a distinct paper-ledger
 transaction id respectively. Price/size/time are not a safe deduplication key
 because two legitimate executions may share them. A future provider with no
 execution identity must define an adapter session-id plus monotonic sequence
 and document that deduplication does not survive brokerd restart; it may not
 silently synthesize identity from equal-valued fill fields.
 Use one broker status vocabulary:
 ```python
 BrokerdOrderState = Literal[
    'pending',
    'open',
    'canceled',
    'closed',
 ]
 ```
 Fills are events, not status values. A partial execution emits a
 `BrokerdFill` and may then emit `BrokerdStatus(status='open', ...)`. A complete
 execution emits the final `BrokerdFill` followed by
 `BrokerdStatus(status='closed', remaining=0)`. Binance `fill`, Kraken
 `filled`, and IB provider status names must be translated inside their
 backend adapters and must never cross the brokerd IPC boundary.
 ### Startup payloads
 Replace every tuple and string sentinel with tagged startup structs.
 `BrokerdStartup` contains:
 - `mode: Literal['live', 'paper', 'paper_fallback']`;
 - `broker: str`;
 - `accounts: tuple[str, ...]`;
 - `positions: tuple[BrokerdPosition, ...]`.
 A live endpoint without credentials returns
 `BrokerdStartup(mode='paper_fallback', broker=..., accounts=(), positions=())`.
 `open_brokerd_dialog()` then opens the paper endpoint, which returns the same
 type with `mode='paper'`, its paper account, and persisted positions. The
 fallback value is a typed delegation result, not an active broker snapshot: it
 is exempt from account/position and open-order-completion requirements, and
 its context returns after startup. This removes the raw `'paper'` union branch
 without hiding fallback behavior.
 Existing open orders remain `BrokerdOpenOrder` stream events because some
 providers discover them only after authenticated stream subscription. The
 contract must state that startup positions/accounts are a complete snapshot,
 while open-order synchronization completes through typed stream events. Every
 active live or paper backend emits exactly one `BrokerdOpenOrdersComplete`
 after the initial set, including when it found zero orders, so EMS can
 distinguish completion from delay and can order initial client readiness
 deterministically. A `paper_fallback` delegation does not open a stream and
 does not emit this event; the subsequently opened paper endpoint does.
 `EmsStartup` contains:
 - `broker: str`;
 - `positions: tuple[BrokerdPosition, ...]`;
 - `accounts: tuple[str, ...]`;
 - `dialogs: tuple[Status, ...]`.
 `open_ems()` may continue returning the existing nested lookup maps to callers,
 but it builds them locally from `EmsStartup`. The public convenience shape is
 therefore separated from the wire contract.
 Treat `BrokerdStartup` as the brokerd equivalent of datad's `FeedInit`, but do
 not reproduce `validate_backend()`'s long-lived dictionary compatibility
 branches. Add a broker initialization validator beside the clearing protocol
 that checks backend name, unique/non-empty accounts, position account
 membership, `position.broker == startup.broker`, unique
 `(broker, account, symbol)` position keys, and mode-specific requirements.
 Give malformed startup its own registered initialization error so failures
 retain a useful remote exception type before strict Tractor decoding takes
 over. Each account name must be non-empty; the accounts collection may be
 empty only for `paper_fallback`. Paper sessions announce exactly
 `('paper',)`. EMS readiness also requires at least one account.
 Replace `open_brokerd_dialog()`'s manual probing with one broker order-control
 endpoint resolver. It may use `get_eps(mod, 'brokerd')` as input, but it must
 select exactly one of `open_trade_dialog` or the temporary legacy
 `trades_dialogue` alias and must not confuse unrelated `norm_trade` support
 with a trading dialog. The runtime validator proves startup; the conformance
 suite proves request/event behavior. Neither mechanism should import or enable
 datad modules inside the credentialed broker actor.
 ### Payload aliases and Tractor limits
 The endpoint-wide payload aliases include startup, stream traffic, and the
 final `None` result:
 ```python
 EmsPayload = EmsStartup | EmsRequest | EmsEvent | None
 BrokerdPayload = BrokerdStartup | BrokerdRequest | BrokerdEvent | None
 ```
 Apply broad endpoint specs and narrow receive-side specs as follows:
 | Actor side | Endpoint `pld_spec` | Narrow receive spec |
 | --- | --- | --- |
 | emsd `_emsd_main()` | `EmsPayload` | `EmsRequest` around the client command loop |
 | order client | inherited from endpoint | `EmsEvent` around trade event receives |
 | backend/paper `open_trade_dialog()` | `BrokerdPayload` | `BrokerdRequest` around backend request loops |
 | emsd broker relay | inherited from endpoint | `BrokerdEvent` around broker event receives |
 The endpoint payload spec supplies the broad decoder used for `Started`, stream,
 and return payloads. `ctx.started(..., validate_pld_spec=True)` can also perform
 local started-value validation. Ordinary `MsgStream.send()` does not validate
 locally: malformed stream data raises `tractor.MsgTypeError` when the peer
 receives/decodes it. `ctx.pld_rx.limit_plds(spec=...)` narrows that receiving
 direction so a valid message from the wrong direction still fails there.
 Do not add either decorator spec until the corresponding boundary has no
 remaining legacy sender. The final enforcement must follow the proven pattern
 in `remote_gap_overlays()` and Tractor's own payload-limiting tests.
 ### Provider metadata
 Provider-native diagnostics remain inside adapters and their logs. The
 clearing wire schema has no raw-data extension point or `Any` field.
 Source backend, broker timestamp, symbol, account, order ID, request ID,
 and execution ID have explicit typed fields. Errors can carry a typed
 request. Position summaries contain no mutable accounting history.
 ## Ordering and State Semantics
 Typed messages alone do not repair ordering defects. Define these invariants
 before enforcement:
 1. A backend sends exactly one `BrokerdOrderAck` for each accepted new order.
 2. The ack establishes `oid <-> reqid` before status/fill events are exposed
   to EMS whenever the provider API permits that ordering.
 3. A backend adapter that receives provider events before it can send the ack
   buffers them by `reqid` and flushes them after ack.
 4. EMS keeps a bounded defensive pre-ack buffer because actor scheduling
   can still expose ordering bugs. The first reducer bounds event count
   and fails loudly on overflow without dropping retained executions.
   Add wall-clock expiry in its asynchronous owner when porting EMS;
   the pure reducer does not claim an age bound.
 5. A cancel received before ack marks the dialog `cancel_called`. After
   replaying buffered events, the first ack sends one typed
   `BrokerdCancel` if the dialog is still active.
 6. Duplicate ack, fill, and terminal status events are idempotent or rejected
   deterministically. Fill deduplication uses `fill_id`. They never resurrect
   a closed dialog.
 7. `closed` and `canceled` are terminal. A late fill may be recorded and
   reported according to an explicit test, but it cannot recreate mutable
   active state.
 8. Existing external orders enter through `BrokerdOpenOrder` with the backend's
   stable namespaced `oid` and are registered before broadcast.
 9. Initial open-order discovery is complete only after
   `BrokerdOpenOrdersComplete`; client readiness cannot infer completion from
   an empty or temporarily quiet stream.
 10. Order status is routed by dialog and FQME. Account authorization and
    position privacy remain a separately scoped subscription design because
    `_emsd_main()` currently receives no account authorization input. Tests in
    this series document the current position broadcast and must not claim
    unauthorized-account isolation.
 ## Implementation Phases
 ### Phase 0: Characterize the Existing Boundary
 Add tests that record current intended behavior before restructuring:
 - new order, update, cancel, ack, open, partial fill, close, and position flow;
 - cancel-before-ack;
 - status-before-ack;
 - existing provider order discovery;
 - bad account, zero size, and backend submission error as typed domain errors;
 - bad action/tag/field type as protocol decode failures, not `BrokerdError`;
 - startup with live credentials, no credentials, paper account, persisted
  positions, and no positions;
 - Binance startup containing a negative/short position;
 - IB startup containing at least two accounts with distinct contracts;
 - zero, one, and multiple existing orders followed by the explicit snapshot
  completion event;
 - disconnect while requests or events are in flight.
 Use fake streams and fake provider clients; do not require public market data
 or credentials. Capture known undesirable outcomes as explicit expected
 failures only when a same-phase fix would obscure the baseline.
 Gate: every behavior changed later has a deterministic assertion describing
 the old behavior and the target behavior.
 ### Phase 1: Define and Audit the Wire Schemas
 Introduce the strict schema in `piker/clearing/_schema.py` alongside the
 legacy `._messages` imports. Qualify it independently before moving each
 producer/consumer pair together. Remove the legacy module after every
 endpoint and public consumer has migrated; do not add dictionary bridges.
 Replace mutable literal defaults with factories. Resolve the accidental tuple
 default on `BrokerdStatus.account`. Remove or complete the unusable
 `Error.from_status()` path. Add explicit startup types and
 `BrokerdOpenOrder`. Add the broker initialization error/validator without a
 legacy dictionary fallback.
 Inventory every current mutation of `Order`, `Status`, and embedded requests.
 Introduce a mutable internal dialog-state record and construct frozen wire
 snapshots from it. Replace mutation-based `Struct.copy(update=...)` usage with
 new-instance replacement that is proven to work for frozen structs.
 Add actor-free serialization tests using the same msgpack semantics as
 Tractor:
 - each concrete type round-trips with exact type equality;
 - each union decodes to the correct concrete type by tag;
 - nested `Order`, broker event, position, `Decimal`, and optional
  identifier fields round-trip;
 - unknown tag, missing required field, wrong literal, wrong primitive type,
  malformed nested struct, and duplicate/ambiguous tag fail;
 - unknown fields fail from the first use of the new schema;
 - request and event aliases do not accept each other's members.
 Gate: all protocol unions have unique tags and can be decoded without manual
 dictionary inspection.
 ### Phase 2: Build a Backend Conformance Harness
 Create reusable conformance helpers that exercise a backend request handler
 with a memory-backed stream and fake provider client. The harness supplies the
 same typed request cases and validates only typed events. Every request and
 emitted event must be encoded and decoded through its declared union before
 assertion; `isinstance` checks alone are insufficient because direct
 `msgspec.Struct` construction does not validate annotated field types.
 The shared suite must verify:
 - startup is one `BrokerdStartup`, including typed fallback-to-paper;
 - malformed startup fails the shared broker initialization validator;
 - new order produces one correlated ack;
 - update preserves `oid` and `reqid`;
 - cancel uses the correlated request id;
 - invalid account/size produces `BrokerdError` without killing the dialog;
 - invalid action/tag/type is rejected by codec or Tractor before the handler;
 - provider statuses normalize to the four broker states;
 - partial and complete fills use `BrokerdFill` plus normalized statuses;
 - emitted positions and existing orders are typed;
 - every active live/paper startup emits one open-order completion event, even
  for zero orders; `paper_fallback` delegates without one;
 - duplicate fills are rejected by `fill_id`, while distinct equal-valued fills
  remain valid;
 - provider-native dictionaries never cross the test stream;
 - cancellation closes all handler tasks and leaves no blocked sender;
 - one malformed request fails at the protocol boundary once limits are
  enabled.
 Separate backend-neutral contract assertions from provider translation
 fixtures. Provider fixtures should be the smallest recorded websocket/API
 payload required to prove normalization. Include regressions for Binance's
 tuple-shaped error reason, string fill size, short startup position, and
 float-vs-`Decimal` open-order price; Kraken's `filled` status and open-order
 price; and IB multi-account startup association.
 Gate: the harness can fail a deliberately nonconforming fake backend for each
 contract rule.
 ### Phase 3: Port the Paper Engine as the Reference Backend
 Make paper `open_trade_dialog()` the reference implementation of
 `BrokerdPayload`:
 - return `BrokerdStartup`;
 - consume `BrokerdRequest` instances directly;
 - emit only `BrokerdEvent` instances;
 - remove `BrokerdOrder(**request_msg)` and dictionary pattern dispatch;
 - allocate a distinct deterministic `fill_id`/transaction id instead of
  reusing `oid` as `tid`;
 - separate its order protocol tests from market-feed simulation;
 - allow deterministic synthetic ticks to drive fill tests without a public
  feed.
 Do not enable endpoint `pld_spec` yet if live backends share code paths that
 still send legacy payloads.
 Gate: the complete broker conformance suite passes for paper without network,
 credentials, sleeps, or teardown suppression.
 ### Phase 4: Port Live Backend Adapters Individually
 Port one backend per independently reviewable change. A backend is complete
 only when its provider translation fixtures and shared conformance suite pass.
 Recommended order:
 1. Binance: replace raw request matching, type startup/fallback, replace
   existing-order `Status` with `BrokerdOpenOrder`, and map `fill` to fill plus
   `open/closed` events.
 2. Kraken: replace raw request matching, type startup/fallback, replace
   existing-order `Status`, map `filled` to fill plus `closed`, and formalize
   the too-fast edit/cancel behavior.
 3. IB: replace raw request indexing, type startup, replace existing-order
   `Status`, and put an explicit sequencing gate between asyncio provider
   events and ack publication.
 For every backend, core fields are extracted before constructing the event.
 The provider's original message stays inside its adapter and logs.
 Keep brokerd's enabled RPC module set minimal and do not reintroduce
 datad modules into the credentialed actor.
 Each adapter must preserve a provider execution id as `fill_id`, emit the
 initial open-order completion event, and populate a stable namespaced `oid` for
 external orders. Binance must retain negative positions in startup. IB startup
 must associate each position with the correct account/client rather than loop
 state from the last account.
 Gate: paper, Binance, Kraken, and IB all pass the same offline conformance
 contract; no backend emits a dictionary or client-facing `Status` to EMS.
 ### Phase 5: Convert EMS Broker Translation and State
 Change `open_brokerd_dialog()` to consume `BrokerdStartup` and return a typed
 relay object. Remove startup tuple/string branching and
 `BrokerdPosition(**msg)` reconstruction.
 Remove the current `client_ready` dependency from initial broker-event
 processing. Start the broker relay as soon as its stream opens, consume and
 reduce `BrokerdOpenOrder` events through `BrokerdOpenOrdersComplete`, and only
 then construct `EmsStartup` and call `_emsd_main().started()`. The resulting
 startup dialogs contain the completed initial snapshot. Events arriving before
 the first client stream exists remain in dialog state/history for startup or
 later delivery rather than blocking the broker sender or being discarded.
 Split the current large translation loop into a small typed event reducer plus
 async effects:
 - dispatch with class patterns, not mapping patterns;
 - update correlation/dialog state in a deterministic function or narrowly
  scoped state object;
 - keep that state mutable internally but emit only new frozen `Status`
  snapshots;
 - return typed client events and typed broker requests to send;
 - keep stream broadcast and logging in the async shell;
 - implement bounded pre-ack buffering and terminal-state idempotence;
 - translate `BrokerdOpenOrder` to `Status` in one place;
 - stop reading routing fields from `broker_details`.
 The reducer must be directly testable with event sequences and no actors.
 Gate: all order-state permutations pass actor-free tests, and
 `translate_and_relay_brokerd_events()` contains no struct reconstruction from
 dictionaries.
 ### Phase 6: Convert the Client Boundary
 Change `process_client_order_cmds()` to receive `EmsRequest` directly and use
 class-pattern dispatch. Return `EmsStartup` from `_emsd_main()` and construct
 the existing `open_ems()` convenience maps client-side.
 Make `OrderClient.send()` and `send_nowait()` accept `EmsRequest` only. If a
 mapping convenience API is required by a known caller, put it in an explicit
 constructor that performs local msgspec validation; do not put dictionaries on
 the actor stream. Audit UI consumers and tests so receive loops work with
 already-decoded `Status` and `BrokerdPosition` instances.
 Replace `OrderClient.update*()` and UI mutations with construction of a new
 `Order`. Audit mutations of `Status.req`, `Status.resp`, and embedded orders in
 order mode and clearing code; all should update internal state or replace a
 wire value, never mutate a frozen payload.
 Gate: there are no `Order(**cmd)`, `Cancel(**cmd)`, `Status(**msg)`, or
 `BrokerdPosition(**msg)` conversions on the EMS client boundary.
 ### Phase 7: Enable Tractor Payload Enforcement
 Enable one boundary at a time:
 1. Add `BrokerdPayload` to every paper/live `open_trade_dialog()` decorator.
 2. Limit backend receive scopes to `BrokerdRequest`.
 3. Limit EMS's broker receive scope to `BrokerdEvent`.
 4. Add `EmsPayload` to `_emsd_main()`.
 5. Limit EMS's client receive scope to `EmsRequest`.
 6. Limit order-client receive scopes to `EmsEvent`.
 7. Verify every producer satisfies the already-strict wire schemas.
 At each step, test invalid startup, stream, wrong-direction, and return
 payloads. Assert where decoding occurs, `tractor.MsgTypeError` provenance,
 context cancellation, and bounded teardown rather than assuming every sender
 raises locally. Add an actor-free metadata test that each paper/Binance/Kraken/
 IB endpoint's `_tractor_context_meta['pld_spec']` resolves to `BrokerdPayload`
 and `_emsd_main()` resolves to `EmsPayload`.
 Gate: no valid wrong-direction type or malformed payload reaches business
 logic, and malformed traffic cannot wedge the actor tree.
 ### Phase 8: Real Actor and Lifecycle Qualification
 Add test-only Tractor context endpoints in a dedicated test fixture module to
 exercise actual encoding, process boundaries, and cancellation. Do not mock
 `MsgStream` in this layer.
 The real-actor matrix includes:
 - valid startup and bidirectional exchange for both protocol families;
 - exact concrete types observed after remote decode;
 - invalid `Started`, stream send, wrong-direction send, and return payload;
 - optional local `Started` validation and peer receive-side stream/return
  validation;
 - child failure before and after `ctx.started()`;
 - client cancellation while backend send is blocked;
 - backend cancellation while EMS waits for an event;
 - simultaneous stream close and terminal status;
 - repeated open/close cycles with no stale subscribers, tasks, actors, or SHM;
 - two clients with distinct dialogs and no cross-delivery of order status;
 - deterministic ack/status and cancel/ack race barriers;
 - brokerd restart/reconnect policy, explicitly failing if unsupported rather
  than hanging.
 Use Trio events/channels as barriers. Do not use timing sleeps to force races.
 Use bounded `fail_after()` scopes around every scenario. Do not use
 `run_and_tollerate_cancels()` for these tests; expected cancellation must be
 asserted precisely.
 Gate: the offline real-actor suite passes repeatedly under the repository's
 supported Tractor spawn methods and fails loudly within a bounded timeout.
 ### Phase 9: Remove Compatibility Paths
 After all strict tests pass:
 - delete mapping-pattern IPC dispatch and `Struct(**wire_dict)` conversions;
 - delete the `'paper'` startup sentinel and startup tuple handling;
 - delete backend emission of client `Status`;
 - delete obsolete status aliases such as `fill` and `filled`;
 - delete legacy `trades_dialogue` lookup if no backend implements it;
 - remove stale dictionary type annotations and compatibility comments;
 - remove expected failures introduced in Phase 0;
 - update backend author documentation with the protocol aliases and
  conformance command.
 Gate: searches for the removed patterns are empty or have an explicit local,
 non-IPC justification.
 ## Test Layout
 Use focused flat modules consistent with the existing test tree:
 | Test module | Responsibility | Network |
 | --- | --- | --- |
 | `tests/test_ems_messages.py` | union tags, codec round-trips, invalid schemas, strict fields | never |
 | `tests/test_brokerd_protocol.py` | shared backend conformance and provider translation fixtures | never |
 | `tests/test_ems_state.py` | reducer transitions, correlation, races, idempotence, routing | never |
 | `tests/test_ems_ipc.py` | real Tractor actor payload limits and lifecycle | never |
 | `tests/test_ems.py` | public `open_ems()` behavior, persistence, paper integration | deterministic by default |
 | backend-specific live qualification | authenticated smoke tests only | opt-in |
 The default CI lane runs every actor-free and offline actor test. Authenticated
 or public-network checks use explicit markers and are never the only coverage
 for a protocol behavior.
 ## Required Test Matrix
 ### Schema validation
 - every valid concrete message;
 - every startup mode;
 - `int` and `str` request identifiers;
 - zero, positive, and negative sizes where semantically permitted;
 - `Decimal` prices;
 - optional and populated typed request/event causes;
 - unknown/missing/wrong tags;
 - unknown fields rejected from schema introduction;
 - wrong literal and primitive field types;
 - malformed nested order/event/position;
 - a request sent in the event direction and vice versa.
 ### Dialog transitions
 - submit -> ack -> pending -> open;
 - submit -> ack -> fill -> open -> fill -> closed;
 - submit -> ack -> canceled;
 - submit -> cancel -> ack -> broker cancel -> canceled;
 - submit -> provider error;
 - status -> ack reordered delivery;
 - fill -> ack reordered delivery;
 - duplicate ack;
 - duplicate `fill_id` and distinct equal-valued fills;
 - duplicate and conflicting terminal statuses;
 - late fill after close;
 - update before and after ack;
 - cancel and update for unknown dialog;
 - existing external open order;
 - position update with and without an active order.
 ### Backend conformance
 - paper, Binance, Kraken, and IB request acceptance;
 - account rejection;
 - provider status normalization;
 - provider exception to `BrokerdError` conversion;
 - open-order conversion;
 - zero-order and populated open-order snapshot completion;
 - startup account/position conversion;
 - Binance negative-position startup and IB multi-account association;
 - provider metadata isolation;
 - backend handler cancellation and stream closure.
 ### Actor semantics
 - valid typed startup and streams;
 - local `Started` and peer receive-side `MsgTypeError` provenance;
 - malformed startup cancels context cleanly;
 - malformed stream traffic cancels both ends cleanly;
 - caller and callee cancellation;
 - no swallowed teardown errors;
 - no actor/task/channel leaks;
 - repeated execution to expose scheduling-sensitive failures.
 ### Routing and isolation
 - event reaches the originating dialog subscribers;
 - unrelated FQME subscriber does not receive it;
 - current all-client position broadcast is explicitly characterized as a
  known limitation pending an account authorization/subscription schema;
 - disconnected clients are removed from subscriber/dialog tables;
 - reconnect receives only the documented startup snapshot and active dialogs.
 ## CI and Qualification Gates
 Run gates in increasing cost order:
 1. schema and reducer unit tests;
 2. backend conformance tests with fake provider clients;
 3. real Tractor IPC tests;
 4. deterministic paper EMS integration;
 5. the broader clearing/service regression set;
 6. repeated race tests under each supported spawn method;
 7. opt-in live read-only startup for Binance, Kraken, and IB;
 8. manually authorized live order qualification with minimal size.
 No live credential or public-network result can waive an offline failure.
 Never place an order as part of default CI.
 ## Commit Boundaries
 Keep the implementation reviewable and bisectable:
 1. add the refreshed dependency baseline, strict schemas, pure reducer,
   lifecycle guards, and offline/actor protocol qualification;
 2. characterize and port paper together with its EMS consumer;
 3. port each live adapter and its paired EMS handling with provider
   translation fixtures and execution-identity tests;
 4. migrate client/EMS and UI consumers together;
 5. enforce each migrated endpoint pair with directional limits and
   lifecycle/race/failure qualification;
 6. remove legacy paths and document backend requirements.
 Every commit must leave deterministic tests green. The first boundary
 keeps legacy endpoints operational while independently qualifying the
 new protocol. Subsequent migrations move both ends together; no commit
 requires a raw-dictionary bridge into the strict schema.
 ## Rollback Strategy
 Revert a backend port with its paired EMS changes. Once enforcement
 begins, include the matching decorator and receive limits in that
 rollback; never leave one actor side strict while restoring a
 dictionary sender.
 No long-lived persisted wire data requires a schema migration. Accounting and
 ledger files remain outside this protocol change. If a strictness regression
 is discovered, repair the producer or roll back the complete endpoint pair.
 Do not widen the schema to accept arbitrary dictionaries or unknown fields.
 ## Exit Criteria
 The formal EMS messaging upgrade is complete when:
 - all client/EMS and EMS/backend IPC payloads are tagged structs;
 - paper, Binance, Kraken, and IB share one broker startup/request/event
  contract;
 - both Tractor context families declare broad `pld_spec` aliases and narrow
  each receive direction;
 - startup, stream, and return payload failures produce tested
  `MsgTypeError` behavior;
 - no broker status outside `pending/open/canceled/closed` crosses IPC;
 - fills and existing orders use dedicated variants;
 - no business logic reconstructs wire dictionaries into message structs;
 - ack/status/cancel races are deterministic and bounded;
 - multi-client order routing is tested and the unresolved position-broadcast
  privacy limitation is explicit;
 - default EMS tests run without network or credentials;
 - actor teardown failures are asserted rather than tolerated;
 - backend authors have one conformance suite and protocol definition to follow.
 ## Primary References
 - `piker/clearing/_messages.py`: current EMS and broker message structs.
 - `piker/clearing/_ems.py`: broker dialog opening, translation, client command
  processing, and `_emsd_main()`.
 - `piker/clearing/_client.py`: public EMS client and startup handling.
 - `piker/clearing/_paper_engine.py`: paper reference backend.
 - `piker/brokers/binance/broker.py`: Binance request and event adapter.
 - `piker/brokers/kraken/broker.py`: Kraken request and event adapter.
 - `piker/brokers/ib/broker.py`: IB request and event adapter.
 - `piker/data/validate.py`: existing datad-side `FeedInit` backend validation
  direction to improve upon rather than duplicate.
 - `piker/ui/_remote_ctl.py`: in-tree `pld_spec` and per-direction
  `limit_plds()` precedent.
 - `piker/ui/_gaps.py`: in-tree tagged union precedent.
 - `tests/test_ems.py`: current network-dependent clearing coverage and empty
  order tests.
 - `/home/goodboy/repos/tractor/tests/msg/test_pldrx_limiting.py`: canonical
  payload limiting and `MsgTypeError` behavior.
 - `/home/goodboy/repos/tractor/tests/msg/test_ext_types_msgspec.py`: canonical
  msgspec codec behavior.
 - `/home/goodboy/repos/tractor/tests/test_context_stream_semantics.py`:
  context stream cancellation and serialization semantics.
 Relevant issue/design threads:
 - https://github.com/goodboy/tractor/issues/196
 - https://github.com/goodboy/tractor/issues/365
 - https://github.com/goodboy/tractor/issues/514
 - https://github.com/goodboy/tractor/pull/376
 - https://github.com/pikers/piker/issues/514
 - https://github.com/pikers/piker/issues/297
 - https://github.com/pikers/piker/issues/466
--- a/tests/providers/test_ib_schedule.py
+++ b/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
--- a/tests/test_brokerd_protocol.py
+++ b/tests/test_brokerd_protocol.py
@ -0,0 +1,214 @@
 # piker: trading gear for hackers
 # Copyright (C) Tyler Goodlet (in stewardship for pikers)
 # This program is free software: you can redistribute it and/or
 # modify it under the terms of the GNU Affero General Public
 # License as published by the Free Software Foundation, either
 # version 3 of the License, or (at your option) any later version.
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 # GNU Affero General Public License for more details.
 # You should have received a copy of the GNU Affero General Public
 # License along with this program. If not, see
 # <https://www.gnu.org/licenses/>.
 '''
 Offline lifecycle conformance for every future brokerd adapter.
 '''
 from decimal import Decimal
 import pytest
 from piker.clearing._codec import ProtocolError
 from piker.clearing._protocol import BrokerdPhase
 from piker.clearing._schema import (
    BrokerdOpenOrder,
    BrokerdOpenOrdersComplete,
    BrokerdOrder,
    BrokerdPosition,
    BrokerdStartup,
    EmsStartup,
    Order,
    Status,
    validate_ems_startup,
 )
@pytest.mark.parametrize('broker', ['ib', 'kraken', 'binance'])
@pytest.mark.parametrize('mode', ['live', 'paper'])
 def test_startup_barrier(broker: str, mode: str) -> None:
    '''
    Every provider and paper mode shares one lifecycle contract.
    Prohibit order submission before completion, allow an empty
    snapshot, then reject a repeated completion or existing order.
    These fixtures qualify the protocol, not the unmigrated adapters.
    '''
    account: str = 'paper' if mode == 'paper' else f'{broker}.test'
    phase: BrokerdPhase = BrokerdPhase(
        BrokerdStartup(
            broker=broker,
            mode=mode,
            accounts=(account,),
        )
    )
    request: BrokerdOrder = BrokerdOrder(
        oid='one',
        account=account,
        symbol='btc.spot',
        time_ns=1,
        price=Decimal('10'),
        size=1.0,
        action='buy',
    )
    with pytest.raises(ProtocolError, match='before startup'):
        phase.request(request)
    complete: BrokerdOpenOrdersComplete = BrokerdOpenOrdersComplete(
        broker=broker
    )
    assert phase.receive(complete) == complete
    assert phase.request(request) == request
    with pytest.raises(ProtocolError, match='Duplicate'):
        phase.receive(complete)
    with pytest.raises(ProtocolError, match='after startup'):
        phase.receive(
            BrokerdOpenOrder(
                reqid=1,
                time_ns=1,
                broker=broker,
                order=Order(
                    oid='existing',
                    account=account,
                    symbol=f'btc.spot.{broker}',
                    price=Decimal('10'),
                    size=1.0,
                    action='buy',
                    exec_mode='live',
                ),
            )
        )
 def test_paper_fallback_never_opens_live_stream() -> None:
    '''
    A backend without credentials hands control to a paper endpoint.
    Its fallback startup cannot masquerade as an empty live snapshot:
    even a syntactically valid completion event must be rejected.
    '''
    phase: BrokerdPhase = BrokerdPhase(
        BrokerdStartup(
            broker='ib',
            mode='paper_fallback',
        )
    )
    with pytest.raises(ProtocolError, match='Fallback'):
        phase.receive(BrokerdOpenOrdersComplete(broker='ib'))
    assert not phase.complete
@pytest.mark.parametrize(
    'changes',
    [
        {'broker': 'kraken'},
        {'account': 'other'},
        {'symbol': 'btc.spot.kraken'},
    ],
 )
 def test_event_ownership(changes: dict[str, str]) -> None:
    '''
    A well-typed event may still belong to another session. Announce
    one broker/account, then alter a position's owner or market
    suffix. ``BrokerdPhase.receive()`` must reject it before EMS can
    publish the position.
    '''
    phase: BrokerdPhase = BrokerdPhase(
        BrokerdStartup(
            broker='ib',
            mode='live',
            accounts=('ib.test',),
        )
    )
    position: BrokerdPosition = BrokerdPosition(
        broker='ib',
        account='ib.test',
        symbol='btc.spot.ib',
        size=1.0,
        avg_price=10.0,
    )
    with pytest.raises(ProtocolError):
        phase.receive(position.copy(update=changes))
@pytest.mark.parametrize(
    'field,value',
    [
        ('oid', 'mismatch'),
        ('resp', 'closed'),
        ('resp', 'canceled'),
    ],
 )
 def test_ems_snapshot_has_only_active_owned_dialogs(
    field: str,
    value: str,
 ) -> None:
    '''
    Reconnect snapshots must not revive completed orders or associate
    a request with another oid. Corrupt one field in a valid snapshot
    and require semantic rejection after ordinary schema decoding.
    '''
    order: Order = Order(
        oid='one',
        account='paper',
        symbol='btc.spot.ib',
        price=Decimal('10'),
        size=1.0,
        action='buy',
        exec_mode='live',
    )
    status: Status = Status(
        time_ns=1, oid='one', resp='open', req=order
    )
    startup: EmsStartup = EmsStartup(
        broker='ib',
        accounts=('paper',),
        positions=(),
        dialogs=(status,),
    )
    assert validate_ems_startup(startup) == startup
    bad: EmsStartup = startup.copy(
        update={
            'dialogs': (status.copy(update={field: value}),),
        }
    )
    with pytest.raises(ProtocolError):
        validate_ems_startup(bad)
 def test_ems_startup_requires_an_account() -> None:
    '''
    EMS readiness means a live or paper account can accept requests.
    An empty snapshot is legal, but an empty account set cannot
    provide that service. Reject it even with no positions or
    dialogs.
    '''
    with pytest.raises(
        ProtocolError,
        match='requires an account',
    ):
        validate_ems_startup(
            EmsStartup(
                broker='ib',
                accounts=(),
                positions=(),
                dialogs=(),
            )
        )
--- a/tests/test_ems_ipc.py
+++ b/tests/test_ems_ipc.py
@ -0,0 +1,458 @@
 # piker: trading gear for hackers
 # Copyright (C) Tyler Goodlet (in stewardship for pikers)
 # This program is free software: you can redistribute it and/or
 # modify it under the terms of the GNU Affero General Public
 # License as published by the Free Software Foundation, either
 # version 3 of the License, or (at your option) any later version.
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 # GNU Affero General Public License for more details.
 # You should have received a copy of the GNU Affero General Public
 # License along with this program. If not, see
 # <https://www.gnu.org/licenses/>.
 '''
 Real local-actor qualification of clearing's directional contracts.
 No market-data service, credentials, or persistent account is opened.
 '''
 from decimal import Decimal
 import pytest
 import tractor
 from tractor.trionics import collapse_eg
 import trio
 from piker.clearing._codec import ProtocolError, validate
 from piker.clearing._protocol import (
    BrokerdPhase,
    Stream,
    open_stream,
 )
 from piker.clearing._schema import (
    BrokerdEvent,
    BrokerdFill,
    BrokerdOpenOrder,
    BrokerdOpenOrdersComplete,
    BrokerdOrder,
    BrokerdOrderAck,
    BrokerdPayload,
    BrokerdRequest,
    BrokerdStartup,
    Order,
    validate_startup,
    BrokerdStatus,
    Cancel,
    EmsEvent,
    EmsPayload,
    EmsRequest,
    EmsStartup,
    Status,
    validate_ems_startup,
 )
 from piker.clearing._state import Effects, OrderState
@tractor.context(pld_spec=BrokerdPayload)
 async def fake_broker(
    ctx: tractor.Context,
    fault: str,
    empty: bool,
 ) -> None:
    '''
    Exchange typed broker traffic or inject a selected wire fault.
    ``fault`` selects a provider exception, wrong direction, wrong
    phase, or malformed execution. ``empty`` omits the existing order
    while retaining the explicit completion barrier.
    '''
    startup: BrokerdStartup = BrokerdStartup(
        broker='fixture',
        mode='paper',
        accounts=('paper',),
    )
    await ctx.started(validate_startup(startup))
    stream: Stream[BrokerdRequest, BrokerdEvent]
    async with open_stream(
        ctx,
        BrokerdRequest,
        BrokerdEvent,
    ) as stream:
        if not empty:
            await stream.send(
                BrokerdOpenOrder(
                    broker='fixture',
                    reqid='existing-1',
                    time_ns=1,
                    order=Order(
                        oid='fixture:paper:existing-1',
                        account='paper',
                        symbol='btc.spot.fixture',
                        price=Decimal('12.25'),
                        size=1.0,
                        action='buy',
                        exec_mode='live',
                    ),
                )
            )
        await stream.send(
            BrokerdOpenOrdersComplete(
                broker='fixture',
            )
        )
        request: BrokerdOrder = await stream.receive()
        if fault == 'peer_error':
            raise RuntimeError('fixture provider failed')
        if fault == 'wrong_direction':
            # Deliberately bypass the local send guard to prove that
            # the other actor's scoped wire decoder rejects this.
            await stream.raw.send(request)
        elif fault == 'wrong_phase':
            await stream.raw.send(startup)
        elif fault == 'malformed':
            await stream.raw.send(
                {
                    'type': 'broker_fill',
                    'time_ns': 1,
                    'broker_time': 1.0,
                    'reqid': 1,
                    'account': 'paper',
                    'broker': 'fixture',
                    'action': 'buy',
                    'size': 1.0,
                    'price': 12.25,
                }
            )
        else:
            await stream.send(
                BrokerdOrderAck(
                    oid=request.oid,
                    reqid=1,
                    account='paper',
                )
            )
            await stream.send(
                BrokerdFill(
                    time_ns=1,
                    broker_time=1.0,
                    reqid=1,
                    account='paper',
                    broker='fixture',
                    fill_id='fill-1',
                    action='buy',
                    size=1.0,
                    price=12.25,
                )
            )
        # A malicious peer must not hold teardown open after its
        # receiver rejects a payload. The parent cancels this sleep.
        await trio.sleep_forever()
 async def exercise(fault: str = '', empty: bool = False) -> None:
    '''
    Own the fake broker actor tree through exchange and teardown.
    ``fault`` and ``empty`` select ``fake_broker`` behavior. A Trio
    deadline encloses all actor scopes; the test runner adds an
    independent process timeout.
    '''
    # The deadline encloses the entire actor tree, including
    # teardown.
    # An outer process timeout remains necessary for shielded runtime
    # failures; the run-tests harness supplies that second bound.
    with trio.fail_after(15):
        nursery: tractor.ActorNursery
        async with (
            collapse_eg(),
            tractor.open_root_actor(
                name='clearing-schema-test',
                tpt_bind_addrs=[('127.0.0.1', 0)],
            ),
            tractor.open_nursery() as nursery,
        ):
            portal: tractor.Portal = await nursery.start_actor(
                'fake-broker',
                enable_modules=[__name__],
            )
            ctx: tractor.Context
            startup: BrokerdStartup
            async with portal.open_context(
                fake_broker,
                fault=fault,
                empty=empty,
            ) as (ctx, startup):
                phase: BrokerdPhase = BrokerdPhase(startup)
                stream: Stream[BrokerdEvent, BrokerdRequest]
                async with open_stream(
                    ctx,
                    BrokerdEvent,
                    BrokerdRequest,
                ) as stream:
                    if not empty:
                        existing: BrokerdEvent = phase.receive(
                            await stream.receive(),
                        )
                        assert isinstance(existing, BrokerdOpenOrder)
                        assert isinstance(existing.order, Order)
                        assert existing.order.price == Decimal(
                            '12.25',
                        )
                    phase.receive(await stream.receive())
                    assert phase.complete
                    request: BrokerdOrder = BrokerdOrder(
                        oid='client-1',
                        account='paper',
                        time_ns=1,
                        symbol='btc.spot',
                        price=Decimal('12.25'),
                        size=1.0,
                        action='buy',
                    )
                    await stream.send(phase.request(request))
                    ack: BrokerdEvent = phase.receive(
                        await stream.receive()
                    )
                    assert type(ack) is BrokerdOrderAck
                    fill: BrokerdEvent = phase.receive(
                        await stream.receive()
                    )
                    assert type(fill) is BrokerdFill
                    assert fill.fill_id == 'fill-1'
                    await ctx.cancel()
            await nursery.cancel()
@pytest.mark.parametrize('empty', [True, False])
 def test_typed_startup_and_stream(empty: bool) -> None:
    '''
    Typed startup must precede an explicit completion barrier, even
    with no existing orders. Exercise both snapshots through a real
    child process and exchange a request, ack and identified fill.
    Nested Order and Decimal types must survive actor serialization.
    '''
    trio.run(exercise, '', empty)
@pytest.mark.parametrize(
    'fault',
    [
        'wrong_direction',
        'wrong_phase',
        'malformed',
    ],
 )
 def test_wire_rejection_and_bounded_teardown(fault: str) -> None:
    '''
    A peer bypassing outbound checks must still fail at wire decode.
    Send a request in the event direction, a startup in the stream,
    or a fill without its execution ID. Require MsgTypeError and let
    exercise's deadline cover cancellation of the sleeping producer.
    '''
    with pytest.raises(tractor.MsgTypeError):
        trio.run(exercise, fault)
 def test_provider_failure_is_not_swallowed() -> None:
    '''
    Provider failure while a client waits for an ack must abort that
    receive and terminate the actor tree. Raise inside the fake
    broker and require its remote RuntimeError within the teardown
    deadline.
    '''
    error: pytest.ExceptionInfo[tractor.RemoteActorError]
    with pytest.raises(tractor.RemoteActorError) as error:
        trio.run(exercise, 'peer_error')
    assert error.value.boxed_type is RuntimeError
@tractor.context(pld_spec=BrokerdPayload)
 async def wrong_final_result(
    ctx: tractor.Context,
 ) -> BrokerdStartup:
    '''
    Return a wire-valid startup in the forbidden final phase.
    '''
    startup: BrokerdStartup = BrokerdStartup(
        broker='fixture',
        mode='paper',
        accounts=('paper',),
    )
    await ctx.started(startup)
    return startup
 def test_final_result_has_its_own_phase() -> None:
    '''
    A lifecycle union permits startup objects but a finished broker
    context returns only None. Receive a wire-valid wrong Return and
    prove the phase validator rejects it after the child terminates.
    '''
    async def main() -> None:
        '''
        Receive a completed child result and reject its startup
        payload.
        '''
        with trio.fail_after(15):
            nursery: tractor.ActorNursery
            async with (
                collapse_eg(),
                tractor.open_root_actor(
                    tpt_bind_addrs=[('127.0.0.1', 0)],
                ),
                tractor.open_nursery() as nursery,
            ):
                portal: tractor.Portal = await nursery.start_actor(
                    'wrong-return',
                    enable_modules=[__name__],
                )
                ctx: tractor.Context
                _: BrokerdStartup
                async with portal.open_context(
                    wrong_final_result,
                ) as (ctx, _):
                    result: BrokerdPayload = (
                        await ctx.wait_for_result()
                    )
                    with pytest.raises(ProtocolError):
                        validate(result, type(None))
                await nursery.cancel()
    trio.run(main)
@tractor.context(pld_spec=EmsPayload)
 async def fake_ems(ctx: tractor.Context) -> None:
    '''
    Apply client cancellation before publishing the provider ack.
    The receive order fixes the interleaving: ``OrderState.cancel()``
    runs before ``OrderState.accept()`` handles the ack. Assert one
    deferred cancellation and publish its terminal response.
    '''
    await ctx.started(
        EmsStartup(
            broker='fixture',
            positions=(),
            accounts=('paper',),
            dialogs=(),
        )
    )
    state: OrderState = OrderState()
    stream: Stream[EmsRequest, EmsEvent]
    async with open_stream(ctx, EmsRequest, EmsEvent) as stream:
        order: Order = await stream.receive()
        state.register(order)
        cancel: Cancel = await stream.receive()
        assert cancel.oid == order.oid
        assert not state.cancel(cancel.oid).broker
        effects: Effects = state.accept(
            BrokerdOrderAck(
                oid=order.oid,
                account=order.account,
                reqid=1,
            )
        )
        assert len(effects.broker) == 1
        assert effects.broker[0].oid == order.oid
        effects = state.accept(
            BrokerdStatus(
                time_ns=1,
                account=order.account,
                broker='fixture',
                reqid=1,
                status='canceled',
            )
        )
        await stream.send(effects.clients[0])
        await trio.sleep_forever()
 def test_ems_cancel_before_ack_over_ipc() -> None:
    '''
    Client Order and Cancel must arrive as typed requests and produce
    an immutable nested Status. Hold the fake provider ack until the
    cancel reaches the remote EMS reducer, then require one deferred
    broker cancel and a terminal client response through real IPC.
    '''
    async def main() -> None:
        '''
        Send ordered client intent and require a typed canceled
        snapshot.
        '''
        with trio.fail_after(15):
            nursery: tractor.ActorNursery
            async with (
                collapse_eg(),
                tractor.open_root_actor(
                    tpt_bind_addrs=[('127.0.0.1', 0)],
                ),
                tractor.open_nursery() as nursery,
            ):
                portal: tractor.Portal = await nursery.start_actor(
                    'fake-ems',
                    enable_modules=[__name__],
                )
                ctx: tractor.Context
                startup: EmsStartup
                async with portal.open_context(
                    fake_ems,
                ) as (ctx, startup):
                    startup = validate_ems_startup(
                        startup
                    )
                    assert startup.dialogs == ()
                    stream: Stream[EmsEvent, EmsRequest]
                    async with open_stream(
                        ctx,
                        EmsEvent,
                        EmsRequest,
                    ) as stream:
                        order: Order = Order(
                            oid='client-1',
                            account='paper',
                            symbol='btc.spot.fixture',
                            price=Decimal('12.25'),
                            size=1.0,
                            action='buy',
                            exec_mode='live',
                        )
                        await stream.send(order)
                        await stream.send(
                            Cancel(
                                oid=order.oid,
                                symbol=order.symbol,
                            )
                        )
                        status: EmsEvent = await stream.receive()
                        assert isinstance(status, Status)
                        assert status.req == order
                        assert status.resp == 'canceled'
                        assert status.cancel_called
                        assert isinstance(
                            status.brokerd_event,
                            BrokerdStatus,
                        )
                        await ctx.cancel()
                await nursery.cancel()
    trio.run(main)
--- a/tests/test_ems_messages.py
+++ b/tests/test_ems_messages.py
@ -0,0 +1,333 @@
 # piker: trading gear for hackers
 # Copyright (C) Tyler Goodlet (in stewardship for pikers)
 # This program is free software: you can redistribute it and/or
 # modify it under the terms of the GNU Affero General Public
 # License as published by the Free Software Foundation, either
 # version 3 of the License, or (at your option) any later version.
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 # GNU Affero General Public License for more details.
 # You should have received a copy of the GNU Affero General Public
 # License along with this program. If not, see
 # <https://www.gnu.org/licenses/>.
 '''
 Offline clearing schema and local API boundary conformance.
 '''
 from decimal import Decimal
 from types import UnionType
 import msgspec
 from msgspec import msgpack
 import pytest
 from piker.clearing._codec import (
    Message,
    ProtocolError,
    decoder,
    validate,
 )
 from piker.clearing._schema import (
    BrokerdCancel,
    BrokerdError,
    BrokerdEvent,
    BrokerdFill,
    BrokerdOpenOrder,
    BrokerdOpenOrdersComplete,
    BrokerdOrder,
    BrokerdOrderAck,
    BrokerdPayload,
    BrokerdPosition,
    BrokerdRequest,
    BrokerdStartup,
    BrokerdStatus,
    Cancel,
    EmsEvent,
    EmsPayload,
    EmsRequest,
    EmsStartup,
    Order,
    Status,
    validate_startup,
 )
 ORDER: Order = Order(
    oid='client-1',
    symbol='btc.usdt.binance',
    account='paper',
    action='buy',
    exec_mode='live',
    price=Decimal('12.25'),
    size=1.0,
    brokers=('binance',),
 )
 REQUEST: BrokerdOrder = BrokerdOrder(
    oid=ORDER.oid,
    account=ORDER.account,
    symbol='btc.usdt',
    time_ns=1,
    price=ORDER.price,
    size=ORDER.size,
    action='buy',
 )
 POSITION: BrokerdPosition = BrokerdPosition(
    broker='binance',
    account='paper',
    symbol=ORDER.symbol,
    size=1.0,
    avg_price=12.25,
 )
 EVENTS: tuple[BrokerdEvent, ...] = (
    BrokerdOrderAck(oid=ORDER.oid, reqid=7, account='paper'),
    BrokerdStatus(
        time_ns=1,
        reqid=7,
        status='open',
        account='paper',
        broker='binance',
    ),
    BrokerdFill(
        time_ns=1,
        broker_time=1.0,
        reqid=7,
        fill_id='exec-1',
        account='paper',
        broker='binance',
        action='buy',
        size=1.0,
        price=12.25,
    ),
    BrokerdError(
        account='paper',
        broker='binance',
        reason='rejected',
        request=REQUEST,
    ),
    POSITION,
    BrokerdOpenOrder(
        reqid=7,
        order=ORDER,
        broker='binance',
        time_ns=1,
    ),
    BrokerdOpenOrdersComplete(broker='binance'),
 )
 STATUS: Status = Status(
    time_ns=1,
    oid=ORDER.oid,
    resp='fill',
    req=ORDER,
    brokerd_event=EVENTS[2],
 )
 STARTUP: BrokerdStartup = BrokerdStartup(
    broker='binance',
    mode='paper',
    accounts=('paper',),
    positions=(POSITION,),
 )
 event: BrokerdEvent
@pytest.mark.parametrize(
    'msg,spec',
    [
        (ORDER, EmsRequest),
        (Cancel(oid=ORDER.oid, symbol=ORDER.symbol), EmsRequest),
        (REQUEST, BrokerdRequest),
        (
            BrokerdCancel(
                oid=ORDER.oid,
                symbol=ORDER.symbol,
                account='paper',
                time_ns=1,
                reqid=7,
            ),
            BrokerdRequest,
        ),
        *((event, BrokerdEvent) for event in EVENTS),
        (STATUS, EmsEvent),
        (STARTUP, BrokerdPayload),
        (
            EmsStartup(
                broker='binance',
                positions=(POSITION,),
                accounts=('paper',),
                dialogs=(STATUS,),
            ),
            EmsPayload,
        ),
    ],
 )
 def test_protocol_roundtrip(
    msg: Message|None,
    spec: type|UnionType,
 ) -> None:
    '''
    Every clearing variant must round-trip through its real union,
    including nested orders, fills, Decimal prices and startup
    tuples. Decoding with the declared peer spec proves
    discriminators are unique and fields are supported without hooks
    or Any fallbacks.
    '''
    decoded: Message = decoder(spec).decode(msgpack.encode(msg))
    assert type(decoded) is type(msg)
    assert decoded == msg
@pytest.mark.parametrize(
    'msg,spec',
    [
        (ORDER, BrokerdRequest),
        (REQUEST, BrokerdEvent),
        (EVENTS[0], BrokerdRequest),
        (STARTUP, BrokerdEvent),
        (STATUS, EmsRequest),
        (None, EmsRequest),
    ],
 )
 def test_wrong_direction_and_phase(
    msg: Message|None,
    spec: type|UnionType,
 ) -> None:
    '''
    A valid lifecycle message is invalid in another direction or
    streaming phase. Test both the local send guard and wire decoder:
    sharing a transport does not widen an endpoint's receive set.
    '''
    with pytest.raises(ProtocolError):
        validate(msg, spec)
    with pytest.raises(msgspec.ValidationError):
        decoder(spec).decode(msgpack.encode(msg))
@pytest.mark.parametrize(
    'changes',
    [
        {'extra': 'unknown'},
        {'action': 'cancel'},
        {'size': '1.0'},
        {'account': None},
        {'price': []},
    ],
 )
 def test_invalid_order_fields(changes: dict[str, object]) -> None:
    '''
    Foreign fields and incorrectly typed fields must fail at decode.
    Modify a genuine encoded order so only the tested field changes;
    successful parsing of the remaining fields cannot hide the error.
    '''
    raw: dict[str, object] = (
        msgpack.decode(msgpack.encode(ORDER)) | changes
    )
    with pytest.raises(msgspec.ValidationError):
        decoder(EmsRequest).decode(msgpack.encode(raw))
 def test_construction_is_not_validation() -> None:
    '''
    msgspec constructors accept an invalid status literal. The local
    boundary must reject it before any transport send is attempted. A
    structurally valid tagged dictionary is also rejected locally,
    although its wire representation correctly decodes to a Struct.
    '''
    bad: BrokerdStatus = BrokerdStatus(
        time_ns=1,
        reqid=7,
        status='filled',
        account='paper',
        broker='binance',
    )
    with pytest.raises(msgspec.ValidationError):
        validate(bad, BrokerdEvent)
    raw: dict[str, object] = msgpack.decode(msgpack.encode(ORDER))
    with pytest.raises(ProtocolError):
        validate(raw, EmsRequest)
    assert decoder(EmsRequest).decode(msgpack.encode(raw)) == ORDER
 def test_immutable_snapshot_and_copy() -> None:
    '''
    Editing an order must not mutate a previously published snapshot.
    Exercise the public copy method, frozen fields and nested request
    validation so an invalid replacement cannot enter local history.
    '''
    replacement: Order = ORDER.copy(update={'price': Decimal('13')})
    assert ORDER.price == Decimal('12.25')
    assert replacement.price == Decimal('13')
    with pytest.raises(AttributeError):
        ORDER.price = Decimal('14')
    with pytest.raises(msgspec.ValidationError):
        STATUS.copy(
            update={
                'req': msgspec.structs.replace(ORDER, size='bad'),
            }
        )
@pytest.mark.parametrize(
    'changes',
    [
        {'accounts': ('paper', 'paper')},
        {'accounts': (), 'positions': ()},
        {'mode': 'live', 'accounts': (), 'positions': ()},
        {'accounts': ('other',), 'positions': ()},
        {'accounts': ('other',)},
        {'broker': 'kraken'},
        {'positions': (POSITION, POSITION)},
        {'mode': 'paper_fallback'},
    ],
 )
 def test_startup_ownership(changes: dict[str, object]) -> None:
    '''
    A typed startup can still claim another backend's positions or
    duplicate an account. Alter one invariant at a time and verify
    the semantic validator rejects it before EMS publishes the state.
    '''
    with pytest.raises(ProtocolError):
        validate_startup(STARTUP.copy(update=changes))
    assert validate_startup(STARTUP) == STARTUP
@pytest.mark.parametrize(
    'bad',
    [
        {'req': msgpack.decode(msgpack.encode(ORDER))},
        {
            'req': msgspec.structs.replace(
                ORDER,
                price=Decimal('NaN'),
            ),
        },
        {'req': msgspec.structs.replace(ORDER, price='12.25')},
        {'req': msgspec.structs.replace(ORDER, brokers=['binance'])},
        {'req': msgspec.structs.replace(ORDER, size=float('inf'))},
    ],
 )
 def test_local_nested_objects_are_typed(
    bad: dict[str, object],
 ) -> None:
    '''
    A tagged nested map can encode identically to an Order and a
    string can encode identically to Decimal. The local API must
    still require typed objects, immutable tuples and finite prices.
    Replace nested fields without constructor validation and require
    rejection before the snapshot can enter a queue or order history.
    '''
    with pytest.raises(ProtocolError):
        validate(msgspec.structs.replace(STATUS, **bad), EmsEvent)
--- a/tests/test_ems_state.py
+++ b/tests/test_ems_state.py
@ -0,0 +1,332 @@
 # piker: trading gear for hackers
 # Copyright (C) Tyler Goodlet (in stewardship for pikers)
 # This program is free software: you can redistribute it and/or
 # modify it under the terms of the GNU Affero General Public
 # License as published by the Free Software Foundation, either
 # version 3 of the License, or (at your option) any later version.
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 # GNU Affero General Public License for more details.
 # You should have received a copy of the GNU Affero General Public
 # License along with this program. If not, see
 # <https://www.gnu.org/licenses/>.
 '''
 Deterministic order-event sequencing without feeds or credentials.
 '''
 from decimal import Decimal
 import pytest
 from piker.clearing._codec import ProtocolError
 from piker.clearing._schema import (
    BrokerdCancel,
    BrokerdError,
    BrokerdFill,
    BrokerdOrderAck,
    BrokerdStatus,
    Order,
    Status,
 )
 from piker.clearing._state import Effects, OrderState
 def order(
    oid: str = 'one',
    account: str = 'paper',
 ) -> Order:
    '''
    Build live client intent with controllable dialog/account
    identity.
    '''
    return Order(
        oid=oid,
        account=account,
        symbol='btc.usdt.binance',
        price=Decimal('10'),
        size=1.0,
        action='buy',
        exec_mode='live',
    )
 def ack(
    oid: str = 'one',
    account: str = 'paper',
    reqid: int|str = 1,
 ) -> BrokerdOrderAck:
    '''
    Build the acknowledgement establishing an account/provider ID.
    '''
    return BrokerdOrderAck(oid=oid, account=account, reqid=reqid)
 def fill(
    fill_id: str = 'fill-1',
    account: str = 'paper',
    reqid: int|str = 1,
 ) -> BrokerdFill:
    '''
    Build an execution whose delivery may precede its
    acknowledgement.
    '''
    return BrokerdFill(
        time_ns=1,
        broker_time=1.0,
        account=account,
        broker='binance',
        reqid=reqid,
        fill_id=fill_id,
        size=0.5,
        price=10.0,
        action='buy',
    )
 def closed() -> BrokerdStatus:
    '''
    Build a terminal provider status for the default test order.
    '''
    return BrokerdStatus(
        time_ns=1,
        reqid=1,
        account='paper',
        broker='binance',
        status='closed',
    )
 def test_cancel_before_ack_once() -> None:
    '''
    A cancel can reach EMS while broker submission is suspended.
    Register intent, cancel twice, then repeat the ack. Only the
    first ack may produce a cancel, carrying the original account and
    oid. The earlier published snapshot must remain unchanged.
    '''
    state: OrderState = OrderState()
    initial: Status = state.register(order())
    assert not state.cancel('one').broker
    assert not state.cancel('one').broker
    effects: Effects = state.accept(ack())
    assert len(effects.broker) == 1
    cancel: BrokerdCancel = effects.broker[0]
    assert (cancel.oid, cancel.reqid, cancel.account) == (
        'one',
        1,
        'paper',
    )
    assert not state.accept(ack()).broker
    assert initial.cancel_called is False
@pytest.mark.parametrize('terminal_first', [True, False])
 def test_early_duplicate_and_late_fills(
    terminal_first: bool,
 ) -> None:
    '''
    Native callbacks may precede an ack and a final fill may follow
    closed. Buffer both event orders, correlate with the ack, then
    repeat the fill and inject a distinct late execution. Each fill
    must be emitted once; neither may reopen the completed dialog.
    '''
    state: OrderState = OrderState()
    state.register(order())
    events: list[BrokerdStatus|BrokerdFill] = (
        [closed(), fill()] if terminal_first else [fill(), closed()]
    )
    event: BrokerdStatus|BrokerdFill
    for event in events:
        assert not state.accept(event).clients
    effects: Effects = state.accept(ack())
    responses: list[str] = []
    msg: Status
    for msg in effects.clients:
        responses.append(msg.resp)
    assert responses == (
        ['closed', 'fill'] if terminal_first else ['fill', 'closed']
    )
    assert not state.accept(fill()).clients
    assert state.accept(fill('fill-2')).clients[0].resp == 'fill'
    assert not state.active
    assert state.terminal['one'].status.resp == 'closed'
 def test_closed_before_ack_cancels_no_finished_order() -> None:
    '''
    An order can finish while its ack is delayed and cancel is
    queued. Replay the pre-ack terminal status before generating
    effects so the delayed acknowledgement cannot cancel an already
    closed order.
    '''
    state: OrderState = OrderState()
    state.register(order())
    state.cancel('one')
    state.accept(closed())
    assert not state.accept(ack()).broker
 def test_accounts_partition_request_ids() -> None:
    '''
    Separate broker sessions may both allocate request ID one.
    Register and acknowledge two accounts, then route their fills.
    Correlation must retain account ownership even when IDs collide.
    '''
    state: OrderState = OrderState()
    oid: str
    account: str
    for oid, account in [('one', 'paper'), ('two', 'other')]:
        state.register(order(oid, account))
        state.accept(ack(oid, account))
        effect: Effects = state.accept(fill(account=account))
        assert effect.clients[0].oid == oid
 def test_pre_ack_capacity_is_bounded() -> None:
    '''
    A backend that never acknowledges must not grow EMS memory
    forever. Exceed a deliberately tiny buffer and assert a protocol
    failure preserves the already buffered execution instead of
    dropping it.
    '''
    state: OrderState = OrderState(max_early=1)
    state.accept(fill())
    with pytest.raises(ProtocolError, match='capacity'):
        state.accept(fill('fill-2'))
    assert state.early[('paper', 1)] == [fill()]
 def test_tombstone_eviction_removes_correlation() -> None:
    '''
    Terminal dialogs have finite retention. Complete two orders with
    a one-entry limit; evicting the first must also remove every old
    request ID so a later event cannot target a missing dialog.
    '''
    state: OrderState = OrderState(max_terminal=1)
    oid: str
    reqid: int
    for oid, reqid in [('one', 1), ('two', 2)]:
        state.register(order(oid))
        state.accept(ack(oid, reqid=reqid))
        state.accept(closed().copy(update={'reqid': reqid}))
    assert tuple(state.terminal) == ('two',)
    assert state.reqids == {('paper', 2): 'two'}
 def test_edit_before_ack_cannot_duplicate_submission() -> None:
    '''
    An edit with no known backend request ID could place a second
    order. Hold the ack and try an edit; rejection must leave the
    original request intact for subsequent acknowledgement and
    cancellation.
    '''
    state: OrderState = OrderState()
    initial: Status = state.register(order())
    with pytest.raises(
        ProtocolError,
        match='before acknowledgement',
    ):
        state.register(order().copy(update={'price': Decimal('11')}))
    assert state.active['one'] is initial
 def test_edit_ack_keeps_late_fills_on_original_dialog() -> None:
    '''
    An edit may replace the provider request ID while executions for
    the old ID remain in flight. A delayed duplicate ack must not
    revert the cancel target, and either ID must still route fills to
    the same EMS order. Cancel once per accepted request identity.
    '''
    state: OrderState = OrderState()
    state.register(order())
    state.accept(ack())
    state.register(order().copy(update={'price': Decimal('11')}))
    state.accept(ack(reqid=2))
    state.accept(ack(reqid=1))
    assert state.active['one'].reqid == 2
    assert state.cancel('one').broker[0].reqid == 2
    assert not state.cancel('one').broker
    assert state.accept(fill(reqid=1)).clients[0].oid == 'one'
    effects: Effects = state.accept(fill('fill-2', reqid=2))
    assert effects.clients[0].oid == 'one'
    assert state.active['one'].reqid == 2
    state.accept(closed())
    assert 'one' in state.active
 def test_startup_collision_does_not_register_second_order() -> None:
    '''
    Two existing orders cannot claim one account/request ID pair.
    Correlate a normal order, then inject a startup order with the
    same provider ID. Rejection leaves ``OrderState.active`` and
    ``OrderState.dialogs`` free of the rejected second order.
    '''
    from piker.clearing._schema import BrokerdOpenOrder
    state: OrderState = OrderState()
    state.register(order())
    state.accept(ack())
    with pytest.raises(ProtocolError, match='ownership'):
        state.accept(
            BrokerdOpenOrder(
                reqid=1,
                order=order('two'),
                broker='binance',
                time_ns=1,
            )
        )
    assert tuple(state.dialogs) == ('one',)
    assert tuple(state.active) == ('one',)
 def test_submission_and_edit_rejections() -> None:
    '''
    A new submission error has no future ack to unblock cancellation.
    Its typed request identifies a terminal rejection; an edit error
    must retain the original live control ID so it can still cancel.
    Exercise both with identical reasons to prove request semantics,
    instead of parsing provider error strings, choose the transition.
    '''
    from piker.clearing._state import broker_request
    state: OrderState = OrderState()
    state.register(order())
    error: BrokerdError = BrokerdError(
        account='paper',
        broker='binance',
        reason='rejected',
        request=broker_request(order()),
    )
    assert state.accept(error).clients[0].resp == 'error'
    assert not state.active
    assert 'one' in state.terminal
    state = OrderState()
    state.register(order())
    state.accept(ack())
    error = error.copy(
        update={
            'request': broker_request(order(), 1),
        }
    )
    state.accept(error)
    assert state.cancel('one').broker[0].reqid == 1