From 0055c993f13a564c85d2dde9af474423719fb2d1 Mon Sep 17 00:00:00 2001 From: Jackson Atassi Date: Tue, 9 Jun 2026 22:45:58 -0600 Subject: [PATCH 1/3] fix(execution): equity CLOSE re-checks the live broker position at dispatch (ALP-943) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 2026-06-09 17:00Z run decided to close MRVL long 4 and dispatched that CLOSE 48 minutes later, after the monitor's re-protection stop had already flattened the position. The dispatch resolved qty/side from the frozen positions projection, tolerated the 422 already-filled rejection on the protective-leg cancel by design (ALP-937), and sold 4 shares into a flat position — Alpaca classified the sell sell_to_open and opened an unmanaged -4 MRVL short. Fix — execution-time live-position recheck in the shared equity CLOSE dispatch (covers both the PM-directed and engine-envelope paths): - (A) AccountStateQueries.get_open_position(symbol) -> PositionSnapshot | None, wrapping TradingClient.get_open_position with 404 -> None like get_asset; mirrored on AccountStateQueriesP and the debug-e2e stand-in. - (B) _close_equity resolves the requested quantity (hoisted from submit_equity_close) and consults the live position before any leg-cancel RPC: flat or side-flipped -> PermanentRejectionError carrying PermanentRejection(code="position_state_drift", http_status=0); live qty below requested -> clamp the sell to the live quantity. - (C) _cancel_protective_legs resolves a permanently-rejected leg cancel via get_order_by_id: filled / partially_filled aborts the close with the same drift rejection (the protective exit executed); canceled / expired / 404-unknown proceeds as before. - The PM dispatch branch emits a command_abandoned entry for local guard rejections (http_status == 0) — the command never reached the broker, mirroring the stale-anchor backstop. The engine-envelope path returns a rejected SubmissionResult with the trigger not marked seen, unchanged. - PermanentRejectionCode widened with "position_state_drift"; runbook gains § 8.11 documenting the rejection class (operator posture: no action). Tests: the ALP-937 tolerate-already-terminal test is rewritten as the canceled-proceeds / filled-aborts pair; new drift-guard tests cover flat, side-flip, clamp, and both route-through paths; the broker fake now models live position existence and per-leg terminal states. Closes ALP-943. Co-Authored-By: Claude Fable 5 --- scripts/RUNBOOK_production.md | 27 ++ .../submit_envelope/dispatch.py | 14 +- .../execution/broker_adapter/errors.py | 1 + .../execution/broker_adapter/order_equity.py | 11 +- .../execution/broker_adapter/protocols.py | 14 +- .../execution/broker_adapter/queries.py | 54 +++- .../execution/oms/broker_dispatch.py | 194 ++++++++++--- src/alphamind/scheduler/debug_e2e/broker.py | 14 + .../broker_adapter/test_order_equity.py | 42 +-- .../execution/broker_adapter/test_queries.py | 60 ++++ tests/execution/oms/test_broker_dispatch.py | 273 ++++++++++++++++-- .../oms/test_engine_stub_broker_routing.py | 165 +++++++++++ .../oms/test_submit_engine_envelope.py | 7 +- .../scheduler/test_fill_collection_inputs.py | 9 + tests/scheduler/test_fresh_start.py | 5 + 15 files changed, 769 insertions(+), 121 deletions(-) diff --git a/scripts/RUNBOOK_production.md b/scripts/RUNBOOK_production.md index d93f05c0b..b44d75a4a 100644 --- a/scripts/RUNBOOK_production.md +++ b/scripts/RUNBOOK_production.md @@ -1315,6 +1315,33 @@ Two operational consequences to know: live (§ 8.9), watch the named position's P/L, and — if the next scheduled PM pass is far off — consider a manual invocation (§ 3) so the PM re-brackets or closes it. +### 8.11 Equity CLOSE rejected `position_state_drift` (ALP-943) + +An equity CLOSE (PM-directed or engine-envelope) was rejected at dispatch by the +live-position drift guard: between the invocation snapshot the decision was made +on and the moment of execution, the broker position changed — it exited entirely +(a monitor re-protection stop or any protective leg filled), flipped side, or +shrank below the requested quantity. The guard re-checks the live broker +position (and, on a rejected protective-leg cancel, the leg's actual broker +state) before any close order reaches the wire; a flat or side-flipped position +rejects the command, and a shrunken one submits a sell clamped to the live +quantity instead. The submission log carries the rejection with +`gateway_reason=position_state_drift` plus a one-line message naming the symbol, +the expected side/qty, and what the broker actually reported, and the PM path +records a `command_abandoned` activity entry. + +This is the guard working, not a fault. Without it, the sell executes against a +flat position and Alpaca opens an unmanaged opposite-side position +(`sell_to_open` — the 2026-06-09 −4 MRVL short). + +**Operator action: none.** The position is already flat or smaller than the +projection believed — there is nothing left for the rejected CLOSE to do. The +next invocation's fill collection integrates the missed exit fills and the PM +re-evaluates with corrected state; protection for any surviving remainder is the +monitor's auto re-protection job (ALP-938, § 8.10). Only investigate if the same +symbol rejects across consecutive invocations — that means fill integration is +not catching the projection up to broker reality. + --- ## 9. Feedback loop — analytics CLIs, cadences, and review skills diff --git a/src/alphamind/decision/portfolio_manager/submit_envelope/dispatch.py b/src/alphamind/decision/portfolio_manager/submit_envelope/dispatch.py index 0df470f31..11eecf42c 100644 --- a/src/alphamind/decision/portfolio_manager/submit_envelope/dispatch.py +++ b/src/alphamind/decision/portfolio_manager/submit_envelope/dispatch.py @@ -392,7 +392,19 @@ async def _dispatch_to_broker( await _abandon_if_atomic( ctx, command=command, result=result, reason=f"permanent_rejection: {exc.rejection.code}" ) - return _to_rejection(result, code=exc.rejection.code, reason=str(exc)), None, None + # A local guard rejection (http_status == 0, e.g. the ALP-943 + # ``position_state_drift`` drift guard) means the command never reached + # the broker — emit the ``command_abandoned`` forensic entry, mirroring + # the stale-anchor backstop. A broker-rejected command (real 4xx) has + # the broker's rejection as its record and emits none, as before. + abandoned_entry = ( + _abandoned(result, command, str(exc), 0) if exc.rejection.http_status == 0 else None + ) + return ( + _to_rejection(result, code=exc.rejection.code, reason=str(exc)), + None, + abandoned_entry, + ) except Exception as exc: # Translation seam per runtime §G1: equity/mleg translators re-raise raw # alpaca-py APIError on permanent failure; classify here for a uniform diff --git a/src/alphamind/execution/broker_adapter/errors.py b/src/alphamind/execution/broker_adapter/errors.py index 6883d42ba..78042edbe 100644 --- a/src/alphamind/execution/broker_adapter/errors.py +++ b/src/alphamind/execution/broker_adapter/errors.py @@ -23,6 +23,7 @@ "underlying_halted", # 403 options "invalid_legs", # 422 mleg "asset_not_tradable", # 403 / 422 generic + "position_state_drift", # local dispatch-time guard, http_status=0 (ALP-943) "other_permanent", # fallback for unmapped 4xx ] diff --git a/src/alphamind/execution/broker_adapter/order_equity.py b/src/alphamind/execution/broker_adapter/order_equity.py index 4ba49fa7e..f8639314e 100644 --- a/src/alphamind/execution/broker_adapter/order_equity.py +++ b/src/alphamind/execution/broker_adapter/order_equity.py @@ -220,7 +220,7 @@ async def submit_equity_close( execution: ExecutionConfig, client_order_id: str, symbol: str, - position_qty: float, + qty: float, position_side: Literal["long", "short"], ) -> SubmissionOutcome[EquitySubmission]: """Translate a CLOSE command and submit. @@ -228,13 +228,16 @@ async def submit_equity_close( Always SIMPLE. Uses opposite side of the position: long → SELL, short → BUY (buy-to-cover). - ``symbol``, ``position_qty``, and ``position_side`` are threaded from - portfolio state because ``CloseCommand`` references the position by ID. + ``symbol``, ``qty``, and ``position_side`` are threaded from the dispatch + layer because ``CloseCommand`` references the position by ID. ``qty`` is + the FINAL resolved quantity: ``_close_equity`` resolves ``command.quantity`` + (``"all"`` → projected share count) and runs the ALP-943 live-position + drift guard — including its clamp to the live broker quantity — before + this translator is reached. """ _validate_client_order_id(client_order_id) side = OrderSide.SELL if position_side == "long" else OrderSide.BUY - qty = position_qty if command.quantity == "all" else float(command.quantity) if command.order_type == "limit": if command.limit_price is None: diff --git a/src/alphamind/execution/broker_adapter/protocols.py b/src/alphamind/execution/broker_adapter/protocols.py index c5f446219..c96b06c97 100644 --- a/src/alphamind/execution/broker_adapter/protocols.py +++ b/src/alphamind/execution/broker_adapter/protocols.py @@ -33,20 +33,24 @@ @runtime_checkable class AccountStateQueriesP(Protocol): - """Sync read-only account/positions surface gather_fill_collection_inputs consumes. + """Sync read-only account/positions surface the OMS-facing consumers depend on. Mirrors the as-built methods on :class:`~alphamind.execution.broker_adapter.queries.AccountStateQueries` - — ``get_account`` / ``get_positions`` are synchronous because the - underlying ``alpaca-py`` ``TradingClient`` wraps httpx synchronously; - ``get_orders`` paginates, so it is an async generator (the fresh-start - open-orders precondition drains its ``status="open"`` cursor). + — ``get_account`` / ``get_positions`` / ``get_open_position`` are + synchronous because the underlying ``alpaca-py`` ``TradingClient`` wraps + httpx synchronously; ``get_orders`` paginates, so it is an async generator + (the fresh-start open-orders precondition drains its ``status="open"`` + cursor). ``get_open_position`` is the equity CLOSE dispatch's execution-time + drift guard read (ALP-943). """ def get_account(self) -> TradeAccountSnapshot: ... def get_positions(self) -> tuple[PositionSnapshot, ...]: ... + def get_open_position(self, symbol: str) -> PositionSnapshot | None: ... + def get_orders( self, *, diff --git a/src/alphamind/execution/broker_adapter/queries.py b/src/alphamind/execution/broker_adapter/queries.py index 0307fcbed..f843b6c9e 100644 --- a/src/alphamind/execution/broker_adapter/queries.py +++ b/src/alphamind/execution/broker_adapter/queries.py @@ -351,6 +351,21 @@ def _convert_order(order: Order) -> OrderSnapshot: ) +def _convert_position(pos: Position) -> PositionSnapshot: + return PositionSnapshot( + symbol=str(pos.symbol), + asset_class=_enum_str(pos.asset_class), # type: ignore[arg-type] + qty=float(pos.qty or 0), + avg_entry_price=price(_broker_decimal(pos.avg_entry_price or "1")), + market_value=signed_money(_broker_decimal(pos.market_value or "0")), + cost_basis=signed_money(_broker_decimal(pos.cost_basis or "0")), + unrealized_pl=signed_money(_broker_decimal(pos.unrealized_pl or "0")), + unrealized_plpc=float(pos.unrealized_plpc or 0), + current_price=_optional_price(pos.current_price), + side=_enum_str(pos.side), # type: ignore[arg-type] + ) + + def _broker_decimal(value: str | float | int) -> Decimal: """Parse Alpaca's loosely-typed monetary scalars without binary drift. @@ -471,25 +486,34 @@ def get_positions(self) -> tuple[PositionSnapshot, ...]: raise TypeError(msg) positions: list[Position] = result snapshots = sorted( - ( - PositionSnapshot( - symbol=str(pos.symbol), - asset_class=_enum_str(pos.asset_class), # type: ignore[arg-type] - qty=float(pos.qty or 0), - avg_entry_price=price(_broker_decimal(pos.avg_entry_price or "1")), - market_value=signed_money(_broker_decimal(pos.market_value or "0")), - cost_basis=signed_money(_broker_decimal(pos.cost_basis or "0")), - unrealized_pl=signed_money(_broker_decimal(pos.unrealized_pl or "0")), - unrealized_plpc=float(pos.unrealized_plpc or 0), - current_price=_optional_price(pos.current_price), - side=_enum_str(pos.side), # type: ignore[arg-type] - ) - for pos in positions - ), + (_convert_position(pos) for pos in positions), key=lambda s: s.symbol, ) return tuple(snapshots) + def get_open_position(self, symbol: str) -> PositionSnapshot | None: + """Return the live ``PositionSnapshot`` for *symbol*, or ``None`` when flat. + + Alpaca returns 404 from ``GET /v2/positions/{symbol}`` when the account + holds no position in the symbol; the wrapper converts 404 → ``None`` so + callers treat "no live position" as data (mirroring :meth:`get_asset`). + Non-404 errors propagate unchanged. The equity CLOSE dispatch consumes + this as its execution-time drift guard (ALP-943): the local positions + projection is frozen between fill-collection phases, so a CLOSE resolved + from it must be re-checked against the broker's live position before any + order reaches the wire. + """ + try: + result = self._client.get_open_position(symbol) + except APIError as exc: + if exc.status_code == 404: + return None + raise + if not isinstance(result, Position): + msg = "get_open_position returned unexpected raw-data response" + raise TypeError(msg) + return _convert_position(result) + async def get_orders( self, *, diff --git a/src/alphamind/execution/oms/broker_dispatch.py b/src/alphamind/execution/oms/broker_dispatch.py index a4dcb0c9e..b1d2a984a 100644 --- a/src/alphamind/execution/oms/broker_dispatch.py +++ b/src/alphamind/execution/oms/broker_dispatch.py @@ -80,7 +80,10 @@ submit_options_open, submit_replace, ) -from alphamind.execution.broker_adapter.errors import classify_alpaca_error +from alphamind.execution.broker_adapter.errors import ( + PermanentRejection, + classify_alpaca_error, +) from alphamind.execution.broker_adapter.order_modify import ( AssetClass as ReplaceAssetClass, ) @@ -93,6 +96,8 @@ from alphamind.execution.broker_adapter.order_options import ( PermanentRejectionError, ) +from alphamind.execution.broker_adapter.queries import PositionSnapshot +from alphamind.execution.broker_adapter.retry import bounded_broker_call __all__ = [ "BrokerDispatchResult", @@ -181,13 +186,11 @@ async def dispatch_command_to_broker( # noqa: PLR0913 — caller threads every portfolio-state context for CLOSE / ADD (no embedded instrument on the canonical command) and order-record context for ADJUST / CANCEL. - The ``queries`` parameter is part of the runner-facing signature so the - caller (broker-routing coordinated swap, story 03e) supplies one canonical - ``AccountStateQueries`` bundle per invocation; the dispatcher does not - consult it directly at this story. + The caller (broker-routing coordinated swap, story 03e) supplies one + canonical ``AccountStateQueries`` bundle per invocation; the dispatcher + consults it on the equity CLOSE path, whose ALP-943 drift guard re-checks + the live broker position before any order reaches the wire. """ - del queries # accepted for runner-signature parity; consulted by callers/wiring downstream. - if isinstance(command, OpenCommand): return await _dispatch_open( command, @@ -212,6 +215,7 @@ async def dispatch_command_to_broker( # noqa: PLR0913 — caller threads every return await _dispatch_close( command, client=client, + queries=queries, execution=execution, client_order_id=client_order_id, position_asset_type=position_asset_type, @@ -439,6 +443,7 @@ async def _dispatch_close( # noqa: PLR0913 — close threads every per-asset-ty command: CloseCommand, *, client: TradingClient, + queries: AccountStateQueries, execution: ExecutionConfig, client_order_id: ClientOrderId, position_asset_type: Literal["equity", "option", "strategy"] | None, @@ -457,6 +462,7 @@ async def _dispatch_close( # noqa: PLR0913 — close threads every per-asset-ty return await _close_equity( command, client=client, + queries=queries, execution=execution, client_order_id=client_order_id, position_symbol=position_symbol, @@ -485,10 +491,11 @@ async def _dispatch_close( # noqa: PLR0913 — close threads every per-asset-ty ) -async def _close_equity( +async def _close_equity( # noqa: PLR0913 — close threads every broker-translator parameter plus the drift-guard query surface. command: CloseCommand, *, client: TradingClient, + queries: AccountStateQueries, execution: ExecutionConfig, client_order_id: ClientOrderId, position_symbol: str | None, @@ -502,14 +509,31 @@ async def _close_equity( raise _missing("position_qty", command_kind="CLOSE equity") if position_side is None: raise _missing("position_side", command_kind="CLOSE equity") + symbol = position_symbol + + # ALP-943 — execution-time drift guard. The local positions projection is + # frozen between fill-collection phases, while the monitor flattens + # positions between them by design (re-protection stops, ALP-938), so the + # projected symbol/qty/side may be stale by the time this dispatch runs. + # Re-check the live broker position BEFORE any leg-cancel RPC: a flat or + # side-flipped position rejects (a "close" would OPEN a new position — + # the 2026-06-09 -4 MRVL naked short); a shrunken one clamps the quantity. + requested_qty = position_qty if command.quantity == "all" else float(command.quantity) + live = await bounded_broker_call(lambda: queries.get_open_position(symbol)) + qty = _checked_close_quantity( + live, + position_symbol=symbol, + position_side=position_side, + requested_qty=requested_qty, + ) # ALP-937 — cancel the native bracket's broker-enforced protective legs # BEFORE the close sell. The resting OCO legs reserve 100% of the position's # shares (``held_for_orders``), so without this the SIMPLE sell sees # ``available: 0`` and Alpaca rejects it (``other_permanent`` / 403). Each - # cancel is best-effort: an already-terminal leg (filled / cancelled OCO - # sibling) frees its shares regardless. Monitor-enforced legs (no broker id) - # are never threaded here. + # cancel is best-effort for an already-CANCELED / unknown leg, but a leg the + # broker reports FILLED aborts the close (ALP-943 — the position exited). + # Monitor-enforced legs (no broker id) are never threaded here. protection_torn_down = await _cancel_protective_legs( client=client, execution=execution, @@ -528,39 +552,98 @@ async def _close_equity( client=client, execution=execution, client_order_id=client_order_id, - symbol=position_symbol, - position_qty=position_qty, + symbol=symbol, + qty=qty, position_side=position_side, ) except Exception: if protection_torn_down: - _alert_close_rejected_after_cancel(command, position_symbol=position_symbol) + _alert_close_rejected_after_cancel(command, position_symbol=symbol) raise if protection_torn_down and isinstance(outcome, GatewaySubmissionFailed): - _alert_close_rejected_after_cancel(command, position_symbol=position_symbol) + _alert_close_rejected_after_cancel(command, position_symbol=symbol) return _wrap_equity(outcome) +# Float-dust tolerance for the live-vs-requested close-quantity comparison: a +# live qty within 1e-9 of the requested qty is the same position, not drift. +_CLOSE_QTY_EPSILON = 1e-9 + + +def _position_state_drift(message: str) -> PermanentRejectionError: + """Build the ALP-943 local drift rejection (``http_status=0`` — no HTTP exchange).""" + return PermanentRejectionError( + PermanentRejection( + code="position_state_drift", + http_status=0, + alpaca_message=message, + ) + ) + + +def _checked_close_quantity( + live: PositionSnapshot | None, + *, + position_symbol: str, + position_side: Literal["long", "short"], + requested_qty: float, +) -> float: + """Validate the projected equity CLOSE against the live broker position (ALP-943). + + Returns the final close quantity. A flat broker (no live position) or a live + side contradicting the projection raises the ``position_state_drift`` + rejection — submitting the "close" would open a NEW position in the opposite + direction (with ``short_selling_enabled`` Alpaca executes a sell on a flat + position as ``sell_to_open``). A live absolute quantity below the requested + quantity clamps the close to what actually exists at the broker. + """ + if live is None or live.qty == 0: + raise _position_state_drift( + f"equity CLOSE drift guard: {position_symbol} expected {position_side} " + f"{requested_qty}, live broker position is flat" + ) + live_side: Literal["long", "short"] = "long" if live.qty > 0 else "short" + if live_side != position_side: + raise _position_state_drift( + f"equity CLOSE drift guard: {position_symbol} expected {position_side} " + f"{requested_qty}, live broker position is {live_side} qty {live.qty}" + ) + live_abs = abs(live.qty) + if live_abs < requested_qty - _CLOSE_QTY_EPSILON: + logger.warning( + "broker_dispatch: equity CLOSE of %s clamped from %s to live broker qty %s " + "(position shrank between snapshot and dispatch)", + position_symbol, + requested_qty, + live_abs, + ) + return live_abs + return requested_qty + + async def _cancel_protective_legs( *, client: TradingClient, execution: ExecutionConfig, leg_alpaca_order_ids: Sequence[AlpacaOrderId] | None, ) -> bool: - """Best-effort cancel each broker-enforced protective leg ahead of a CLOSE. + """Cancel each broker-enforced protective leg ahead of a CLOSE. Returns whether at least one cancel was CONFIRMED accepted by the broker — i.e. live protection was actively torn down (the ALP-937 (F) naked-position signal). Two cases that do NOT count, because neither establishes that we removed live protection: - * **Already-terminal leg** — ``submit_cancel`` re-raises the raw alpaca-py - ``APIError`` on a permanent 404 (not-found) / 422 (already-filled), NOT a - ``PermanentRejectionError`` (it wraps the equity ``_is_transient`` classifier - that re-raises permanent errors). The OCO sibling fired or the leg already - filled, so the position likely already exited; its shares are already free - and the close proceeds. A genuinely non-broker exception re-raises so a real - fault is never masked. + * **Permanently-rejected cancel** — ``submit_cancel`` re-raises the raw + alpaca-py ``APIError`` on a permanent 404 (not-found) / 422 + (already-terminal), NOT a ``PermanentRejectionError`` (it wraps the equity + ``_is_transient`` classifier that re-raises permanent errors). The leg is + already terminal — but WHICH terminal state matters (ALP-943): a FILLED + leg is broker-confirmed proof the position exited, so the close aborts via + :func:`_resolve_rejected_leg_state` rather than selling into a flat + position; a canceled / expired / unknown leg is benign and the close + proceeds. A genuinely non-broker exception re-raises so a real fault is + never masked. * **Gateway-failed cancel** — unconfirmed (the leg may still rest and hold shares); the close sell itself surfaces the real problem if so. """ @@ -577,16 +660,10 @@ async def _cancel_protective_legs( rejection = classify_alpaca_error(exc) if rejection is None: raise - # A permanent rejection on a cancel is realistically a 404 not-found / - # 422 already-filled — the leg is already terminal (OCO sibling fired / - # leg filled), so its shares are already free. Any other permanent code - # is likewise un-cancellable; proceed regardless and let the close sell - # surface the held-shares problem if the leg somehow still rests. - logger.info( - "broker_dispatch: protective leg %s permanently rejected at cancel (%s) — " - "assuming already terminal; proceeding with the close", - leg_alpaca_order_id, - rejection.code, + await _resolve_rejected_leg_state( + client=client, + leg_alpaca_order_id=leg_alpaca_order_id, + rejection_code=rejection.code, ) continue if isinstance(cancel_outcome, GatewaySubmissionFailed): @@ -601,6 +678,57 @@ async def _cancel_protective_legs( return protection_torn_down +# Broker order statuses proving a protective leg EXECUTED — the protected +# position (or part of it) exited, so a close built from the projection must +# not proceed (ALP-943). +_LEG_EXECUTED_STATUSES = frozenset({"filled", "partially_filled"}) + + +async def _resolve_rejected_leg_state( + *, + client: TradingClient, + leg_alpaca_order_id: AlpacaOrderId, + rejection_code: str, +) -> None: + """Resolve a permanently-rejected leg cancel into proceed-or-abort (ALP-943). + + The cancel rejection alone is ambiguous: the leg may be canceled / expired + (benign — the OCO sibling fired or the leg lapsed; its shares are free) or + FILLED (the protective exit executed — the position is gone). Reading the + leg's actual broker state via ``get_order_by_id`` disambiguates without + parsing rejection message text. A filled / partially-filled leg raises the + ``position_state_drift`` rejection so the close never sells into the exited + position; every other resolved state — and an unresolvable leg (404 or a + failing lookup) — proceeds as before, with the close sell itself surfacing + any held-shares problem. + """ + try: + order = await bounded_broker_call(lambda: client.get_order_by_id(leg_alpaca_order_id)) + except Exception: + logger.info( + "broker_dispatch: protective leg %s permanently rejected at cancel (%s) and " + "its state could not be resolved — assuming already terminal; proceeding " + "with the close", + leg_alpaca_order_id, + rejection_code, + ) + return + status = getattr(getattr(order, "status", None), "value", getattr(order, "status", None)) + if status in _LEG_EXECUTED_STATUSES: + raise _position_state_drift( + f"equity CLOSE drift guard: protective leg {leg_alpaca_order_id} is " + f"{status} at the broker — the protective exit executed, so the " + "position this CLOSE targets no longer exists as projected" + ) + logger.info( + "broker_dispatch: protective leg %s permanently rejected at cancel (%s); broker " + "reports it %s — benign already-terminal, proceeding with the close", + leg_alpaca_order_id, + rejection_code, + status, + ) + + def _alert_close_rejected_after_cancel(command: CloseCommand, *, position_symbol: str) -> None: """Emit the ALP-937 (F) broker-naked-position operator alert.""" logger.critical( diff --git a/src/alphamind/scheduler/debug_e2e/broker.py b/src/alphamind/scheduler/debug_e2e/broker.py index b0957ce1d..4bc21b4fc 100644 --- a/src/alphamind/scheduler/debug_e2e/broker.py +++ b/src/alphamind/scheduler/debug_e2e/broker.py @@ -168,6 +168,20 @@ def get_positions(self) -> tuple[PositionSnapshot, ...]: log.info("[debug_e2e] LogOnlyAccountStateQueries.get_positions()") return tuple(_position_snapshot(p) for p in self._portfolio.positions) + def get_open_position(self, symbol: str) -> PositionSnapshot | None: + """Resolve *symbol* against the synthetic portfolio (404 → ``None`` analog). + + The equity CLOSE dispatch's drift guard (ALP-943) consults this before + submitting; resolving from the same synthetic portfolio that seeded the + run keeps an offline debug-e2e CLOSE consistent with its own state. + """ + log.info("[debug_e2e] LogOnlyAccountStateQueries.get_open_position(%s)", symbol) + for position in self._portfolio.positions: + snapshot = _position_snapshot(position) + if snapshot.symbol == symbol: + return snapshot + return None + async def get_orders( self, *, diff --git a/tests/execution/broker_adapter/test_order_equity.py b/tests/execution/broker_adapter/test_order_equity.py index 7dcdf24c9..9000c8069 100644 --- a/tests/execution/broker_adapter/test_order_equity.py +++ b/tests/execution/broker_adapter/test_order_equity.py @@ -737,7 +737,9 @@ def fake_submit(request: Any) -> Any: # --------------------------------------------------------------------------- -# RED → GREEN cycle 10: submit_equity_close side derivation + quantity resolution +# RED → GREEN cycle 10: submit_equity_close side derivation +# (quantity resolution hoisted into _close_equity's ALP-943 drift guard — +# covered by tests/execution/oms/test_broker_dispatch.py) # --------------------------------------------------------------------------- @@ -760,13 +762,13 @@ def fake_submit(request: Any) -> Any: execution=_make_execution_config(), client_order_id=_CLIENT_ORDER_ID_INV, symbol="AAPL", - position_qty=100.0, + qty=100.0, position_side="long", ) req = captured[0] assert req.side == OrderSide.SELL - assert req.qty == 100.0 # "all" → position_qty + assert req.qty == 100.0 # the threaded, pre-resolved final qty @pytest.mark.asyncio @@ -788,7 +790,7 @@ def fake_submit(request: Any) -> Any: execution=_make_execution_config(), client_order_id=_CLIENT_ORDER_ID_INV, symbol="MSFT", - position_qty=50.0, + qty=50.0, position_side="short", ) @@ -797,34 +799,6 @@ def fake_submit(request: Any) -> Any: assert req.qty == 50.0 -@pytest.mark.asyncio -async def test_close_numeric_quantity_uses_command_quantity() -> None: - """When quantity is numeric, use it directly, not position_qty.""" - captured: list[Any] = [] - fake_order = _make_fake_order(order_class=OrderClass.SIMPLE) - - def fake_submit(request: Any) -> Any: - captured.append(request) - return fake_order - - client = MagicMock() - client.submit_order = fake_submit - - cmd = _make_close_command(quantity=25.0, close_rationale="conviction_reduced") - await submit_equity_close( - cmd, - client=client, - execution=_make_execution_config(), - client_order_id=_CLIENT_ORDER_ID_INV, - symbol="AAPL", - position_qty=100.0, # should NOT be used - position_side="long", - ) - - req = captured[0] - assert req.qty == 25.0 - - @pytest.mark.asyncio async def test_close_limit_order_type() -> None: """Close with order_type=limit sends LimitOrderRequest.""" @@ -845,7 +819,7 @@ def fake_submit(request: Any) -> Any: execution=_make_execution_config(), client_order_id=_CLIENT_ORDER_ID_INV, symbol="AAPL", - position_qty=100.0, + qty=100.0, position_side="long", ) @@ -1012,7 +986,7 @@ def fake_submit(request: Any) -> Any: execution=_make_execution_config(), client_order_id=_CLIENT_ORDER_ID_INV, symbol="AAPL", - position_qty=100.0, + qty=100.0, position_side="long", ) diff --git a/tests/execution/broker_adapter/test_queries.py b/tests/execution/broker_adapter/test_queries.py index d7d6def0b..35d2f9871 100644 --- a/tests/execution/broker_adapter/test_queries.py +++ b/tests/execution/broker_adapter/test_queries.py @@ -389,6 +389,66 @@ def test_empty_positions_returns_empty_tuple(self) -> None: assert qs.get_positions() == () +# --------------------------------------------------------------------------- +# 3b. get_open_position (ALP-943) +# --------------------------------------------------------------------------- + + +class TestGetOpenPosition: + """Single-symbol live-position lookup — the equity CLOSE drift guard's read.""" + + def test_open_symbol_returns_position_snapshot(self) -> None: + from alphamind.execution.broker_adapter.queries import ( + AccountStateQueries, + PositionSnapshot, + ) + + client = _fake_client() + client.get_open_position.return_value = _make_position("MRVL") + + qs = AccountStateQueries(client) + result = qs.get_open_position("MRVL") + + assert isinstance(result, PositionSnapshot) + assert result.symbol == "MRVL" + assert result.qty == 10.0 + assert result.side == "long" + client.get_open_position.assert_called_once_with("MRVL") + + def test_no_position_404_returns_none(self) -> None: + from alphamind.execution.broker_adapter.queries import AccountStateQueries + + client = _fake_client() + client.get_open_position.side_effect = _make_404_api_error() + + qs = AccountStateQueries(client) + assert qs.get_open_position("FLAT") is None + + def test_non_404_error_propagates(self) -> None: + from alphamind.execution.broker_adapter.queries import AccountStateQueries + + mock_request = httpx.Request("GET", "https://paper-api.alpaca.markets/v2/positions/X") + mock_response = httpx.Response( + 500, + json={"code": 50000000, "message": "internal server error"}, + request=mock_request, + ) + http_error = httpx.HTTPStatusError( + "500 Internal Server Error", request=mock_request, response=mock_response + ) + err_500 = APIError( # type: ignore[no-untyped-call] + error={"code": 50000000, "message": "internal server error"}, + http_error=http_error, + ) + + client = _fake_client() + client.get_open_position.side_effect = err_500 + + qs = AccountStateQueries(client) + with pytest.raises(APIError): + qs.get_open_position("BROKEN") + + # --------------------------------------------------------------------------- # 4. get_asset # --------------------------------------------------------------------------- diff --git a/tests/execution/oms/test_broker_dispatch.py b/tests/execution/oms/test_broker_dispatch.py index e0ac5a695..4b42d14e2 100644 --- a/tests/execution/oms/test_broker_dispatch.py +++ b/tests/execution/oms/test_broker_dispatch.py @@ -48,7 +48,7 @@ PositionId, Symbol, ) -from alphamind._kernel.money import money, price +from alphamind._kernel.money import money, price, signed_money from alphamind.commands.command_models import ( AddCommand, AdjustCommand, @@ -85,6 +85,8 @@ OptionsSubmission, Submitted, ) +from alphamind.execution.broker_adapter.order_options import PermanentRejectionError +from alphamind.execution.broker_adapter.queries import PositionSnapshot from alphamind.execution.oms.broker_dispatch import ( BrokerDispatchResult, dispatch_command_to_broker, @@ -368,6 +370,50 @@ def _order_not_found_api_error() -> APIError: return cast(APIError, cast(Any, APIError)(body, http_error=fake_http_error)) +def _already_in_state_api_error(state: str) -> APIError: + """The 422 Alpaca returns when cancelling an already-terminal order — + ``order is already in "" state`` (the exact production payload from + the 2026-06-09 MRVL incident's rejected re-protection-leg CANCELs).""" + body = json.dumps({"code": 42210000, "message": f'order is already in "{state}" state'}) + fake_http_error = MagicMock() + fake_http_error.response.status_code = 422 + return cast(APIError, cast(Any, APIError)(body, http_error=fake_http_error)) + + +def _live_position(symbol: str, qty: float) -> PositionSnapshot: + """A live broker ``PositionSnapshot`` with signed *qty* (negative = short).""" + return PositionSnapshot( + symbol=symbol, + asset_class="us_equity", + qty=qty, + avg_entry_price=price(100.0), + market_value=signed_money(qty * 100.0), + cost_basis=signed_money(qty * 100.0), + unrealized_pl=signed_money(0.0), + unrealized_plpc=0.0, + current_price=price(100.0), + side="long" if qty > 0 else "short", + ) + + +class _FakeAccountQueries: + """Fake ``AccountStateQueries`` modelling live broker position existence (ALP-943). + + ``positions`` maps symbol → signed live qty (negative = short). A symbol + absent from the map is flat at the broker — ``get_open_position`` returns + ``None``, mirroring the real wrapper's 404 → ``None`` translation. + """ + + def __init__(self, positions: dict[str, float] | None = None) -> None: + self._positions = dict(positions or {}) + + def get_open_position(self, symbol: str) -> PositionSnapshot | None: + qty = self._positions.get(symbol) + if qty is None: + return None + return _live_position(symbol, qty) + + class _HeldForOrdersClient: """Fake Alpaca client modelling ``held_for_orders`` share reservation (ALP-937). @@ -376,9 +422,14 @@ class _HeldForOrdersClient: until those legs are cancelled. ``cancel_order_by_id`` releases a reserving leg; once no leg still reserves shares, the sell is accepted. - * ``terminal_leg_ids`` — legs already terminal at the broker (filled / cancelled - OCO sibling): they reserve no shares and ``cancel_order_by_id`` on them raises - a 404, exercising the benign already-terminal path. + * ``terminal_leg_ids`` — legs unknown to the broker: ``cancel_order_by_id`` + raises a 404 and ``get_order_by_id`` raises the same 404 (the benign + already-terminal path — nothing confirms an exit). + * ``canceled_leg_ids`` / ``filled_leg_ids`` — legs already terminal with a + resolvable state: the cancel raises the production 422 + ``order is already in "" state`` and ``get_order_by_id`` reports + the matching status, exercising the ALP-943 leg-state resolution (a + CANCELED leg is benign; a FILLED leg means the position exited). * ``sell_fails_after_cancel`` — forces the post-cancel sell to still reject (the ALP-937 (F) naked-position case). @@ -391,22 +442,44 @@ def __init__( *, protective_leg_ids: tuple[str, ...] = (), terminal_leg_ids: tuple[str, ...] = (), + canceled_leg_ids: tuple[str, ...] = (), + filled_leg_ids: tuple[str, ...] = (), sell_fails_after_cancel: bool = False, ) -> None: self._reserving: set[str] = {str(i) for i in protective_leg_ids} self._terminal: set[str] = {str(i) for i in terminal_leg_ids} + self._canceled: set[str] = {str(i) for i in canceled_leg_ids} + self._filled: set[str] = {str(i) for i in filled_leg_ids} self._sell_fails_after_cancel = sell_fails_after_cancel self.calls: list[tuple[str, str]] = [] + self.submitted_qtys: list[float] = [] def cancel_order_by_id(self, order_id: str) -> None: self.calls.append(("cancel", str(order_id))) if str(order_id) in self._terminal: raise _order_not_found_api_error() + if str(order_id) in self._canceled: + raise _already_in_state_api_error("canceled") + if str(order_id) in self._filled: + raise _already_in_state_api_error("filled") self._reserving.discard(str(order_id)) + def get_order_by_id(self, order_id: str) -> MagicMock: + self.calls.append(("get_order", str(order_id))) + if str(order_id) in self._canceled: + order = MagicMock() + order.status = OrderStatus.CANCELED + return order + if str(order_id) in self._filled: + order = MagicMock() + order.status = OrderStatus.FILLED + return order + raise _order_not_found_api_error() + def submit_order(self, request: Any) -> MagicMock: side = getattr(request, "side", None) self.calls.append(("submit", str(getattr(request, "symbol", "")))) + self.submitted_qtys.append(float(getattr(request, "qty", 0))) if side == OrderSide.SELL and (self._reserving or self._sell_fails_after_cancel): raise _insufficient_qty_api_error() return _make_fake_alpaca_order(order_class=OrderClass.SIMPLE) @@ -790,12 +863,12 @@ async def test_dispatch_close_equity_routes_to_submit_equity_close() -> None: fake_order = _make_fake_alpaca_order(order_class=OrderClass.SIMPLE) client = MagicMock() client.submit_order = MagicMock(return_value=fake_order) - queries = MagicMock() + queries = _FakeAccountQueries(positions={"AAPL": 10.0}) outcome = await dispatch_command_to_broker( _close_command(), client=client, - queries=queries, + queries=cast(Any, queries), execution=_execution_config(), client_order_id=ClientOrderId(_CLIENT_ORDER_ID), position_symbol="AAPL", @@ -826,12 +899,12 @@ async def test_close_equity_without_cancel_is_rejected_by_held_for_orders_fake() so the cancel-first test below fails for a reason this one establishes is real. """ client = _HeldForOrdersClient(protective_leg_ids=("alp-tp-1", "alp-stop-1")) - queries = MagicMock() + queries = _FakeAccountQueries(positions={"MRVL": 8.0}) with pytest.raises(APIError): await dispatch_command_to_broker( _close_command("POS-MRVL-001"), client=cast(Any, client), - queries=queries, + queries=cast(Any, queries), execution=_execution_config(), client_order_id=ClientOrderId(_CLIENT_ORDER_ID), position_symbol="MRVL", @@ -848,12 +921,12 @@ async def test_close_equity_cancels_protective_legs_before_the_sell() -> None: close sell, so the sell sees freed shares rather than ``available: 0``.""" leg_ids = ("alp-tp-1", "alp-stop-1") client = _HeldForOrdersClient(protective_leg_ids=leg_ids) - queries = MagicMock() + queries = _FakeAccountQueries(positions={"MRVL": 8.0}) outcome = await dispatch_command_to_broker( _close_command("POS-MRVL-001"), client=cast(Any, client), - queries=queries, + queries=cast(Any, queries), execution=_execution_config(), client_order_id=ClientOrderId(_CLIENT_ORDER_ID), position_symbol="MRVL", @@ -871,18 +944,19 @@ async def test_close_equity_cancels_protective_legs_before_the_sell() -> None: @pytest.mark.asyncio -async def test_close_equity_tolerates_already_terminal_protective_leg() -> None: - """ALP-937 — `submit_cancel` re-raises a raw 404 `APIError` (not a - `PermanentRejectionError`) when a protective leg is already terminal (the OCO - sibling fired / the leg filled). `_cancel_protective_legs` must classify it as - a benign already-terminal leg and proceed with the close, not let it abort.""" - client = _HeldForOrdersClient(terminal_leg_ids=("alp-tp-1",)) - queries = MagicMock() +async def test_close_equity_proceeds_when_protective_leg_already_canceled() -> None: + """ALP-937 / ALP-943 — a protective-leg cancel rejected because the leg is + already CANCELED at the broker (the OCO sibling fired) is benign: the leg's + shares are free and the position still exists, so the close proceeds. The + leg's actual state is confirmed via ``get_order_by_id`` — never inferred + from the rejection alone.""" + client = _HeldForOrdersClient(canceled_leg_ids=("alp-tp-1",)) + queries = _FakeAccountQueries(positions={"MRVL": 8.0}) outcome = await dispatch_command_to_broker( _close_command("POS-MRVL-001"), client=cast(Any, client), - queries=queries, + queries=cast(Any, queries), execution=_execution_config(), client_order_id=ClientOrderId(_CLIENT_ORDER_ID), position_symbol="MRVL", @@ -893,7 +967,37 @@ async def test_close_equity_tolerates_already_terminal_protective_leg() -> None: ) assert isinstance(outcome, Submitted) - assert [op for op, _ in client.calls] == ["cancel", "submit"] + assert [op for op, _ in client.calls] == ["cancel", "get_order", "submit"] + + +@pytest.mark.asyncio +async def test_close_equity_aborts_when_protective_leg_already_filled() -> None: + """ALP-943 — a protective-leg cancel rejected because the leg is already + FILLED is broker-confirmed proof the position exited (the stop/target + executed). The close must NOT proceed — selling would open a naked short. + The 2026-06-09 incident's exact signal: the ALP-937 design tolerated this + and sold 4 MRVL into a flat position.""" + client = _HeldForOrdersClient(filled_leg_ids=("alp-stop-1",)) + queries = _FakeAccountQueries(positions={"MRVL": 8.0}) # (B) raced: still live + + with pytest.raises(PermanentRejectionError) as excinfo: + await dispatch_command_to_broker( + _close_command("POS-MRVL-001"), + client=cast(Any, client), + queries=cast(Any, queries), + execution=_execution_config(), + client_order_id=ClientOrderId(_CLIENT_ORDER_ID), + position_symbol="MRVL", + position_qty=8.0, + position_side="long", + position_asset_type="equity", + close_protective_leg_alpaca_order_ids=(AlpacaOrderId("alp-stop-1"),), + ) + + assert excinfo.value.rejection.code == "position_state_drift" + assert excinfo.value.rejection.http_status == 0 + # The leg state was resolved, and no sell ever reached the broker. + assert [op for op, _ in client.calls] == ["cancel", "get_order"] @pytest.mark.asyncio @@ -905,13 +1009,13 @@ async def test_close_equity_no_naked_alert_when_legs_already_terminal( likely already exited) and the sell then fails, no alert is raised — the position was not made naked by this close.""" client = _HeldForOrdersClient(terminal_leg_ids=("alp-tp-1",), sell_fails_after_cancel=True) - queries = MagicMock() + queries = _FakeAccountQueries(positions={"MRVL": 8.0}) with caplog.at_level("CRITICAL"), pytest.raises(APIError): await dispatch_command_to_broker( _close_command("POS-MRVL-001"), client=cast(Any, client), - queries=queries, + queries=cast(Any, queries), execution=_execution_config(), client_order_id=ClientOrderId(_CLIENT_ORDER_ID), position_symbol="MRVL", @@ -933,13 +1037,13 @@ async def test_close_equity_rejected_after_cancel_emits_naked_position_alert( position is emitted, and the rejection still propagates to the normal path.""" leg_ids = ("alp-tp-1", "alp-stop-1") client = _HeldForOrdersClient(protective_leg_ids=leg_ids, sell_fails_after_cancel=True) - queries = MagicMock() + queries = _FakeAccountQueries(positions={"MRVL": 8.0}) with caplog.at_level("CRITICAL"), pytest.raises(APIError): await dispatch_command_to_broker( _close_command("POS-MRVL-001"), client=cast(Any, client), - queries=queries, + queries=cast(Any, queries), execution=_execution_config(), client_order_id=ClientOrderId(_CLIENT_ORDER_ID), position_symbol="MRVL", @@ -955,6 +1059,125 @@ async def test_close_equity_rejected_after_cancel_emits_naked_position_alert( assert "MRVL" in caplog.text +# --------------------------------------------------------------------------- +# 6c. CLOSE equity execution-time drift guard (ALP-943) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_close_equity_flat_at_broker_is_rejected_before_any_rpc() -> None: + """ALP-943 — the broker is flat on the symbol (the position exited between + the invocation snapshot and dispatch, e.g. a monitor stop-out). The CLOSE is + rejected ``position_state_drift`` BEFORE any leg-cancel or order-submit RPC + reaches the broker — selling into a flat position would open a naked short + (the 2026-06-09 -4 MRVL incident).""" + client = _HeldForOrdersClient(protective_leg_ids=("alp-tp-1", "alp-stop-1")) + queries = _FakeAccountQueries(positions={}) # broker flat on MRVL + + with pytest.raises(PermanentRejectionError) as excinfo: + await dispatch_command_to_broker( + _close_command("POS-MRVL-001"), + client=cast(Any, client), + queries=cast(Any, queries), + execution=_execution_config(), + client_order_id=ClientOrderId(_CLIENT_ORDER_ID), + position_symbol="MRVL", + position_qty=4.0, + position_side="long", + position_asset_type="equity", + close_protective_leg_alpaca_order_ids=( + AlpacaOrderId("alp-tp-1"), + AlpacaOrderId("alp-stop-1"), + ), + ) + + assert excinfo.value.rejection.code == "position_state_drift" + assert excinfo.value.rejection.http_status == 0 + assert client.calls == [] + + +@pytest.mark.asyncio +async def test_close_equity_side_flipped_at_broker_is_rejected() -> None: + """ALP-943 — the live broker position's side contradicts the projection + (e.g. the projected long already exited and a short now exists). The CLOSE + is rejected ``position_state_drift``; a SELL against a live short would + grow the wrong-side exposure, not close it.""" + client = _HeldForOrdersClient() + queries = _FakeAccountQueries(positions={"MRVL": -4.0}) # live SHORT 4 + + with pytest.raises(PermanentRejectionError) as excinfo: + await dispatch_command_to_broker( + _close_command("POS-MRVL-001"), + client=cast(Any, client), + queries=cast(Any, queries), + execution=_execution_config(), + client_order_id=ClientOrderId(_CLIENT_ORDER_ID), + position_symbol="MRVL", + position_qty=4.0, + position_side="long", + position_asset_type="equity", + ) + + assert excinfo.value.rejection.code == "position_state_drift" + assert client.calls == [] + + +@pytest.mark.asyncio +async def test_close_equity_clamps_to_live_qty_when_broker_holds_fewer_shares() -> None: + """ALP-943 — the live broker position is smaller than the requested close + quantity (part of the position exited between snapshot and dispatch). The + submitted sell is clamped to exactly the live quantity — never more shares + than exist at the broker.""" + client = _HeldForOrdersClient() + queries = _FakeAccountQueries(positions={"MRVL": 3.0}) # live long 3 of projected 8 + + outcome = await dispatch_command_to_broker( + _close_command("POS-MRVL-001"), + client=cast(Any, client), + queries=cast(Any, queries), + execution=_execution_config(), + client_order_id=ClientOrderId(_CLIENT_ORDER_ID), + position_symbol="MRVL", + position_qty=8.0, + position_side="long", + position_asset_type="equity", + ) + + assert isinstance(outcome, Submitted) + assert client.submitted_qtys == [3.0] + + +@pytest.mark.asyncio +async def test_close_equity_numeric_quantity_within_live_qty_is_unchanged() -> None: + """ALP-943 — a numeric ``command.quantity`` (partial close) within the live + broker quantity dispatches unchanged: the guard resolves the requested + quantity from the command, not the projected share count.""" + client = _HeldForOrdersClient() + queries = _FakeAccountQueries(positions={"AAPL": 10.0}) + + partial_close = CloseCommand( + command_type="close", + position_id=PositionId("POS-AAPL-001"), + quantity=4.0, + order_type="market", + close_rationale_type="conviction_reduced", + ) + outcome = await dispatch_command_to_broker( + partial_close, + client=cast(Any, client), + queries=cast(Any, queries), + execution=_execution_config(), + client_order_id=ClientOrderId(_CLIENT_ORDER_ID), + position_symbol="AAPL", + position_qty=10.0, + position_side="long", + position_asset_type="equity", + ) + + assert isinstance(outcome, Submitted) + assert client.submitted_qtys == [4.0] + + # --------------------------------------------------------------------------- # 7. CLOSE options threads OCC symbol + position_intent from caller # --------------------------------------------------------------------------- @@ -1193,7 +1416,7 @@ async def test_dispatch_engine_guardrail_close_routes_to_submit_equity_close() - fake_order = _make_fake_alpaca_order(order_class=OrderClass.SIMPLE) client = MagicMock() client.submit_order = MagicMock(return_value=fake_order) - queries = MagicMock() + queries = _FakeAccountQueries(positions={"AAPL": 10.0}) engine_close = CloseCommand( command_type="close", @@ -1207,7 +1430,7 @@ async def test_dispatch_engine_guardrail_close_routes_to_submit_equity_close() - outcome = await dispatch_command_to_broker( engine_close, client=client, - queries=queries, + queries=cast(Any, queries), execution=_execution_config(), client_order_id=ClientOrderId( "MON.session-abc.42.0~the-THE-AAPL-0123456789abcdef0123456789abcdef~inv-X" diff --git a/tests/execution/oms/test_engine_stub_broker_routing.py b/tests/execution/oms/test_engine_stub_broker_routing.py index 0c09836d3..094f43111 100644 --- a/tests/execution/oms/test_engine_stub_broker_routing.py +++ b/tests/execution/oms/test_engine_stub_broker_routing.py @@ -117,6 +117,7 @@ from alphamind.state.tables.theses_codec import ( record_to_rows as thesis_record_to_rows, ) +from tests.execution.oms.test_broker_dispatch import _live_position _NOW = datetime(2026, 5, 9, 14, 30, 0, tzinfo=UTC) _TRIGGER_TS = datetime(2026, 5, 9, 14, 30, tzinfo=UTC) @@ -610,6 +611,9 @@ def _submit_order(req: Any) -> MagicMock: # Mock TradingClient isinstance check for AccountStateQueries. client.__class__ = type("MockTradingClient", (MagicMock,), {}) queries = MagicMock(spec=AccountStateQueries) + # ALP-943 — the equity CLOSE drift guard consults the live broker position; + # report it consistent with the seeded projection so the close proceeds. + queries.get_open_position.return_value = _live_position("NVDA", 10.0) execution_config = ExecutionConfig( greeks_refresh=GreeksRefresh(scheduled_interval_minutes=5, move_trigger_pct=0.01), conservative_delta_buffer_pct=0.0, @@ -1089,6 +1093,9 @@ def _submit_order(req: Any) -> MagicMock: client = MagicMock() client.submit_order = MagicMock(side_effect=_submit_order) queries = MagicMock(spec=AccountStateQueries) + # ALP-943 — report the live broker position consistent with the seeded + # projection so the equity CLOSE drift guard lets the close proceed. + queries.get_open_position.return_value = _live_position("NVDA", 10.0) pm_view = _make_pm_view(positions=(_position_view("POS-NVDA-001"),)) @@ -1207,6 +1214,9 @@ def _submit_order(req: Any) -> MagicMock: client = MagicMock() client.submit_order = MagicMock(side_effect=_submit_order) queries = MagicMock(spec=AccountStateQueries) + # ALP-943 — report the live broker position consistent with the seeded + # projection so the equity CLOSE drift guard lets the close proceed. + queries.get_open_position.return_value = _live_position("NVDA", 10.0) pm_view = _make_pm_view(positions=(_position_view("POS-NVDA-001"),)) @@ -1349,6 +1359,161 @@ async def test_pm_envelope_permanent_rejection_carries_code_in_gateway_reason( await async_engine.dispose() +async def test_pm_envelope_close_drift_rejection_abandons_and_tears_down( + tmp_path: Any, +) -> None: + """ALP-943 — an equity CLOSE whose symbol is flat at the broker (the + position exited between snapshot and dispatch) is rejected + ``position_state_drift`` through ``_route_through_broker``: the per-command + result is rejected with the code in ``gateway_reason``, the pre-committed + CLOSE order row is torn down to CANCELLED, a ``command_abandoned`` + activity-log entry records why the command never reached the broker, and + no order-submit RPC fires.""" + from alphamind.decision.portfolio_manager.submit_envelope import ( + _handle_submit_envelope, + build_initial_submit_envelope_state, + ) + from alphamind.execution.broker_adapter import AccountStateQueries + from tests.execution.oms.test_submit_envelope_mcp import ( + _DEFAULT_ACTIVE_SECTORS, + _close_command, + _make_bundle, + _make_pm_view, + _make_strategist_envelope, + _make_validation_state, + _position_assessment_stub, + _position_view, + _retrieval_store, + _sector_resolver, + ) + + async_engine, factory = _build_db_factory(tmp_path) + try: + await _seed_substrate_with_cash(factory) + await _seed_position_cluster(factory, _open_position(), _active_thesis(), _active_bracket()) + + invocation_id = "inv-close-drift-1" + ctx = InvocationContext( + session_factory=factory, + record=_make_invocation_record(invocation_id=invocation_id), + ) + handle = await ctx.__aenter__() + + envelope = _make_strategist_envelope( + verdict="approve", + commands=(_close_command(position_id=PositionId("POS-NVDA-001")),), + ) + validation_state = _make_validation_state() + state = build_initial_submit_envelope_state( + invocation_id=validation_state.invocation_id, + starting_validation_state=validation_state, + ) + bundle = _make_bundle(position_assessments=(_position_assessment_stub("SA-1"),)) + + client = MagicMock() + queries = MagicMock(spec=AccountStateQueries) + queries.get_open_position.return_value = None # broker flat — the position exited + + pm_view = _make_pm_view(positions=(_position_view("POS-NVDA-001"),)) + + _response, state = await _handle_submit_envelope( + envelope.model_dump(mode="json"), + state=state, + retrieval_store=_retrieval_store(), + pre_processor_bundle=bundle, + pm_view=pm_view, + active_sectors=_DEFAULT_ACTIVE_SECTORS, + halt_mode=False, + sector_resolver=_sector_resolver, + state_persistence_config=_make_state_persistence_config(), + invocation_handle=handle, + client=client, + queries=queries, + execution_config=_default_execution_config(), + ) + await ctx.__aexit__(None, None, None) + + # No order ever reached the broker. + assert client.submit_order.call_count == 0 + + # The per-command result is rejected with the drift code. + assert len(state.submission_log) == 1 + result = state.submission_log[0].submission_results[0] + assert result.status == "rejected" + assert result.rejection_payload is not None + assert result.rejection_payload.gateway_reason == "position_state_drift" + + # ALP-836 — pre-committed before dispatch, so the drift rejection tears + # the CLOSE row down to CANCELLED; the command_abandoned entry records + # why nothing reached the broker. + async with factory() as sess: + order_rows = (await sess.execute(select(OrderRow))).scalars().all() + close_orders = [o for o in order_rows if o.order_role == "CLOSE"] + assert len(close_orders) == 1 + assert close_orders[0].status == "CANCELLED" + + log_rows = ( + ( + await sess.execute( + select(ActivityLogRow).where(ActivityLogRow.invocation_id == invocation_id) + ) + ) + .scalars() + .all() + ) + types = {r.event_type for r in log_rows} + assert EventType.COMMAND_ABANDONED.value in types + finally: + await async_engine.dispose() + + +async def test_engine_envelope_close_drift_rejection_leaves_trigger_unseen( + db: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], +) -> None: + """ALP-943 — through ``submit_engine_envelope``, the drift guard's rejection + yields a ``rejected`` SubmissionResult naming ``position_state_drift``, no + order reaches the broker, nothing is persisted, and the trigger is NOT + marked seen — the monitor may legitimately retry.""" + from alphamind.execution.broker_adapter import AccountStateQueries + from alphamind.execution.oms import build_initial_submit_engine_envelope_state + from alphamind.execution.oms.submit_engine_envelope import submit_engine_envelope + + _, factory = db + await _seed_invocation_substrate(factory) + await _seed_cash_ledger(factory) + await _seed_position_cluster(factory, _open_position(), _active_thesis(), _active_bracket()) + + state = build_initial_submit_engine_envelope_state(monitor_session_id=_MONITOR_SESSION) + + client = MagicMock() + queries = MagicMock(spec=AccountStateQueries) + queries.get_open_position.return_value = None # broker flat — the position exited + + ctx, handle = await _open_handle(factory) + result, new_state = await submit_engine_envelope( + _engine_envelope(), + handle=handle, + state=state, + config=_make_state_persistence_config(), + client=client, + queries=queries, + execution_config=_default_execution_config(), + ) + await ctx.__aexit__(None, None, None) + + assert client.submit_order.call_count == 0 + assert result.status == "rejected" + assert result.rejection_payload is not None + assert "position_state_drift" in (result.rejection_payload.suggested_modification or "") + # The trigger is not marked seen — the monitor may retry next trigger. + assert new_state.seen_trigger_ids == frozenset() + + # Nothing was persisted for the rejected close. + async with factory() as sess: + orders = (await sess.execute(select(OrderRow))).scalars().all() + assert [o for o in orders if o.order_role == "CLOSE"] == [] + + # --------------------------------------------------------------------------- # ALP-743 — a guardrail-PASS command that the broker then rejects must release # the cumulative-impact delta it credited in Step 3, so a resize/retry of the diff --git a/tests/execution/oms/test_submit_engine_envelope.py b/tests/execution/oms/test_submit_engine_envelope.py index 86004e4e6..763c8fd30 100644 --- a/tests/execution/oms/test_submit_engine_envelope.py +++ b/tests/execution/oms/test_submit_engine_envelope.py @@ -669,13 +669,12 @@ async def test_engine_equity_close_cancels_broker_enforced_legs_before_sell( only the two broker-enforced legs are sent to the broker. """ from typing import Any, cast - from unittest.mock import MagicMock - from alphamind.execution.broker_adapter import AccountStateQueries from alphamind.execution.oms import build_initial_submit_engine_envelope_state from alphamind.execution.oms.submit_engine_envelope import submit_engine_envelope from tests.execution.oms.test_broker_dispatch import ( _execution_config, + _FakeAccountQueries, _HeldForOrdersClient, ) @@ -693,7 +692,7 @@ async def test_engine_equity_close_cancels_broker_enforced_legs_before_sell( state = build_initial_submit_engine_envelope_state(monitor_session_id=_MONITOR_SESSION) client = _HeldForOrdersClient(protective_leg_ids=("alp-tp-1", "alp-stop-1")) - queries = MagicMock(spec=AccountStateQueries) + queries = _FakeAccountQueries(positions={"NVDA": 10.0}) ctx, handle = await _open_handle(factory) result, _state = await submit_engine_envelope( @@ -702,7 +701,7 @@ async def test_engine_equity_close_cancels_broker_enforced_legs_before_sell( state=state, config=_make_state_persistence_config(), client=cast(Any, client), - queries=queries, + queries=cast(Any, queries), execution_config=_execution_config(), ) await ctx.__aexit__(None, None, None) diff --git a/tests/scheduler/test_fill_collection_inputs.py b/tests/scheduler/test_fill_collection_inputs.py index 4504aa7ad..04d03224f 100644 --- a/tests/scheduler/test_fill_collection_inputs.py +++ b/tests/scheduler/test_fill_collection_inputs.py @@ -124,6 +124,11 @@ def get_account(self) -> TradeAccountSnapshot: def get_positions(self) -> tuple[PositionSnapshot, ...]: return self._positions + def get_open_position(self, symbol: str) -> PositionSnapshot | None: + # gather_fill_collection_inputs never consults the single-symbol read; + # present to satisfy the AccountStateQueriesP protocol surface (ALP-943). + return next((p for p in self._positions if p.symbol == symbol), None) + async def get_orders( self, *, @@ -723,6 +728,10 @@ def get_account(self) -> TradeAccountSnapshot: def get_positions(self) -> tuple[PositionSnapshot, ...]: return () + def get_open_position(self, symbol: str) -> PositionSnapshot | None: + # Present only for the AccountStateQueriesP surface (ALP-943). + return None + async def get_orders( self, *, diff --git a/tests/scheduler/test_fresh_start.py b/tests/scheduler/test_fresh_start.py index e4ece79c1..dc2f9c0b6 100644 --- a/tests/scheduler/test_fresh_start.py +++ b/tests/scheduler/test_fresh_start.py @@ -124,6 +124,11 @@ def get_account(self) -> TradeAccountSnapshot: def get_positions(self) -> tuple[PositionSnapshot, ...]: return self._positions + def get_open_position(self, symbol: str) -> PositionSnapshot | None: + # The fresh-start preconditions never consult the single-symbol read; + # present to satisfy the AccountStateQueriesP protocol surface (ALP-943). + return next((p for p in self._positions if p.symbol == symbol), None) + async def get_orders( self, *, From 1a6a0ab4ac02634c25b413dd88636e3ef684606f Mon Sep 17 00:00:00 2001 From: Jackson Atassi Date: Tue, 9 Jun 2026 23:14:36 -0600 Subject: [PATCH 2/3] fix: address code-review findings on ALP-943 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Emit the ALP-937 (F) NAKED POSITION alert when the close aborts (drift rejection or unclassifiable error) AFTER an earlier protective leg's cancel was confirmed — previously the abort propagated out of _cancel_protective_legs before the alert path, leaving a surviving remainder silently unprotected (+ regression test). - Compare the live side via PositionSnapshot.side (Alpaca's own PositionSide field) instead of the sign of qty, removing the guard's dependence on the broker's qty sign convention for shorts. - Share recovery.py's fill-bearing status set (promoted to public FILL_BEARING_STATUSES) instead of a second hand-maintained copy. - Single-evaluation status extraction in _resolve_rejected_leg_state; drop the redundant symbol alias in _close_equity. - Document the http_status == 0 local-guard sentinel on PermanentRejection. Co-Authored-By: Claude Fable 5 --- .../execution/broker_adapter/errors.py | 8 ++- .../execution/broker_adapter/recovery.py | 12 ++-- .../execution/oms/broker_dispatch.py | 66 ++++++++++++------- tests/execution/oms/test_broker_dispatch.py | 37 +++++++++++ 4 files changed, 92 insertions(+), 31 deletions(-) diff --git a/src/alphamind/execution/broker_adapter/errors.py b/src/alphamind/execution/broker_adapter/errors.py index 78042edbe..613b99923 100644 --- a/src/alphamind/execution/broker_adapter/errors.py +++ b/src/alphamind/execution/broker_adapter/errors.py @@ -30,11 +30,17 @@ @dataclass(frozen=True) class PermanentRejection: - """Alpaca rejected the submission for a non-retriable reason. + """The submission was rejected for a non-retriable reason. Surfaced to the OMS as a synchronous rejection — the caller does NOT re-enqueue the command. The PM may revise and resubmit on its next invocation. + + ``http_status`` carries Alpaca's 4xx status for broker rejections. + ``http_status == 0`` marks a LOCAL guard rejection — no HTTP exchange + occurred and the command never reached the broker (e.g. the ALP-943 + ``position_state_drift`` dispatch-time guard); consumers may key forensic + handling (the ``command_abandoned`` emit) on that sentinel. """ code: PermanentRejectionCode diff --git a/src/alphamind/execution/broker_adapter/recovery.py b/src/alphamind/execution/broker_adapter/recovery.py index 921cdfd5e..e59621b9b 100644 --- a/src/alphamind/execution/broker_adapter/recovery.py +++ b/src/alphamind/execution/broker_adapter/recovery.py @@ -78,12 +78,14 @@ def get_orders( "done_for_day": "done_for_day", } -# Statuses whose ``filled_avg_price`` + ``filled_qty`` populate the -# corresponding ``FillReport`` fields. Every other status leaves +# Statuses carrying executed shares: ``filled_avg_price`` + ``filled_qty`` +# populate the corresponding ``FillReport`` fields. Every other status leaves # ``fill_price`` / ``fill_quantity`` as ``None`` per the design (the order # may carry a non-zero ``filled_qty`` from earlier partials, but a recovery -# event for a *new* terminal status carries no incremental fill). -_FILL_BEARING_STATUSES: Final[frozenset[str]] = frozenset({"filled", "partially_filled"}) +# event for a *new* terminal status carries no incremental fill). Public: +# the equity CLOSE drift guard (ALP-943, ``oms.broker_dispatch``) shares this +# set to classify a protective leg's broker state as executed. +FILL_BEARING_STATUSES: Final[frozenset[str]] = frozenset({"filled", "partially_filled"}) # OMS PositionIntentLiteral alphabet — used to coerce OrderLegSnapshot's # untyped ``str`` field at the boundary. @@ -256,7 +258,7 @@ def _fill_metrics(status: str, source: _FillBearing) -> tuple[Price | None, floa non-zero ``filled_qty`` from prior partials — the recovery report describes the *current* terminal event, not historical fill increments. """ - if status not in _FILL_BEARING_STATUSES: + if status not in FILL_BEARING_STATUSES: return None, None return source.filled_avg_price, source.filled_qty diff --git a/src/alphamind/execution/oms/broker_dispatch.py b/src/alphamind/execution/oms/broker_dispatch.py index b1d2a984a..5e77c9158 100644 --- a/src/alphamind/execution/oms/broker_dispatch.py +++ b/src/alphamind/execution/oms/broker_dispatch.py @@ -97,6 +97,7 @@ PermanentRejectionError, ) from alphamind.execution.broker_adapter.queries import PositionSnapshot +from alphamind.execution.broker_adapter.recovery import FILL_BEARING_STATUSES from alphamind.execution.broker_adapter.retry import bounded_broker_call __all__ = [ @@ -509,7 +510,6 @@ async def _close_equity( # noqa: PLR0913 — close threads every broker-transla raise _missing("position_qty", command_kind="CLOSE equity") if position_side is None: raise _missing("position_side", command_kind="CLOSE equity") - symbol = position_symbol # ALP-943 — execution-time drift guard. The local positions projection is # frozen between fill-collection phases, while the monitor flattens @@ -519,10 +519,10 @@ async def _close_equity( # noqa: PLR0913 — close threads every broker-transla # side-flipped position rejects (a "close" would OPEN a new position — # the 2026-06-09 -4 MRVL naked short); a shrunken one clamps the quantity. requested_qty = position_qty if command.quantity == "all" else float(command.quantity) - live = await bounded_broker_call(lambda: queries.get_open_position(symbol)) + live = await bounded_broker_call(lambda: queries.get_open_position(position_symbol)) qty = _checked_close_quantity( live, - position_symbol=symbol, + position_symbol=position_symbol, position_side=position_side, requested_qty=requested_qty, ) @@ -535,9 +535,11 @@ async def _close_equity( # noqa: PLR0913 — close threads every broker-transla # broker reports FILLED aborts the close (ALP-943 — the position exited). # Monitor-enforced legs (no broker id) are never threaded here. protection_torn_down = await _cancel_protective_legs( + command, client=client, execution=execution, leg_alpaca_order_ids=close_protective_leg_alpaca_order_ids, + position_symbol=position_symbol, ) # Submit the close sell. If it fails AFTER live protection was torn down the @@ -552,16 +554,16 @@ async def _close_equity( # noqa: PLR0913 — close threads every broker-transla client=client, execution=execution, client_order_id=client_order_id, - symbol=symbol, + symbol=position_symbol, qty=qty, position_side=position_side, ) except Exception: if protection_torn_down: - _alert_close_rejected_after_cancel(command, position_symbol=symbol) + _alert_close_rejected_after_cancel(command, position_symbol=position_symbol) raise if protection_torn_down and isinstance(outcome, GatewaySubmissionFailed): - _alert_close_rejected_after_cancel(command, position_symbol=symbol) + _alert_close_rejected_after_cancel(command, position_symbol=position_symbol) return _wrap_equity(outcome) @@ -596,17 +598,20 @@ def _checked_close_quantity( direction (with ``short_selling_enabled`` Alpaca executes a sell on a flat position as ``sell_to_open``). A live absolute quantity below the requested quantity clamps the close to what actually exists at the broker. + + The side comparison reads ``live.side`` — Alpaca's own ``PositionSide`` + field — rather than inferring from the sign of ``qty``, so the guard does + not depend on the broker's qty sign convention for shorts. """ if live is None or live.qty == 0: raise _position_state_drift( f"equity CLOSE drift guard: {position_symbol} expected {position_side} " f"{requested_qty}, live broker position is flat" ) - live_side: Literal["long", "short"] = "long" if live.qty > 0 else "short" - if live_side != position_side: + if live.side != position_side: raise _position_state_drift( f"equity CLOSE drift guard: {position_symbol} expected {position_side} " - f"{requested_qty}, live broker position is {live_side} qty {live.qty}" + f"{requested_qty}, live broker position is {live.side} qty {live.qty}" ) live_abs = abs(live.qty) if live_abs < requested_qty - _CLOSE_QTY_EPSILON: @@ -622,10 +627,12 @@ def _checked_close_quantity( async def _cancel_protective_legs( + command: CloseCommand, *, client: TradingClient, execution: ExecutionConfig, leg_alpaca_order_ids: Sequence[AlpacaOrderId] | None, + position_symbol: str, ) -> bool: """Cancel each broker-enforced protective leg ahead of a CLOSE. @@ -646,6 +653,11 @@ async def _cancel_protective_legs( never masked. * **Gateway-failed cancel** — unconfirmed (the leg may still rest and hold shares); the close sell itself surfaces the real problem if so. + + If the loop aborts (drift rejection or an unclassifiable error) AFTER an + earlier leg's cancel was confirmed, live protection was already removed and + the close will never run — the ALP-937 (F) operator alert is emitted before + the abort propagates, so a surviving remainder is never silently naked. """ leg_ids = tuple(leg_alpaca_order_ids or ()) protection_torn_down = False @@ -659,12 +671,19 @@ async def _cancel_protective_legs( except Exception as exc: rejection = classify_alpaca_error(exc) if rejection is None: + if protection_torn_down: + _alert_close_rejected_after_cancel(command, position_symbol=position_symbol) + raise + try: + await _resolve_rejected_leg_state( + client=client, + leg_alpaca_order_id=leg_alpaca_order_id, + rejection_code=rejection.code, + ) + except PermanentRejectionError: + if protection_torn_down: + _alert_close_rejected_after_cancel(command, position_symbol=position_symbol) raise - await _resolve_rejected_leg_state( - client=client, - leg_alpaca_order_id=leg_alpaca_order_id, - rejection_code=rejection.code, - ) continue if isinstance(cancel_outcome, GatewaySubmissionFailed): logger.warning( @@ -678,12 +697,6 @@ async def _cancel_protective_legs( return protection_torn_down -# Broker order statuses proving a protective leg EXECUTED — the protected -# position (or part of it) exited, so a close built from the projection must -# not proceed (ALP-943). -_LEG_EXECUTED_STATUSES = frozenset({"filled", "partially_filled"}) - - async def _resolve_rejected_leg_state( *, client: TradingClient, @@ -698,9 +711,11 @@ async def _resolve_rejected_leg_state( leg's actual broker state via ``get_order_by_id`` disambiguates without parsing rejection message text. A filled / partially-filled leg raises the ``position_state_drift`` rejection so the close never sells into the exited - position; every other resolved state — and an unresolvable leg (404 or a - failing lookup) — proceeds as before, with the close sell itself surfacing - any held-shares problem. + position — partial fills abort rather than clamp because the remainder is + mid-execution and ambiguous at this instant; the PM re-evaluates against the + next snapshot and the monitor backstops the interim. Every other resolved + state — and an unresolvable leg (404 or a failing lookup) — proceeds as + before, with the close sell itself surfacing any held-shares problem. """ try: order = await bounded_broker_call(lambda: client.get_order_by_id(leg_alpaca_order_id)) @@ -713,8 +728,9 @@ async def _resolve_rejected_leg_state( rejection_code, ) return - status = getattr(getattr(order, "status", None), "value", getattr(order, "status", None)) - if status in _LEG_EXECUTED_STATUSES: + raw_status = getattr(order, "status", None) + status = getattr(raw_status, "value", raw_status) + if status in FILL_BEARING_STATUSES: raise _position_state_drift( f"equity CLOSE drift guard: protective leg {leg_alpaca_order_id} is " f"{status} at the broker — the protective exit executed, so the " diff --git a/tests/execution/oms/test_broker_dispatch.py b/tests/execution/oms/test_broker_dispatch.py index 4b42d14e2..cd726ec6a 100644 --- a/tests/execution/oms/test_broker_dispatch.py +++ b/tests/execution/oms/test_broker_dispatch.py @@ -1000,6 +1000,43 @@ async def test_close_equity_aborts_when_protective_leg_already_filled() -> None: assert [op for op, _ in client.calls] == ["cancel", "get_order"] +@pytest.mark.asyncio +async def test_close_equity_abort_after_live_cancel_emits_naked_position_alert( + caplog: pytest.LogCaptureFixture, +) -> None: + """ALP-943 — the drift abort fires AFTER an earlier leg's cancel was + confirmed: live protection was actively torn down and the close will never + run, so the ALP-937 (F) CRITICAL alert must be emitted before the abort + propagates — a surviving remainder is never silently naked.""" + client = _HeldForOrdersClient( + protective_leg_ids=("alp-tp-1",), # live — cancel succeeds, protection torn + filled_leg_ids=("alp-stop-1",), # filled — resolution aborts the close + ) + queries = _FakeAccountQueries(positions={"MRVL": 8.0}) + + with caplog.at_level("CRITICAL"), pytest.raises(PermanentRejectionError): + await dispatch_command_to_broker( + _close_command("POS-MRVL-001"), + client=cast(Any, client), + queries=cast(Any, queries), + execution=_execution_config(), + client_order_id=ClientOrderId(_CLIENT_ORDER_ID), + position_symbol="MRVL", + position_qty=8.0, + position_side="long", + position_asset_type="equity", + close_protective_leg_alpaca_order_ids=( + AlpacaOrderId("alp-tp-1"), + AlpacaOrderId("alp-stop-1"), + ), + ) + + assert "NAKED POSITION" in caplog.text + assert "MRVL" in caplog.text + # No sell ever reached the broker. + assert [op for op, _ in client.calls] == ["cancel", "cancel", "get_order"] + + @pytest.mark.asyncio async def test_close_equity_no_naked_alert_when_legs_already_terminal( caplog: pytest.LogCaptureFixture, From e09050d65e6a440c4e95fd914b075ed7222e6c5f Mon Sep 17 00:00:00 2001 From: Jackson Atassi Date: Tue, 9 Jun 2026 23:21:14 -0600 Subject: [PATCH 3/3] fix: address /review findings on ALP-943 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wrap the drift-guard live-position read in submit_with_retry, matching the transient-retry discipline of the path's submit/cancel calls. On window exhaustion the close is not submitted blind — the dispatch returns GatewaySubmissionFailed before any leg-cancel or order-submit RPC, surfacing through the callers' existing gateway-failure handling (+ regression test). Rejected nits with reasons in the PR conversation: the raw-data guard asymmetry in _resolve_rejected_leg_state already converges to the documented fail-open path (production client is typed); the silent clamp is in-spec per Scope (B) and logs a warning. Co-Authored-By: Claude Fable 5 --- .../execution/oms/broker_dispatch.py | 15 +++++++-- tests/execution/oms/test_broker_dispatch.py | 31 +++++++++++++++++++ 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/src/alphamind/execution/oms/broker_dispatch.py b/src/alphamind/execution/oms/broker_dispatch.py index 5e77c9158..f3543d5e1 100644 --- a/src/alphamind/execution/oms/broker_dispatch.py +++ b/src/alphamind/execution/oms/broker_dispatch.py @@ -98,7 +98,7 @@ ) from alphamind.execution.broker_adapter.queries import PositionSnapshot from alphamind.execution.broker_adapter.recovery import FILL_BEARING_STATUSES -from alphamind.execution.broker_adapter.retry import bounded_broker_call +from alphamind.execution.broker_adapter.retry import bounded_broker_call, submit_with_retry __all__ = [ "BrokerDispatchResult", @@ -518,10 +518,19 @@ async def _close_equity( # noqa: PLR0913 — close threads every broker-transla # Re-check the live broker position BEFORE any leg-cancel RPC: a flat or # side-flipped position rejects (a "close" would OPEN a new position — # the 2026-06-09 -4 MRVL naked short); a shrunken one clamps the quantity. + # The read shares the submit calls' transient-retry discipline; if the + # retry window exhausts, the close is NOT submitted blind — the gateway + # failure surfaces through the callers' existing rejection handling and + # the PM / monitor retries with fresh state. requested_qty = position_qty if command.quantity == "all" else float(command.quantity) - live = await bounded_broker_call(lambda: queries.get_open_position(position_symbol)) + live_outcome = await submit_with_retry( + lambda: bounded_broker_call(lambda: queries.get_open_position(position_symbol)), + window_seconds=execution.submission_retry_window_seconds, + ) + if isinstance(live_outcome, GatewaySubmissionFailed): + return live_outcome qty = _checked_close_quantity( - live, + live_outcome.payload, position_symbol=position_symbol, position_side=position_side, requested_qty=requested_qty, diff --git a/tests/execution/oms/test_broker_dispatch.py b/tests/execution/oms/test_broker_dispatch.py index cd726ec6a..3044cd9fe 100644 --- a/tests/execution/oms/test_broker_dispatch.py +++ b/tests/execution/oms/test_broker_dispatch.py @@ -1133,6 +1133,37 @@ async def test_close_equity_flat_at_broker_is_rejected_before_any_rpc() -> None: assert client.calls == [] +@pytest.mark.asyncio +async def test_close_equity_drift_read_failure_is_gateway_failure_not_blind_close() -> None: + """ALP-943 — the drift-guard read shares the submit calls' transient-retry + discipline. When it exhausts the window the close is NOT submitted blind: + the dispatch returns ``GatewaySubmissionFailed`` before any leg-cancel or + order-submit RPC, surfacing through the callers' existing rejection path.""" + import httpx + + class _UnreachableQueries: + def get_open_position(self, symbol: str) -> PositionSnapshot | None: + raise httpx.ConnectError("network down") + + client = _HeldForOrdersClient(protective_leg_ids=("alp-tp-1",)) + + outcome = await dispatch_command_to_broker( + _close_command("POS-MRVL-001"), + client=cast(Any, client), + queries=cast(Any, _UnreachableQueries()), + execution=_execution_config_with_window(1), + client_order_id=ClientOrderId(_CLIENT_ORDER_ID), + position_symbol="MRVL", + position_qty=8.0, + position_side="long", + position_asset_type="equity", + close_protective_leg_alpaca_order_ids=(AlpacaOrderId("alp-tp-1"),), + ) + + assert isinstance(outcome, GatewaySubmissionFailed) + assert client.calls == [] + + @pytest.mark.asyncio async def test_close_equity_side_flipped_at_broker_is_rejected() -> None: """ALP-943 — the live broker position's side contradicts the projection