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

[jatassi]

Summary

• alpaca-py's stream mutators (subscribe_quotes / unsubscribe_quotes / subscribe_trade_updates) are loop-affined sync APIs: on a running stream they block the calling thread on an untimed Future.result() whose resolution needs the very loop the caller holds — a guaranteed self-deadlock from the monitor's subscription diff, wedging the monitor ~150 s on every position-set-changing invocation (4 production wedges since 2026-06-10).
• New broker_adapter/bounded_streams.py: bounded_call (off-thread + wait_for-bounded offload, generalizing queries._rest_call, now deleted) plus BoundedStockDataStream / BoundedTradingStream facades; the stream factories hand out only facades.
• StockDataStreamProtocol / _SubscribableStream mutators are now async by contract, so a bare sync vendor call on the loop is unrepresentable at the seam; all call sites await. retry.bounded_broker_call and the safety-core quote fetcher collapse onto bounded_call.
• A facade TimeoutError from the diff path rides the existing budget-counted reconnect; the rebuilt connection re-subscribes the full target set (pinned by a new test).

Closes https://linear.app/alphamind-jatassi/issue/ALP-946/alpaca-py-sync-subscribeunsubscribe-self-deadlocks-the-monitor-event

Test plan

• CI (.github/workflows/ci.yml) green on the PR — lint on Linux + full pytest on Windows
• Local lint chain clean (ruff check, ruff format --check, mypy, lint-imports)
• ALP-946 acceptance criteria all met (note: bounded_call's keyword is timeout_seconds, not the spec's literal timeout — ruff ASYNC109 flags async defs with a timeout parameter; the spec's own facade constructors already use timeout_seconds)

🤖 Generated with Claude Code

[linear]

ALP-946 alpaca-py sync subscribe/unsubscribe self-deadlocks the monitor event loop on every position-set change — wrap the broker websocket seams in bounded async facades

Summary

Corrected from the original report: the report attributed the block to a degraded/mid-reconnect websocket whose subscribe future "never resolves," resolving "in milliseconds" when healthy. The installed library refutes that: StockDataStream.subscribe_quotes → DataStream._subscribe calls asyncio.run_coroutine_threadsafe(self._send_subscribe_msg(), self._loop).result() (.venv/Lib/site-packages/alpaca/data/live/websocket.py:277-280) where self._loop is the monitor's own event loop (set at websocket.py:319 from _run_forever, which task.py:342 runs as a sibling task on that loop). Called from the loop's own thread, the scheduled coroutine can only run when the loop is free — and the loop is blocked waiting on the future. That is a guaranteed self-deadlock whenever the stream is connected (_running=True), and a silent no-op handler registration when it is not (_running=False skips the blocking branch). Connection health is the gate, not the cause: healthy connection = certain deadlock, degraded connection = no block at all — the inverse of the report. unsubscribe_quotes → _unsubscribe (websocket.py:296-300) has the identical branch, and production has wedged on both.

Also corrected: this is not a one-off. The faulthandler deadman (ALP-941) has captured four wedges — 2026-06-10 13:02 MT (the reported 42-minute outage, bounded only by the operator because of ALP-945) and three on 2026-06-11 at 07:37, 11:03, and 13:02 MT, each immediately after a pipeline run whose Phase-1 changed the open-position ticker set, each auto-recovered by the now-working out-of-process watchdog in ~150 s + restart + replay. The monitor currently takes a deterministic ~150 s full outage (no fills, no options stops, no entry-window, no monitor-side guardrails; broker bracket floor still in place) on every position-set-changing invocation, ~3×/trading day.

The report's preserved-artifact pointer is also stale: diagnostics/monitor_faulthandler_20260610_1302_subscribe_quotes_wedge.log does not exist (prod diagnostics/ is empty). All four dumps live appended in C:\Users\jacks\AlphaMind\logs\monitor_faulthandler.log.

Evidence (prod logs + prod DB + installed library; log times MT, DB/IDs UTC)

• Frozen frames — logs/monitor_faulthandler.log holds 4 dumps (markers at file lines 1/60/131/190). Wedged frames: 3× subscribe_quotes (websocket.py:280 _subscribe → stock.py:79 → task.py:420 _periodic_subscription_diff, dump lines 47-49, 118-120, 236-238) and 1× unsubscribe_quotes (websocket.py:300 _unsubscribe → stock.py:167 → task.py:422, dump lines 177-179). Above each: concurrent/futures/_base.py:445 result / threading.py:369 wait — the untimed .result().
• Wedge 1 (2026-06-10) — monitor_watchdog.log.2026-06-10: 13:02:50 CRITICAL heartbeat stale (152s > bound 150s) → last beat ≈ 13:00:18 MT, ~18 s after pre_close inv-20260610T190000Z-f379259e (19:00:00Z) started Phase-1. Every nssm restart refused (CalledProcessError, retried each tick — ALP-945, since fixed in PR fix(execution): remove CC's SCM dependency edges so the monitor watchdog's restart isn't refused (ALP-945) #350). Manual §8.9 recovery: monitor.log.2026-06-10:403 session restart 13:42:03. Wedged PID 10064, git_sha 3f10c8e3 (process_lifetimes, started 17:06:16Z).
• Wedges 2-4 (2026-06-11) — monitor_watchdog.log: heartbeat-stale restarts at 07:37:46, 11:03:01, 13:02:41 MT, each matching a monitor restart in process_lifetimes (13:37:50Z, 17:03:05Z, 19:02:44Z) and each ~2-3 min after a run start (market_open 13:35Z, mhr 17:00Z, pre_close 19:00Z). The 11:03 wedge is the unsubscribe frame — a position-close delta.
• Trigger = Phase-1 position-set integration. compute_target_underlyings reads get_open_positions() (subscriptions.py:34-45); position rows flip to OPEN at the next invocation's Phase-1 (two-phase write discipline), so the 30 s-cadence diff (config/continuous_monitor.yaml:58 subscription_refresh_seconds: 30) sees the new ticker set seconds after a run starts and issues the fatal delta. Prod positions: COP/BAC entries stamped 17:31Z 06-10 (13:00 ET run's fills) → integrated at the 19:00Z run → 19:00:18Z wedge; GOOGL/USB entered 17:38Z 06-11 → 19:02Z wedge.
• Fills missed during wedge 1, then recaptured — broker filled the BAC 18-share cover 19:36:58Z (fill-45c1cebdc40a94fe, order e2016766…) and the DVN 32-share entry 19:38:50Z (fill-597943b3909172b1, order 5bd367e3…) while the loop was frozen; both rows carry persistence_timestamp 19:42:05Z — five seconds after the manual restart, written by the startup replay/backfill. No data was lost.
• Why only since 2026-06-10 — the blocking branch requires _running=True at the instant of a delta. The deadman has existed only since the ALP-941 deploy (watchdog start 2026-06-09 23:31 MT), and every wedge it has ever captured (4/4) is this frame. Before 06-09 the monitor's market-data websocket rarely held a connection (ALP-940 single-connection collision on 06-08/09; no post-genesis equity positions before 06-08), so deltas landed in the no-op _running=False branch. ALP-940's fix gave the monitor a stably connected stream; the first position-turnover day after it (06-10) produced the first wedge, and every set-changing run since has reproduced it. (Pre-ALP-941-era wedges were separately root-caused with different mechanisms — ALP-819, ALP-841.)
• The call sites are original to the monitor (_periodic_subscription_diff landed in PR Continuous monitor: implement work tree (ALP-123) #48, 2026-05-11, commit a0dd5ca) — a latent defect, not a regression.

Root cause (confirmed)

1. alpaca-py's stream-mutator methods are loop-affined synchronous APIs: when the stream is running they marshal the actual send onto self._loop via run_coroutine_threadsafe and block the calling thread on an untimed Future.result() (websocket.py:277-280, :296-300; trading/stream.py:113-128 is the same pattern). They are designed to be called from a thread other than the stream's loop thread.
2. The monitor calls them on the stream's own loop thread: _periodic_subscription_diff (task.py:420,422) is an asyncio task on the monitor loop, and the stream's _run_forever (which sets _loop to that same loop) runs as its sibling (task.py:342). The scheduled _send_subscribe_msg can never execute while .result() holds the loop thread → permanent freeze of the entire process loop: heartbeat, fills, stops, SSE, in-loop watchdog.
3. The remaining bare sync call sites are safe only by construction-order: connect-time subscribe_quotes (task.py:324,328) and the fill stream's subscribe_trade_updates (broker_adapter/fill_stream.py:381) run on freshly built streams before _run_forever starts (_running=False, blocking branch skipped). Nothing in the code marks or enforces that invariant — the next caller of a sync mutator on a running stream reintroduces the wedge.

The disease is bare loop-affined sync broker-websocket calls on the event loop (the ALP-841 class, which fixed its REST instances via to_thread offload but did not touch the stream mutators). The 06-10 42-minute blast radius (vs today's ~150 s) was ALP-945's watchdog-restart refusal — fixed separately.

Why it escaped tests

Every test fakes the vendor stream at the sanctioned broker boundary (StockDataStreamProtocol / _SubscribableStream) with benign in-memory objects — the deadlock lives inside the real alpaca-py class behind that seam, exactly the mock-the-failing-seam case (same shape as ALP-940). The existing diff tests (tests/execution/continuous_monitor/underlying_stream/test_task.py:351,391) therefore prove delta computation, not loop safety. No test could assert "the sync mutator is never invoked on the loop thread" because the contract was never encoded in a seam — the facade below makes it a testable property.

Scope — bound every sync alpaca-py websocket call behind an async facade (operator-selected scope, 2026-06-11)

(A) Shared bounded-offload helper in broker_adapter. Add module src/alphamind/execution/broker_adapter/bounded_streams.py with async def bounded_call[T](fn: Callable[..., T], /, *args: object, timeout: float) -> T, implemented as await asyncio.wait_for(asyncio.to_thread(fn, *args), timeout=timeout) — the generalization of queries._rest_call (queries.py:78-93; carry its rationale docstring over: offload keeps the loop turning, the bound guarantees the await resolves, the client-factory socket timeout remains defence-in-depth). Export bounded_call from broker_adapter/__init__.py. Delete queries._rest_call; its call sites use bounded_call(fn, timeout=_REST_TIMEOUT_SECONDS) (constant stays in queries.py). Collapse price_feed.AlpacaLatestQuoteFetcher.__call__'s inline wait_for(to_thread(...)) (safety_core/price_feed.py:120-123) onto bounded_call with its existing _FETCH_TIMEOUT_SECONDS.

(B) Stock-stream facade. In bounded_streams.py, add BoundedStockDataStream wrapping a vendor StockDataStream: async def subscribe_quotes(handler, *symbols) and async def unsubscribe_quotes(*symbols) delegate via bounded_call(self._stream.subscribe_quotes, handler, *symbols, timeout=self._timeout_seconds) (resp. unsubscribe); async def _run_forever() and async def stop_ws() are plain awaited delegations. Constructor: (stream, *, timeout_seconds: float = _STREAM_CALL_TIMEOUT_SECONDS) with _STREAM_CALL_TIMEOUT_SECONDS = 30.0 a module constant mirroring _REST_TIMEOUT_SECONDS (same-comment convention as price_feed.py:38-42).

(C) Trading-stream facade. In bounded_streams.py, add BoundedTradingStream wrapping a vendor TradingStream: async def subscribe_trade_updates(handler) via bounded_call, plus awaited _run_forever delegation. client_factory.build_trading_stream() (client_factory.py:130-137) returns the vendor stream wrapped in BoundedTradingStream (the consumer-facing type is already object via TradingStreamFactory, fill_stream_consumer/task.py:57).

(D) Make the monitor seams async. StockDataStreamProtocol.subscribe_quotes / unsubscribe_quotes (underlying_stream/task.py:80-82) become async def; DefaultAlpacaStreamFactory.build() (task.py:117-138) returns BoundedStockDataStream(StockDataStream(...)). All four call sites await: connect-time (task.py:324,328) and the diff deltas (task.py:420,422). _SubscribableStream.subscribe_trade_updates (broker_adapter/fill_stream.py:330) becomes async def and its call site awaits (fill_stream.py:381). After this, no caller of either Protocol can issue a bare sync mutator on the loop — the bug is unrepresentable at the seam, and the construction-order subtlety in Root-cause step 3 disappears.

(E) Timeout failure path — keep the existing reconnect semantics. A TimeoutError from a facade call during the diff propagates out of _periodic_subscription_diff, through the connection TaskGroup, into run_underlying_stream's budget-counted reconnect branch (task.py:250-269); the rebuild re-subscribes the full current target set at connect time, healing any half-applied subscription state. No new classification, no budget-neutral special case.

Acceptance criteria

• bounded_call exists in broker_adapter.bounded_streams, is exported from broker_adapter, executes its callable off the calling loop's thread (unit test: callable records threading.get_ident(); assert it differs from the loop thread's), passes through positional args and return value, and raises TimeoutError when the callable blocks past timeout.
• queries._rest_call is deleted; queries.py call sites use bounded_call with _REST_TIMEOUT_SECONDS; AlpacaLatestQuoteFetcher uses bounded_call with _FETCH_TIMEOUT_SECONDS; tests/execution/continuous_monitor/safety_core/ and the broker-adapter query tests stay green.
• BoundedStockDataStream.subscribe_quotes / unsubscribe_quotes invoke the wrapped vendor methods off the loop thread with identical arguments, and _run_forever / stop_ws delegate — unit tests with a fake vendor object recording thread id + args.
• BoundedTradingStream.subscribe_trade_updates does the same, and build_trading_stream() returns a BoundedTradingStream.
• StockDataStreamProtocol and _SubscribableStream declare the mutators async; no bare synchronous subscribe_* / unsubscribe_* call on a vendor stream remains anywhere in src/alphamind/ (grep + inspection).
• A facade TimeoutError raised from the diff path produces a budget-counted reconnect and a fresh connection that re-subscribes the full target set — one test driving run_underlying_stream with a fake factory whose first connection's subscribe delta raises TimeoutError.
• Existing diff tests (test_added_position_triggers_subscribe_quotes, test_removed_position_triggers_unsubscribe_quotes) stay green with async fakes; no per-call-site thread assertions duplicated in test_task.py (the facade tests own loop-affinity).
• Lint chain green: uv run ruff check . && uv run ruff format . && uv run mypy && uv run lint-imports.

Verification

• Scoped runs while implementing: uv run pytest tests/execution/broker_adapter/ tests/execution/continuous_monitor/underlying_stream/ tests/execution/continuous_monitor/fill_stream_consumer/ tests/execution/continuous_monitor/safety_core/ -n auto.
• Authoritative gate: the ci workflow (full suite, Windows, -n auto).
• Post-deploy prod spot-check (log times MT): across the next position-set-changing invocation, monitor_watchdog.log gains no heartbeat stale CRITICAL, monitor_faulthandler.log gains no new dump, and monitor.log shows the subscription diff surviving the run boundary.

Reading

• src/alphamind/execution/continuous_monitor/underlying_stream/task.py — :72-99 the Protocol seam to make async; :107-138 the factory that wraps; :283-381 the connection lifecycle the TimeoutError rides; :399-423 the deadlocking diff.
• .venv/Lib/site-packages/alpaca/data/live/websocket.py:264-345 — the _subscribe/_unsubscribe blocking branch, _running lifecycle, _loop assignment. The mechanism; read before writing the facade.
• .venv/Lib/site-packages/alpaca/trading/stream.py:107-134 — TradingStream.subscribe_trade_updates, the identical branch behind the fill seam.
• src/alphamind/execution/broker_adapter/queries.py:60-93 — _REST_TIMEOUT_SECONDS + _rest_call: the helper to generalize and the docstring rationale to preserve.
• src/alphamind/execution/continuous_monitor/safety_core/price_feed.py:38-47,103-124 — the ALP-940 bounded-offload precedent collapsing onto bounded_call.
• src/alphamind/execution/broker_adapter/fill_stream.py:323-397 — _SubscribableStream + the pre-run subscribe call going async.
• src/alphamind/execution/broker_adapter/client_factory.py:75-140 — build_trading_stream (the wrap point) and the socket-timeout defence-in-depth note.
• tests/execution/continuous_monitor/underlying_stream/test_task.py:351-430 — the diff tests whose fakes become async.
• .importlinter — broker_adapter sits below continuous_monitor; the new module adds no upward import.

One-time prod data cleanup

No bulk cleanup needed. Both wedge-1 fills were recaptured by the startup replay (persistence_timestamp 19:42:05Z on fill-45c1cebdc40a94fe and fill-597943b3909172b1), positions and cash are consistent, and the watchdog auto-recovers each recurrence until the fix deploys. The only artifact correction: the frozen frames live in logs/monitor_faulthandler.log (4 appended dumps), not in diagnostics/ as the original report stated.

Separate observation — do NOT bundle

FileHeartbeatSink.beat() intermittently fails with PermissionError [WinError 5] on os.replace(monitor.heartbeat.tmp → monitor.heartbeat) (process_supervision/heartbeat.py:64, caught and WARNING-logged at supervisor.py:295-299; observed monitor.log.2026-06-10 12:39:19 and 12:57:05 MT, before the wedge). A Windows replace-while-reader-holds-the-file race (watchdog probe / SMB / AV). Risk: if failures ever persist past the 150 s stall bound, the external watchdog restarts a healthy monitor. Distinct root cause this fix does not touch — file it as its own issue once cap headroom exists (cap check 2026-06-11: 261/250, over).

Related

• ALP-941 — built the out-of-process watchdog + faulthandler deadman that captured all four frames and bounds each recurrence to ~150 s; this issue removes the freeze cause those guards keep absorbing.
• ALP-945 — the watchdog's nssm-restart refusal that turned wedge 1 into a 42-minute outage (fixed, PR fix(execution): remove CC's SCM dependency edges so the monitor watchdog's restart isn't refused (ALP-945) #350).
• ALP-940 — causal gate: its fix gave the monitor a stably connected market-data websocket (_running=True), first making the deadlock branch reachable on every delta; also the bounded-offload pattern Scope (A) generalizes.
• ALP-841 — the bare-sync-on-loop class: its fix offloaded the REST instances (get_orders etc.); the stream mutators were the remaining members. This issue closes the class at the websocket boundary.
• ALP-832 — built _periodic_subscription_diff's staleness/reconnect machinery the TimeoutError path reuses.

Review in Linear