release: lock-disciplined STREAMS reads (#7339) (#7488)

* fix(streaming): lock-disciplined STREAMS reads via peek_stream()

Writers pop STREAMS under STREAMS_LOCK, but four hot reads used bare
STREAMS.get(): the streaming worker start, the gateway run start, and the two
SSE attach paths in routes.py. Each is individually saved today by GIL-atomic
dict.get plus a None-guard, but the check-then-act windows (get → register,
get → subscribe) race teardown pops, and the bare pattern invites unprotected
copies. peek_stream() centralizes the locked read; all four sites migrate with
no behavioral change.

* test: dual-patch config.STREAMS alongside routes.STREAMS in SSE seam tests

peek_stream() reads the canonical api.config.STREAMS registry; tests that
inject a fake registry by patching only the routes-level re-export left the
helper reading the real (empty) registry and hung the SSE loop to the
pytest-timeout ceiling. Mirror the test_issue6623 pattern: patch both
bindings.

* docs(changelog): stamp #7339 lock-disciplined STREAMS reads

---------

Co-authored-by: Sebastian B <sb@wmc.sh>
Co-authored-by: n <a@n>

CHANGELOG.md

@@ -5,6 +5,8 @@
 
 ### Fixed
 
+- **Hardened a streaming race that could read a stream queue mid-teardown.** Several code paths read the in-memory SSE stream registry with a bare, unlocked `STREAMS.get()`, so a read that raced a concurrent teardown (cancel/finish popping the queue under `STREAMS_LOCK`) could observe-and-use a queue the registry had already released. Registry reads now go through a lock-disciplined `peek_stream()` helper that takes `STREAMS_LOCK`, and a self-enforcing test forbids reintroducing a bare read. Callers keep their existing empty/None-guard fallbacks, so behavior on the common non-race path is unchanged. Thanks @zicochaos. (#7339)
+
 - **Automatic session titles no longer keep an internal placeholder for workspace-prefixed multimodal first turns, and internal scaffolding no longer leaks into generated titles.** Title eligibility compared persisted content that still carried WebUI workspace/attachment scaffolding against a sanitized visible title, so a workspace-prefixed structured multimodal first turn could never be recognized as provisional and stayed stuck on its placeholder title; the same internal scaffolding could also reach the title provider's input. Title generation now sanitizes title-facing structured text (stripping versioned workspace sentinels and generated attachment suffixes) across eligibility, first-exchange generation, manual prefer-latest generation, and adaptive refresh, while preserving genuine user content and native image parts. Persisted and model-facing message content is unchanged (titles only read it), and a real user-chosen title is never misclassified as provisional. Thanks @starship-s. (#7306)
 
 - **The composer footer no longer jitters the transcript while a reply streams.** During SSE streaming, the composer footer's fit-stage probe transiently applied and reverted layout classes, which nudged the transcript viewport a few pixels backward on each frame (a visible clamp/jump on the crown-jewel chat surface). The fit probe now runs synchronously with the transcript geometry frozen and the caller's exact styles restored in a `finally`, so footer sizing is measured without moving the reader's scroll position; adaptive stage selection is unchanged. Thanks @ruizanthony. (#7275)

api/config.py

@@ -9209,6 +9209,21 @@ def create_stream_channel() -> StreamChannel:
 
 STREAMS: dict = {}
 STREAMS_LOCK = threading.Lock()
+
+
+def peek_stream(stream_id):
+    """Lock-disciplined stream queue lookup.
+
+    Writers mutate STREAMS under STREAMS_LOCK (teardown in api/streaming.py,
+    the route layer's start/cancel paths); reads must take the same lock so a
+    read racing a teardown pop can never observe-and-use a queue the registry
+    has already released. Returns the queue or None — callers keep their
+    existing None-guard fallbacks.
+    """
+    with STREAMS_LOCK:
+        return STREAMS.get(stream_id)
+
+
 # stream_id -> session_id owner, populated synchronously before worker startup so
 # stream-id authorization does not depend on worker lifecycle registration.
 STREAM_SESSION_OWNERS: dict = {}

api/gateway_chat.py

@@ -29,6 +29,7 @@ from api.config import (
     coerce_reasoning_effort_for_model,
     gateway_approval_unavailable_reason,
     gateway_supports_approval,
+    peek_stream,
     register_active_run,
     unregister_active_run,
     unregister_stream_owner,
@@ -919,7 +920,7 @@ def _run_gateway_chat_streaming(
     the configured Gateway API server into those local events and persists the
     final user/assistant turn back into the WebUI session.
     """
-    q = STREAMS.get(stream_id)
+    q = peek_stream(stream_id)
     if q is None:
         _finish_gateway_run_starting(stream_id, result="fallback")
         _clear_gateway_run_starting(stream_id)

api/routes.py

@@ -2896,6 +2896,7 @@ from api.config import (
     register_session_writeback_owner,
     clear_session_writeback_owner_if_owned,
     stream_owner_session_id,
+    peek_stream,
     unregister_stream_owner,
     CHAT_LOCK,
     _get_session_agent_lock,
@@ -18793,7 +18794,7 @@ def _handle_sse_stream(handler, parsed):
     # rather than silently skip journal events.
     resume_cursor = _chat_stream_resume_cursor(handler, qs, stream_id)
     resume_after_seq, resume_requested, resume_raw_cursor, runner_resume_cursor = resume_cursor
-    stream = STREAMS.get(stream_id)
+    stream = peek_stream(stream_id)
     if stream is None:
         # Runner-observe path: consume the ALREADY-RESOLVED cursor — do not
         # re-parse query params or re-read the header (Codex r2 #3 / r3). The
@@ -18936,7 +18937,7 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_id):
 
     def attach_active_stream():
         stream_id = _active_run_stream_for_session(session_id)
-        stream = STREAMS.get(stream_id) if stream_id else None
+        stream = peek_stream(stream_id) if stream_id else None
         if stream is None:
             return None, None, None, stream_id
         if hasattr(stream, "subscribe_with_snapshot"):

api/streaming.py

@@ -38,6 +38,7 @@ from api.config import (
     _set_thread_env, _clear_thread_env,
     register_active_run, update_active_run, unregister_active_run,
     unregister_stream_owner,
+    peek_stream,
     stream_owner_session_id,
     session_writeback_owner,
     clear_session_writeback_owner_if_owned,
@@ -8831,7 +8832,7 @@ def _run_agent_streaming(
     """
     _turn_route_model = model
     _turn_route_provider = model_provider
-    q = STREAMS.get(stream_id)
+    q = peek_stream(stream_id)
     if q is None:
         # The stream was cancelled before the worker started; the route layer
         # already registered the stream owner, so release it here to avoid

tests/test_issue4812_session_sse_stream.py

@@ -538,6 +538,12 @@ def test_session_route_live_delivery_skips_replayed_active_run_items(monkeypatch
     )
     monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active")
     monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    import api.config as config
+    monkeypatch.setattr(
+        config,
+        "STREAMS",
+        {"run_active": stream},
+    )
     monkeypatch.setattr(
         routes,
         "read_session_run_events",
@@ -611,6 +617,12 @@ def test_session_route_breaks_on_terminal_even_when_reconciliation_replayed_it(m
     )
     monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_a, **_k: "run_active")
     monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    import api.config as config
+    monkeypatch.setattr(
+        config,
+        "STREAMS",
+        {"run_active": stream},
+    )
     # Reconciliation replays the active run's terminal event (run_active:1) — same id
     # as the queued live copy, at the cutoff.
     monkeypatch.setattr(
@@ -671,6 +683,12 @@ def test_session_route_unsubscribes_when_replay_disconnects(monkeypatch):
     )
     monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active")
     monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    import api.config as config
+    monkeypatch.setattr(
+        config,
+        "STREAMS",
+        {"run_active": stream},
+    )
     monkeypatch.setattr(
         routes,
         "read_session_run_events",
@@ -722,6 +740,12 @@ def test_session_route_live_delivery_without_cursor_keeps_buffered_active_items(
     )
     monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active")
     monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    import api.config as config
+    monkeypatch.setattr(
+        config,
+        "STREAMS",
+        {"run_active": stream},
+    )
     monkeypatch.setattr(
         routes,
         "read_session_run_events",
@@ -764,6 +788,12 @@ def test_session_route_reconciles_late_attach_cutoff_once(monkeypatch):
     monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: SimpleNamespace(session_id=sid, compact=lambda **_kwargs: {}))
     monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active" if available["value"] else None)
     monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    import api.config as config
+    monkeypatch.setattr(
+        config,
+        "STREAMS",
+        {"run_active": stream},
+    )
     monkeypatch.setattr(
         routes,
         "read_session_run_events",
@@ -808,6 +838,12 @@ def test_session_route_emits_snapshot_when_reconciliation_fails(monkeypatch):
     )
     monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active")
     monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    import api.config as config
+    monkeypatch.setattr(
+        config,
+        "STREAMS",
+        {"run_active": stream},
+    )
     monkeypatch.setattr(
         routes,
         "read_session_run_events",
@@ -853,6 +889,12 @@ def test_session_route_bounds_sent_event_id_deduplication(monkeypatch):
     monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: SimpleNamespace(session_id=sid, compact=lambda **_kwargs: {}))
     monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active")
     monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    import api.config as config
+    monkeypatch.setattr(
+        config,
+        "STREAMS",
+        {"run_active": stream},
+    )
 
     handler = _FakeHandler()
     routes._handle_session_sse_stream_for_session(handler, urlparse("/api/sessions/session_1/events"), "session_1")

tests/test_sse_relay_apperror_closes.py

@@ -97,6 +97,12 @@ def test_chat_sse_stream_unsubscribes_on_apperror(monkeypatch):
     stream = _FakeStream()
     monkeypatch.setattr(routes, "_stream_id_visible_to_request_profile", lambda *_a, **_k: True)
     monkeypatch.setattr(routes, "STREAMS", {"run1": stream})
+    import api.config as config
+    monkeypatch.setattr(
+        config,
+        "STREAMS",
+        {"run1": stream},
+    )
     monkeypatch.setattr(
         routes,
         "_sse_replay_run_journal_gap_checked",

tests/test_streams_locked_reads.py

@@ -0,0 +1,71 @@
+"""STREAMS registry reads must go through the lock-disciplined peek_stream().
+
+Writers mutate config.STREAMS under STREAMS_LOCK (teardown in api/streaming.py
+and the route layer); bare STREAMS.get() reads race those pops. These tests pin
+the peek_stream() helper and forbid reintroducing bare STREAMS.get() calls in
+the api package outside api/config.py itself.
+"""
+
+import queue
+import re
+import threading
+from pathlib import Path
+
+import api.config as config
+
+REPO = Path(__file__).resolve().parents[1]
+API_DIR = REPO / "api"
+
+
+def test_peek_stream_returns_registered_queue_and_none_for_missing():
+    stream_id = "peek-stream-registered"
+    q = queue.Queue()
+    registered = False
+    try:
+        with config.STREAMS_LOCK:
+            config.STREAMS[stream_id] = q
+            registered = True
+        assert config.peek_stream(stream_id) is q
+        assert config.peek_stream("nope") is None
+    finally:
+        if registered:
+            with config.STREAMS_LOCK:
+                config.STREAMS.pop(stream_id, None)
+
+
+def test_peek_stream_acquires_streams_lock(monkeypatch):
+    events = []
+    real_lock = threading.Lock()
+
+    class _RecordingLock:
+        def __enter__(self):
+            events.append("enter")
+            real_lock.acquire()
+            return real_lock
+
+        def __exit__(self, exc_type, exc, tb):
+            real_lock.release()
+            return False
+
+    monkeypatch.setattr(config, "STREAMS_LOCK", _RecordingLock())
+    # Missing id: the locked lookup still happens and returns None.
+    assert config.peek_stream("peek-stream-lock-check") is None
+    assert events, "peek_stream did not acquire STREAMS_LOCK"
+
+
+def test_no_bare_streams_get_outside_config():
+    pattern = re.compile(r"STREAMS\.get\(")
+    offenders = []
+    for path in sorted(API_DIR.glob("*.py")):
+        if path.name == "config.py":
+            continue
+        text = path.read_text(encoding="utf-8")
+        for lineno, line in enumerate(text.splitlines(), start=1):
+            if pattern.search(line):
+                offenders.append(
+                    f"{path.relative_to(REPO)}:{lineno}: {line.strip()}"
+                )
+    assert not offenders, (
+        "bare STREAMS.get( outside api/config.py; use config.peek_stream() "
+        "so reads take STREAMS_LOCK:\n" + "\n".join(offenders)
+    )