fix(execution): equity CLOSE re-checks the live broker position at dispatch (ALP-943)

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

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

src/alphamind/execution/broker_adapter/errors.py

@@ -23,17 +23,24 @@
     "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
 ]
 
 
 @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

src/alphamind/execution/broker_adapter/order_equity.py

@@ -220,21 +220,24 @@ 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.
 
     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:

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,
         *,

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,
         *,

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