fix: critical reliability fixes for production readiness (#147)

* fix: critical reliability fixes for production readiness

C1: Add 1-hour timeout to _approval_event.wait() and _plan_event.wait()
    to prevent permanent worker thread hangs when users disconnect.

C2: Atomically check-and-start worker thread under Workstream._lock to
    prevent race condition where two concurrent send_message requests
    spawn duplicate workers on the same non-thread-safe ChatSession.

C3: Bound _watch_pending queue to maxsize=20 to prevent OOM under
    heavy watch load with busy workstreams.

H1: Add timeout to proc.wait() (10s) and stderr_thread.join() (5s)
    after SIGKILL to prevent indefinite hang on D-state processes.

H2: Protect _pending_verdicts with _ws_lock at all three mutation sites
    (reset in approve_tools, append in on_intent_verdict, swap-and-clear
    in resolve_approval) to prevent lost verdicts from concurrent
    judge daemon and approval threads.

H3: Bound global SSE queue to maxsize=10000 with put_nowait() and
    contextlib.suppress(queue.Full) for backpressure. Prevents
    unbounded memory growth when fanout thread is overloaded.

H4: Bridge SSE threads for closed workstreams now check ws_id membership
    in _ws_threads before reconnecting, preventing thread leak on
    workstream close.

* fix: address review — verdict lock consistency, watch queue non-blocking, SSE drop logging

- Move _last_verdict_decision set inside _ws_lock in resolve_approval()
  so swap+decision is atomic with on_intent_verdict() reads
- Read _last_verdict_decision under _ws_lock in on_intent_verdict()
- Build heuristic_verdicts locally then assign under lock in approve_tools()
- Use resolve_approval() for timeout path so verdicts are updated consistently
- Watch queue producer uses put_nowait with log on Full (prevents WatchRunner hang)
- Global SSE state broadcasts log on queue.Full instead of silent suppress
- Plan event wait also gets 1-hour timeout (same class of bug as approval)

turnstone/core/session.py

@@ -302,7 +302,7 @@ class ChatSession:
         self._notify_count = 0
         # Watch support: server-level runner injected via set_watch_runner()
         self._watch_runner: Any = None  # WatchRunner | None
-        self._watch_pending: queue.Queue[dict[str, Any]] = queue.Queue()
+        self._watch_pending: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=20)
         self._watch_dispatch_depth = 0
         # Metacognitive nudges: ephemeral prompts for proactive memory use
         self._metacog_state: dict[str, float] = {}
@@ -549,7 +549,12 @@ class ChatSession:
             pending = self._watch_pending
 
             def _enqueue(msg: str) -> None:
-                pending.put({"message": msg})
+                try:
+                    pending.put_nowait({"message": msg})
+                except queue.Full:
+                    log.warning(
+                        "Watch pending queue full, dropping result for ws_id=%s", self._ws_id
+                    )
 
             runner.set_dispatch_fn(self._ws_id, _enqueue)
 
@@ -3509,8 +3514,11 @@ class ChatSession:
                 finally:
                     timer.cancel()
 
-                proc.wait()
-                stderr_thread.join()
+                try:
+                    proc.wait(timeout=10)
+                except subprocess.TimeoutExpired:
+                    log.warning("Process did not exit after SIGKILL, pid=%d", proc.pid)
+                stderr_thread.join(timeout=5)
             finally:
                 os.unlink(script_path)
 

turnstone/mq/bridge.py

@@ -586,11 +586,18 @@ class Bridge:
             event_hooks={"request": [self._inject_auth]},
         ) as sse_client:
             while self._running:
+                # Stop if workstream was closed (thread removed from registry)
+                with self._lock:
+                    if ws_id not in self._ws_threads:
+                        return
                 try:
                     with sse_client.stream("GET", f"/v1/api/events?ws_id={ws_id}") as resp:
                         for data in _iter_sse_data(resp):
                             if not self._running:
                                 break
+                            with self._lock:
+                                if ws_id not in self._ws_threads:
+                                    return
                             self._handle_ws_event(ws_id, data)
                 except Exception as exc:
                     if self._running:

turnstone/server.py

@@ -80,7 +80,7 @@ class WebUI:
 
     # Shared global event queue for state-change broadcasts across all
     # workstreams.  Set by main() before any WebUI instances are created.
-    _global_queue: queue.Queue[dict[str, Any]] | None = None
+    _global_queue: queue.Queue[dict[str, Any]] | None = None  # bounded in main()
     _workstream_mgr: WorkstreamManager | None = None
 
     def __init__(self, ws_id: str = "", user_id: str = "") -> None:
@@ -158,7 +158,10 @@ class WebUI:
             elif state == "error":
                 self._ws_turn_content = []
                 self._ws_turn_content_size = 0
-            WebUI._global_queue.put(event)
+            try:
+                WebUI._global_queue.put_nowait(event)
+            except queue.Full:
+                log.debug("Global SSE queue full, dropping %s event", event.get("type"))
 
     def _broadcast_activity(self) -> None:
         """Send an activity-change event to the global SSE channel."""
@@ -166,14 +169,15 @@ class WebUI:
             with self._ws_lock:
                 activity = self._ws_current_activity
                 activity_state = self._ws_activity_state
-            WebUI._global_queue.put(
-                {
-                    "type": "ws_activity",
-                    "ws_id": self.ws_id,
-                    "activity": activity,
-                    "activity_state": activity_state,
-                }
-            )
+            with contextlib.suppress(queue.Full):
+                WebUI._global_queue.put_nowait(
+                    {
+                        "type": "ws_activity",
+                        "ws_id": self.ws_id,
+                        "activity": activity,
+                        "activity_state": activity_state,
+                    }
+                )
 
     # --- SessionUI protocol ---
 
@@ -312,12 +316,14 @@ class WebUI:
             self._ws_activity_state = "approval"
         self._broadcast_activity()
 
-        # Persist heuristic verdicts and track for user_decision update
-        self._pending_verdicts = []
+        # Persist heuristic verdicts and track for user_decision update.
+        # Build list locally, then assign under lock to avoid racing with
+        # the judge daemon thread's on_intent_verdict() appends.
+        heuristic_verdicts: list[dict[str, Any]] = []
         for item in items:
             hv = item.get("_heuristic_verdict")
             if hv:
-                self._pending_verdicts.append(hv)
+                heuristic_verdicts.append(hv)
                 try:
                     from turnstone.core.storage._registry import get_storage
 
@@ -347,6 +353,9 @@ class WebUI:
                     hv.get("latency_ms", 0),
                 )
 
+        with self._ws_lock:
+            self._pending_verdicts = heuristic_verdicts
+
         # Send approval request and block
         judge_pending = bool(any(it.get("_heuristic_verdict") for it in items))
         self._approval_event.clear()
@@ -356,7 +365,11 @@ class WebUI:
             "judge_pending": judge_pending,
         }
         self._enqueue(self._pending_approval)
-        self._approval_event.wait()
+        if not self._approval_event.wait(timeout=3600):
+            # Approval timed out (e.g., user disconnected). Deny via
+            # resolve_approval so verdicts and state are updated consistently.
+            log.warning("Approval timed out for ws_id=%s", self.ws_id)
+            self.resolve_approval(False, "Approval timed out after 1 hour")
         self._pending_approval = None
         approved, feedback = self._approval_result
 
@@ -436,7 +449,9 @@ class WebUI:
     def on_plan_review(self, content: str) -> str:
         self._plan_event.clear()
         self._enqueue({"type": "plan_review", "content": content})
-        self._plan_event.wait()
+        if not self._plan_event.wait(timeout=3600):
+            log.warning("Plan review timed out for ws_id=%s", self.ws_id)
+            self._plan_result = ""
         return self._plan_result
 
     def on_info(self, message: str) -> None:
@@ -460,7 +475,10 @@ class WebUI:
     def on_rename(self, name: str) -> None:
         """Update the workstream's display name and broadcast to all clients."""
         if WebUI._global_queue is not None:
-            WebUI._global_queue.put({"type": "ws_rename", "ws_id": self.ws_id, "name": name})
+            with contextlib.suppress(queue.Full):
+                WebUI._global_queue.put_nowait(
+                    {"type": "ws_rename", "ws_id": self.ws_id, "name": name}
+                )
 
     def on_intent_verdict(self, verdict: dict[str, Any]) -> None:
         """Deliver LLM judge verdict to frontend via SSE."""
@@ -494,8 +512,10 @@ class WebUI:
             verdict.get("risk_level", "medium"),
             verdict.get("latency_ms", 0),
         )
-        # If approval already resolved, update user_decision immediately
-        decision = self._last_verdict_decision
+        # If approval already resolved, update user_decision immediately.
+        # Read decision under lock to avoid racing with resolve_approval().
+        with self._ws_lock:
+            decision = self._last_verdict_decision
         if decision:
             try:
                 from turnstone.core.storage._registry import get_storage
@@ -508,7 +528,8 @@ class WebUI:
             except Exception:
                 log.debug("Failed to update late verdict user_decision", exc_info=True)
         else:
-            self._pending_verdicts.append(verdict)
+            with self._ws_lock:
+                self._pending_verdicts.append(verdict)
 
     def on_output_warning(self, call_id: str, assessment: dict[str, Any]) -> None:
         """Deliver output guard warning to frontend via SSE + persist."""
@@ -546,11 +567,13 @@ class WebUI:
             }
         )
         # Update user_decision on all tracked verdicts (fire-and-forget).
-        # Swap-and-clear to avoid racing with the daemon judge thread.
-        pending = self._pending_verdicts
-        self._pending_verdicts = []
+        # Swap-and-clear + set decision under lock to avoid racing with
+        # the daemon judge thread's on_intent_verdict() appends.
         decision_str = "approved" if approved else "denied"
-        self._last_verdict_decision = decision_str
+        with self._ws_lock:
+            pending = self._pending_verdicts
+            self._pending_verdicts = []
+            self._last_verdict_decision = decision_str
         if pending:
             try:
                 from turnstone.core.storage._registry import get_storage
@@ -1092,33 +1115,36 @@ async def send_message(request: Request) -> JSONResponse:
     ws, ui = _get_ws(mgr, ws_id)
     if not ws or not ui:
         return JSONResponse({"error": "Unknown workstream"}, status_code=404)
-    if ws.worker_thread and ws.worker_thread.is_alive():
-        ui._enqueue(
-            {
-                "type": "busy_error",
-                "message": "Already processing a request. Please wait.",
-            }
-        )
-        return JSONResponse({"status": "busy"})
-    session = ws.session
-    assert session is not None
+    # Atomically check-and-start to prevent two concurrent workers on the
+    # same session (ChatSession.send() is not thread-safe).
+    with ws._lock:
+        if ws.worker_thread and ws.worker_thread.is_alive():
+            ui._enqueue(
+                {
+                    "type": "busy_error",
+                    "message": "Already processing a request. Please wait.",
+                }
+            )
+            return JSONResponse({"status": "busy"})
+        session = ws.session
+        assert session is not None
 
-    def run() -> None:
-        assert ui is not None
-        try:
-            session.send(message)
-        except GenerationCancelled:
-            # Safety net — send() normally handles this internally.
-            ui._enqueue({"type": "stream_end"})
-            ui.on_state_change("idle")
-        except Exception as e:
-            ui.on_error(f"Error: {e}")
-            ui._enqueue({"type": "stream_end"})
-            ui.on_state_change("error")
+        def run() -> None:
+            assert ui is not None
+            try:
+                session.send(message)
+            except GenerationCancelled:
+                # Safety net — send() normally handles this internally.
+                ui._enqueue({"type": "stream_end"})
+                ui.on_state_change("idle")
+            except Exception as e:
+                ui.on_error(f"Error: {e}")
+                ui._enqueue({"type": "stream_end"})
+                ui.on_state_change("error")
 
-    t = threading.Thread(target=run, daemon=True)
-    ws.worker_thread = t
-    t.start()
+        t = threading.Thread(target=run, daemon=True)
+        ws.worker_thread = t
+        t.start()
     _metrics.record_message_sent()
     with ui._ws_lock:
         ui._ws_messages += 1
@@ -2169,7 +2195,7 @@ def main() -> None:
     )
 
     # Set up global event queue for state-change broadcasts
-    global_queue: queue.Queue[dict[str, Any]] = queue.Queue()
+    global_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=10000)
     global_listeners: list[queue.Queue[dict[str, Any]]] = []
     global_listeners_lock = threading.Lock()
     WebUI._global_queue = global_queue