Fix — /ws/bus subscriber thread leak

Every client that connects to /ws/bus/{session_id} spawns a thread that never exits after disconnect.

Bug

Root cause: _subscribe() in gateway.py:148 runs in a thread-pool thread via run_in_executor with an unbounded while True loop. The WebSocketDisconnect handler at line 169 exits the coroutine but has no way to signal the thread. The thread keeps running — and holding a ZMQ subscriber socket — until the process restarts.

Lifetime diagram (buggy)

client connects
  → coroutine starts
  → run_in_executor(_subscribe)  ← thread T1 starts, loops forever
  → coroutine pumps queue

client disconnects
  → WebSocketDisconnect caught
  → coroutine exits
  → T1 still running ✗  ZMQ socket still open ✗

Lifetime diagram (fixed)

client connects
  → stop = threading.Event()
  → run_in_executor(_subscribe)  ← thread T1 starts, checks stop
  → coroutine pumps queue

client disconnects
  → WebSocketDisconnect caught
  → finally: stop.set()
  → T1 exits within ≤200 ms (next receive_with_timeout wake-up) ✓

Fix

Approach: Add a threading.Event stop flag. Thread checks it on every loop iteration. Coroutine sets it in a finally block, guaranteeing cleanup on both normal close and error.

Single file: src/local/api/gateway.py, function ws_bus.

Before

    loop = asyncio.get_event_loop()
    queue: asyncio.Queue = asyncio.Queue()

    def _subscribe() -> None:
        sub = ZmqSubscriber(PROXY_BACKEND_ADDR, subscriptions=[""], bind=False)
        try:
            while True:
                msg = sub.receive_with_timeout(200)
                if msg is not None:
                    asyncio.run_coroutine_threadsafe(queue.put(msg), loop)
        finally:
            sub.close()

    loop.run_in_executor(None, _subscribe)

    try:
        while True:
            env: MessageEnvelope = await queue.get()
            await websocket.send_json({...})
    except WebSocketDisconnect:
        pass

After

    import threading
    loop = asyncio.get_event_loop()
    queue: asyncio.Queue = asyncio.Queue()
    stop = threading.Event()

    def _subscribe() -> None:
        sub = ZmqSubscriber(PROXY_BACKEND_ADDR, subscriptions=[""], bind=False)
        try:
            while not stop.is_set():
                msg = sub.receive_with_timeout(200)
                if msg is not None:
                    asyncio.run_coroutine_threadsafe(queue.put(msg), loop)
        finally:
            sub.close()

    loop.run_in_executor(None, _subscribe)

    try:
        while True:
            env: MessageEnvelope = await queue.get()
            await websocket.send_json({...})
    except WebSocketDisconnect:
        pass
    finally:
        stop.set()
threading is already in the stdlib — no new dependency. receive_with_timeout(200) already exists, so the thread wakes every 200 ms regardless. The finally block fires on both clean close and any exception, so no code path leaks.

File Map

FileChange
src/local/api/gateway.pyAdd stop = threading.Event(); change while True → while not stop.is_set(); add finally: stop.set() to coroutine

Acceptance Criteria