diff --git a/CHANGELOG.md b/CHANGELOG.md index 03f4faca..c04502d7 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/api/config.py b/api/config.py index 94c233fc..2301ff85 100644 --- a/api/config.py +++ b/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 = {} diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 6ee388cf..24b275ad 100644 --- a/api/gateway_chat.py +++ b/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) diff --git a/api/routes.py b/api/routes.py index 189fc524..bfdb383e 100644 --- a/api/routes.py +++ b/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"): diff --git a/api/streaming.py b/api/streaming.py index e65f10e0..c276e90e 100644 --- a/api/streaming.py +++ b/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 diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py index 4517f8c0..68970c4e 100644 --- a/tests/test_issue4812_session_sse_stream.py +++ b/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") diff --git a/tests/test_sse_relay_apperror_closes.py b/tests/test_sse_relay_apperror_closes.py index 2fa14172..38db8727 100644 --- a/tests/test_sse_relay_apperror_closes.py +++ b/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", diff --git a/tests/test_streams_locked_reads.py b/tests/test_streams_locked_reads.py new file mode 100644 index 00000000..c9969724 --- /dev/null +++ b/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) + )