fix(execution): bound alpaca-py websocket mutators behind async facades (ALP-946)

src/alphamind/execution/broker_adapter/__init__.py

@@ -25,6 +25,7 @@
 (`ALP-121 <https://linear.app/alphamind-jatassi/issue/ALP-121>`_).
 """
 
+from alphamind.execution.broker_adapter.bounded_streams import bounded_call
 from alphamind.execution.broker_adapter.client_factory import (
     AlpacaClientFactory,
     ExecutionMode,
@@ -150,6 +151,7 @@
     "Submitted",
     "TouchQuote",
     "TradeAccountSnapshot",
+    "bounded_call",
     "build_occ_symbol",
     "classify_alpaca_error",
     "derive_capital_floor_client_order_id",

src/alphamind/execution/broker_adapter/bounded_streams.py

@@ -0,0 +1,150 @@
+"""Bounded async facades over alpaca-py's synchronous seams (ALP-946).
+
+alpaca-py is synchronous in two distinct ways and both can freeze the calling
+event loop:
+
+* **REST calls** issue blocking ``requests`` round-trips with no socket
+  timeout — a hung connection blocks the calling thread forever (the ALP-841
+  wedge: a bare-sync ``get_orders`` froze the monitor loop and its watchdog
+  for ~4.5h).
+* **Stream mutators** (``subscribe_quotes`` / ``unsubscribe_quotes`` /
+  ``subscribe_trade_updates``) are *loop-affined*: on a running stream they
+  marshal the actual send onto the stream's own event loop via
+  ``run_coroutine_threadsafe`` and block the calling thread on an untimed
+  ``Future.result()``. Called from that loop's own thread the scheduled
+  coroutine can never run — a guaranteed self-deadlock (the ALP-946 wedge:
+  every position-set-changing invocation froze the monitor for ~150 s).
+
+:func:`bounded_call` is the one safe calling convention for both: offloading
+to a worker thread keeps the loop turning (and gives the loop-affined
+mutators a free loop to marshal onto), and the :func:`asyncio.wait_for`
+bound guarantees the await resolves even if the callable never returns —
+raising :class:`TimeoutError` for the caller's reconnect/retry machinery.
+The client factory's socket-level timeout remains defence-in-depth: it lets
+the orphaned worker thread eventually unwind rather than leak.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import Any, Protocol
+
+# Wall-clock budget (seconds) for one vendor stream-mutator call run on a
+# worker thread. Mirrors ``queries._REST_TIMEOUT_SECONDS`` (30s) and sits
+# below the client factory's 60s socket-timeout floor, so this loop-level
+# bound is the one that normally fires and a wedged send surfaces as a
+# TimeoutError-driven reconnect rather than a frozen await.
+_STREAM_CALL_TIMEOUT_SECONDS = 30.0
+
+
+async def bounded_call[T](fn: Callable[..., T], /, *args: object, timeout_seconds: float) -> T:
+    """Run a synchronous alpaca-py call off the event loop, time-bounded.
+
+    Offloading to a worker thread keeps the calling loop running — for REST
+    calls that means the loop survives a hung socket (ALP-841); for the
+    loop-affined stream mutators it means the send they marshal back onto
+    the loop can actually drain (ALP-946). The :func:`asyncio.wait_for`
+    bound guarantees the await resolves within *timeout_seconds* even if the
+    callable never returns — raising :class:`TimeoutError`, which callers
+    treat as transient (retry on the next sweep interval, or tear down and
+    rebuild the stream). The client factory's socket-level timeout is the
+    matching thread-side floor so the orphaned worker eventually unwinds.
+    """
+    return await asyncio.wait_for(asyncio.to_thread(fn, *args), timeout=timeout_seconds)
+
+
+QuoteHandler = Callable[[Any], Awaitable[None]]
+
+
+class _SyncStockDataStream(Protocol):
+    """The vendor ``StockDataStream`` surface the facade wraps.
+
+    The mutators are the loop-affined synchronous APIs; ``_run_forever`` is
+    alpaca-py's documented async entry point (the underscore prefix is the
+    library's own convention) and ``stop_ws`` its async teardown.
+    """
+
+    def subscribe_quotes(self, handler: QuoteHandler, *symbols: str) -> None: ...
+
+    def unsubscribe_quotes(self, *symbols: str) -> None: ...
+
+    async def _run_forever(self) -> None: ...
+
+    async def stop_ws(self) -> None: ...
+
+
+class BoundedStockDataStream:
+    """Async facade over a vendor ``StockDataStream`` (ALP-946).
+
+    The vendor mutators must run on a thread *other than* the stream's own
+    event-loop thread (see module docstring); the facade routes every call
+    through :func:`bounded_call` so callers can simply ``await`` from the
+    loop. ``_run_forever`` / ``stop_ws`` are already async — plain awaited
+    delegations.
+    """
+
+    def __init__(
+        self,
+        stream: _SyncStockDataStream,
+        *,
+        timeout_seconds: float = _STREAM_CALL_TIMEOUT_SECONDS,
+    ) -> None:
+        self._stream = stream
+        self._timeout_seconds = timeout_seconds
+
+    async def subscribe_quotes(self, handler: QuoteHandler, *symbols: str) -> None:
+        await bounded_call(
+            self._stream.subscribe_quotes, handler, *symbols, timeout_seconds=self._timeout_seconds
+        )
+
+    async def unsubscribe_quotes(self, *symbols: str) -> None:
+        await bounded_call(
+            self._stream.unsubscribe_quotes, *symbols, timeout_seconds=self._timeout_seconds
+        )
+
+    async def _run_forever(self) -> None:
+        await self._stream._run_forever()
+
+    async def stop_ws(self) -> None:
+        await self._stream.stop_ws()
+
+
+class _SyncTradingStream(Protocol):
+    """The vendor ``TradingStream`` surface the facade wraps."""
+
+    def subscribe_trade_updates(self, handler: Any) -> None: ...
+
+    async def _run_forever(self) -> None: ...
+
+    async def stop_ws(self) -> None: ...
+
+
+class BoundedTradingStream:
+    """Async facade over a vendor ``TradingStream`` (ALP-946).
+
+    ``subscribe_trade_updates`` carries the same loop-affined blocking branch
+    as the stock-stream mutators (``alpaca/trading/stream.py``); the facade
+    routes it through :func:`bounded_call` so the fill stream's subscribe can
+    never block the event loop, however the stream's run state interleaves.
+    """
+
+    def __init__(
+        self,
+        stream: _SyncTradingStream,
+        *,
+        timeout_seconds: float = _STREAM_CALL_TIMEOUT_SECONDS,
+    ) -> None:
+        self._stream = stream
+        self._timeout_seconds = timeout_seconds
+
+    async def subscribe_trade_updates(self, handler: Any) -> None:
+        await bounded_call(
+            self._stream.subscribe_trade_updates, handler, timeout_seconds=self._timeout_seconds
+        )
+
+    async def _run_forever(self) -> None:
+        await self._stream._run_forever()
+
+    async def stop_ws(self) -> None:
+        await self._stream.stop_ws()

src/alphamind/execution/broker_adapter/client_factory.py

@@ -22,6 +22,7 @@
 from alpaca.trading.stream import TradingStream
 
 from alphamind.config.models.venue import AlpacaCredentials, VenueConfig
+from alphamind.execution.broker_adapter.bounded_streams import BoundedTradingStream
 
 ExecutionMode = Literal["paper", "live"]
 
@@ -127,13 +128,20 @@ def build_trading_client(self) -> TradingClient:
         _install_socket_timeout(client)
         return client
 
-    def build_trading_stream(self) -> TradingStream:
-        """Return a fresh ``TradingStream`` bound to the resolved credentials."""
-        return TradingStream(
-            api_key=self._credentials.api_key,
-            secret_key=self._credentials.api_secret,
-            paper=(self._credentials.mode == "paper"),
-            url_override=self._credentials.ws_url,
+    def build_trading_stream(self) -> BoundedTradingStream:
+        """Return a fresh ``TradingStream`` wrapped in the async facade.
+
+        The vendor stream's ``subscribe_trade_updates`` is a loop-affined sync
+        API (ALP-946); the facade is the only surface handed out so no caller
+        can issue the bare blocking call on the event loop.
+        """
+        return BoundedTradingStream(
+            TradingStream(
+                api_key=self._credentials.api_key,
+                secret_key=self._credentials.api_secret,
+                paper=(self._credentials.mode == "paper"),
+                url_override=self._credentials.ws_url,
+            )
         )
 
     def build_stock_data_client(self) -> StockHistoricalDataClient:

src/alphamind/execution/broker_adapter/fill_stream.py

@@ -321,13 +321,18 @@ def _coerce_float(value: object) -> float:
 
 
 class _SubscribableStream(Protocol):
-    """Subset of ``alpaca.trading.stream.TradingStream`` we depend on.
+    """Async surface of the bounded trading-stream facade we depend on.
 
     Defined as a protocol so the test suite can swap a fake without inheriting
-    from the real (network-touching) class.
+    from the real (network-touching) class. The runtime object is a
+    :class:`~alphamind.execution.broker_adapter.bounded_streams.BoundedTradingStream`
+    wrapping the vendor stream. ``subscribe_trade_updates`` is ``async`` by
+    contract (ALP-946): alpaca-py's bare sync mutator is loop-affined —
+    calling it on the stream's own loop thread self-deadlocks the process
+    loop — so the seam only admits the bounded off-thread form.
     """
 
-    def subscribe_trade_updates(self, handler: Any) -> None: ...
+    async def subscribe_trade_updates(self, handler: Any) -> None: ...
 
     async def _run_forever(self) -> None: ...
 
@@ -378,7 +383,7 @@ async def subscribe_trade_updates(
     async def _handler(update: TradeUpdate) -> None:
         await queue.put(update)
 
-    stream.subscribe_trade_updates(_handler)
+    await stream.subscribe_trade_updates(_handler)
     # ``asyncio.TaskGroup`` is incompatible with async-generator cleanup:
     # ``gen.aclose()`` injects ``GeneratorExit`` into the body, which a
     # surrounding ``async with TaskGroup()`` re-raises as

src/alphamind/execution/broker_adapter/queries.py

@@ -15,10 +15,9 @@
 
 from __future__ import annotations
 
-import asyncio
 import datetime as dt
 import functools
-from collections.abc import AsyncGenerator, Callable
+from collections.abc import AsyncGenerator
 from decimal import Decimal
 from typing import Any, Literal
 from zoneinfo import ZoneInfo
@@ -46,6 +45,7 @@
 from pydantic import BaseModel, ConfigDict, TypeAdapter
 
 from alphamind._kernel.money import Money, Price, money, price, signed_money
+from alphamind.execution.broker_adapter.bounded_streams import bounded_call
 
 _ET = ZoneInfo("America/New_York")
 
@@ -61,10 +61,10 @@
 # endpoints cannot share ``_PAGE_SIZE`` — requesting 500 here is a hard 422 (ALP-932).
 _ACTIVITIES_PAGE_SIZE = 100
 
-# Per-page wall-clock budget (seconds) for a synchronous Alpaca REST call run on
-# a worker thread. alpaca-py's ``TradingClient`` issues blocking ``requests``
-# calls with no socket timeout, so a hung connection would otherwise block its
-# worker thread forever; bounding the ``to_thread`` await with ``wait_for``
+# Per-page wall-clock budget (seconds) for a synchronous Alpaca REST call run
+# through ``bounded_call``. alpaca-py's ``TradingClient`` issues blocking
+# ``requests`` calls with no socket timeout, so a hung connection would
+# otherwise block its worker thread forever; the ``bounded_call`` bound
 # guarantees the *calling event loop* never freezes regardless of how long the
 # socket hangs (the ALP-841 lesson — a bare-sync ``get_orders`` froze the
 # monitor loop and its watchdog for ~4.5h). The client factory also installs a
@@ -75,24 +75,6 @@
 _REST_TIMEOUT_SECONDS = 30.0
 
 
-async def _rest_call[T](fn: Callable[[], T]) -> T:
-    """Run a synchronous Alpaca REST call off the event loop, time-bounded.
-
-    The paginating query generators (:meth:`AccountStateQueries.get_orders`,
-    :meth:`~AccountStateQueries.get_account_activities`) are consumed from the
-    continuous monitor's recovery sweep and the account-activities poll — both
-    on an event loop shared with the fill stream and (pre-isolation) the safety
-    watchdog. alpaca-py is synchronous, so calling it inline would block that
-    loop for the full network round-trip; a *hung* call would freeze it
-    indefinitely (ALP-841). Offloading to a worker thread keeps the loop
-    running, and the :func:`asyncio.wait_for` bound guarantees the await resolves
-    within :data:`_REST_TIMEOUT_SECONDS` even if the socket never returns —
-    raising :class:`TimeoutError`, which the adapter's error classifier treats as
-    transient so the sweep retries on its next interval.
-    """
-    return await asyncio.wait_for(asyncio.to_thread(fn), timeout=_REST_TIMEOUT_SECONDS)
-
-
 # ---------------------------------------------------------------------------
 # Typed response records — all frozen
 # ---------------------------------------------------------------------------
@@ -545,7 +527,10 @@ async def get_orders(
             # a hung connection must not freeze the loop (ALP-841). ``partial``
             # binds this page's request by value (the call is awaited before the
             # next iteration rebinds ``req``).
-            page_result = await _rest_call(functools.partial(self._client.get_orders, filter=req))
+            page_result = await bounded_call(
+                functools.partial(self._client.get_orders, filter=req),
+                timeout_seconds=_REST_TIMEOUT_SECONDS,
+            )
             if not isinstance(page_result, list) or not page_result:
                 break
             page: list[Order] = page_result
@@ -587,8 +572,9 @@ async def get_account_activities(
             # a hung connection must not freeze the loop (ALP-841). ``partial``
             # binds this page's params by value (the call is awaited before the
             # next iteration mutates them).
-            result = await _rest_call(
-                functools.partial(self._client.get, "/account/activities", params)
+            result = await bounded_call(
+                functools.partial(self._client.get, "/account/activities", params),
+                timeout_seconds=_REST_TIMEOUT_SECONDS,
             )
             if not isinstance(result, list) or not result:
                 break

src/alphamind/execution/broker_adapter/retry.py

@@ -22,6 +22,7 @@
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 
+from alphamind.execution.broker_adapter.bounded_streams import bounded_call
 from alphamind.execution.broker_adapter.errors import is_transient
 
 
@@ -63,8 +64,8 @@ class GatewaySubmissionFailed:
 # client factory installs a matching 60s socket-level floor so an orphaned worker
 # eventually unwinds, but a single hung attempt would otherwise park the calling
 # task — including the continuous monitor's fire-close path — for the full socket
-# timeout. Mirroring ``queries._rest_call`` (the read side), this ``wait_for``
-# is the event-loop-level bound that resolves the await within a tight budget and
+# timeout. Mirroring ``queries._REST_TIMEOUT_SECONDS`` (the read side), this is
+# the event-loop-level bound that resolves the await within a tight budget and
 # raises ``TimeoutError`` (classified transient → retried) — the ALP-841 lesson
 # applied to the write side. Below the 60s socket floor so this bound fires first.
 _SUBMIT_TIMEOUT_SECONDS = 30.0
@@ -73,13 +74,13 @@ class GatewaySubmissionFailed:
 async def bounded_broker_call[T](fn: Callable[[], T]) -> T:
     """Run a synchronous Alpaca write call off the event loop, time-bounded.
 
-    Offloads the blocking SDK call to a worker thread and bounds the await with
-    :func:`asyncio.wait_for` at :data:`_SUBMIT_TIMEOUT_SECONDS`, so a hung socket
-    cannot park the calling task (e.g. the monitor's fire-close submission) for
-    the full client-factory socket timeout. Raises :class:`TimeoutError` on
-    expiry, which :func:`submit_with_retry`'s transient classifier retries.
+    Delegates to :func:`~alphamind.execution.broker_adapter.bounded_streams.bounded_call`
+    at :data:`_SUBMIT_TIMEOUT_SECONDS`, so a hung socket cannot park the calling
+    task (e.g. the monitor's fire-close submission) for the full client-factory
+    socket timeout. Raises :class:`TimeoutError` on expiry, which
+    :func:`submit_with_retry`'s transient classifier retries.
     """
-    return await asyncio.wait_for(asyncio.to_thread(fn), timeout=_SUBMIT_TIMEOUT_SECONDS)
+    return await bounded_call(fn, timeout_seconds=_SUBMIT_TIMEOUT_SECONDS)
 
 
 # Backoff schedule per the story: 0.5s, 1s, 2s, 4s, capped at half the

src/alphamind/execution/continuous_monitor/safety_core/price_feed.py

@@ -27,6 +27,7 @@
 from alpaca.data.historical import StockHistoricalDataClient
 from alpaca.data.requests import StockLatestQuoteRequest
 
+from alphamind.execution.broker_adapter import bounded_call
 from alphamind.execution.continuous_monitor.underlying_stream.cache import (
     UnderlyingPriceCache,
     UnderlyingQuote,
@@ -106,9 +107,9 @@ class AlpacaLatestQuoteFetcher:
 
     Wraps a ``StockHistoricalDataClient`` (IEX feed, sharing the trading
     credentials, socket-timeout installed by the client factory). The synchronous
-    SDK call runs off the event loop via :func:`asyncio.to_thread`, time-bounded
-    by :func:`asyncio.wait_for` so a hung socket never wedges the poll loop. The
-    keyed response is mapped through the pure :func:`_latest_quotes_to_underlying`.
+    SDK call runs off the event loop via :func:`bounded_call` so a hung socket
+    never wedges the poll loop. The keyed response is mapped through the pure
+    :func:`_latest_quotes_to_underlying`.
     """
 
     client: StockHistoricalDataClient
@@ -117,8 +118,7 @@ class AlpacaLatestQuoteFetcher:
     async def __call__(self, symbols: frozenset[str]) -> list[UnderlyingQuote]:
         ordered = sorted(symbols)
         request = StockLatestQuoteRequest(symbol_or_symbols=ordered, feed=self.feed)
-        response = await asyncio.wait_for(
-            asyncio.to_thread(self.client.get_stock_latest_quote, request),
-            timeout=_FETCH_TIMEOUT_SECONDS,
+        response = await bounded_call(
+            self.client.get_stock_latest_quote, request, timeout_seconds=_FETCH_TIMEOUT_SECONDS
         )
         return _latest_quotes_to_underlying(response, ordered)