diff --git a/api/gateway_chat.py b/api/gateway_chat.py index c759c68d..0807af48 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -981,7 +981,7 @@ def _run_gateway_chat_streaming( except Exception: logger.debug("Failed to note gateway event_id %s for stream %s", event_id, stream_id, exc_info=True) try: - queue_item = (event, data, event_id) if event_id and hasattr(q, "subscribe_with_snapshot") else (event, data) + queue_item = (event, data, event_id) if hasattr(q, "subscribe_with_snapshot") else (event, data) q.put_nowait(queue_item) except Exception: logger.debug("Failed to put gateway event to queue") diff --git a/api/routes.py b/api/routes.py index 062a3772..5a6a319f 100644 --- a/api/routes.py +++ b/api/routes.py @@ -19031,14 +19031,14 @@ def _handle_sse_stream(handler, parsed): continue if len(item) >= 3: event, data, queued_event_id = item[0], item[1], item[2] + event_id = queued_event_id else: event, data = item - queued_event_id = STREAM_LAST_EVENT_ID.get(stream_id) + event_id = STREAM_LAST_EVENT_ID.get(stream_id) # Stage-364: emit `id:` from STREAM_LAST_EVENT_ID side-channel so # the frontend's `_lastRunJournalSeq` cursor advances during live # streaming. Without this, mid-stream error→replay would arrive # with after_seq=0 and double-render every journaled event. - event_id = queued_event_id or STREAM_LAST_EVENT_ID.get(stream_id) event_seq = _run_journal_same_run_seq(event_id, stream_id) if replay_cutoff_seq is not None and event_seq is not None and event_seq <= replay_cutoff_seq: continue @@ -19181,10 +19181,10 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_id): continue if len(item) >= 3: event, data, queued_event_id = item[0], item[1], item[2] + event_id = queued_event_id else: event, data = item - queued_event_id = STREAM_LAST_EVENT_ID.get(active_stream_id) - event_id = queued_event_id or STREAM_LAST_EVENT_ID.get(active_stream_id) + event_id = STREAM_LAST_EVENT_ID.get(active_stream_id) event_seq = _run_journal_same_run_seq(event_id, active_stream_id) _is_terminal = event in SSE_RELAY_CLOSE_EVENTS _already_sent = ( diff --git a/api/streaming.py b/api/streaming.py index 29cda8be..5b8f4108 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -9600,7 +9600,7 @@ def _run_agent_streaming( except Exception: logger.debug("Failed to note event_id %s for stream %s", event_id, stream_id, exc_info=True) try: - queue_item = (event, data, event_id) if event_id and hasattr(q, "subscribe_with_snapshot") else (event, data) + queue_item = (event, data, event_id) if hasattr(q, "subscribe_with_snapshot") else (event, data) q.put_nowait(queue_item) except Exception: logger.debug("Failed to put event to queue") diff --git a/tests/test_run_journal_routes.py b/tests/test_run_journal_routes.py index 95077d2d..ad1ccb5b 100644 --- a/tests/test_run_journal_routes.py +++ b/tests/test_run_journal_routes.py @@ -797,6 +797,61 @@ def test_replay_run_journal_skips_metering_rows(monkeypatch): assert "id: run_1:2\n" not in body +def test_live_unjournaled_metering_does_not_inherit_latest_event_id(monkeypatch): + import api.routes as routes + + class FakeStream: + def __init__(self): + self.q = queue.Queue() + self.q.put_nowait(("metering", {"tps": 47.3}, None)) + self.q.put_nowait(("token", {"text": "visible"}, "run_1:1")) + self.q.put_nowait(("stream_end", {}, "run_1:2")) + self.unsubscribed = False + + def subscribe_with_snapshot(self): + return self.q, {"last_event_id": "run_1:1", "offline_buffered_events": 3} + + def unsubscribe(self, q): + self.unsubscribed = q is self.q + + class Handler: + def __init__(self): + self.wfile = io.BytesIO() + self.headers = {} + + def send_response(self, _code): + pass + + def send_header(self, _name, _value): + pass + + def end_headers(self): + pass + + handler = Handler() + stream = FakeStream() + monkeypatch.setattr( + routes, + "find_run_summary", + lambda stream_id: { + "session_id": "session_1", + "run_id": stream_id, + "terminal": False, + }, + ) + monkeypatch.setattr(routes, "read_run_events", lambda *_args, **_kwargs: {"events": []}) + monkeypatch.setitem(routes.STREAM_LAST_EVENT_ID, "run_1", "run_1:1") + monkeypatch.setitem(routes.STREAMS, "run_1", stream) + + routes._handle_sse_stream(handler, urlparse("/api/chat/stream?stream_id=run_1")) + + body = handler.wfile.getvalue().decode("utf-8") + frames = [frame for frame in body.split("\n\n") if frame.strip()] + assert frames[0] == "event: metering\ndata: {\"tps\": 47.3}" + assert "id: run_1:1\nevent: token\ndata: {\"text\": \"visible\"}" in frames + assert stream.unsubscribed is True + + def test_active_stream_replay_keeps_items_for_new_run_with_same_seq_range(monkeypatch): import api.routes as routes diff --git a/tests/test_run_journal_streaming_static.py b/tests/test_run_journal_streaming_static.py index 7224dbcf..10fdb63d 100644 --- a/tests/test_run_journal_streaming_static.py +++ b/tests/test_run_journal_streaming_static.py @@ -26,7 +26,7 @@ def test_streaming_journals_sse_events_before_queue_delivery(): assert put_idx < journal_idx < queue_idx assert "Failed to append run journal event" in block - assert "queue_item = (event, data, event_id) if event_id and hasattr(q, \"subscribe_with_snapshot\") else (event, data)" in block + assert 'queue_item = (event, data, event_id) if hasattr(q, "subscribe_with_snapshot") else (event, data)' in block diff --git a/tests/test_stage364_opus_live_sse_event_id.py b/tests/test_stage364_opus_live_sse_event_id.py index bc141f5c..83885895 100644 --- a/tests/test_stage364_opus_live_sse_event_id.py +++ b/tests/test_stage364_opus_live_sse_event_id.py @@ -57,7 +57,7 @@ def test_stream_channel_queue_item_carries_per_event_id_with_legacy_fallback(): """StreamChannel queue items need per-frame ids; legacy queues stay 2-tuples.""" put_def_idx = STREAMING_PY.find("def put(event, data):") put_body = STREAMING_PY[put_def_idx:put_def_idx + 2500] - assert 'queue_item = (event, data, event_id) if event_id and hasattr(q, "subscribe_with_snapshot") else (event, data)' in put_body, ( + assert 'queue_item = (event, data, event_id) if hasattr(q, "subscribe_with_snapshot") else (event, data)' in put_body, ( "StreamChannel events must carry their own event_id while legacy queue " "consumers retain the 2-tuple shape" ) @@ -69,7 +69,7 @@ def test_gateway_queue_item_carries_per_event_id_with_legacy_fallback(): put_def_idx = GATEWAY_CHAT_PY.find("def put_gateway_event(event, data):") assert put_def_idx != -1, "put_gateway_event(event, data) not found" put_body = GATEWAY_CHAT_PY[put_def_idx:put_def_idx + 1800] - assert 'queue_item = (event, data, event_id) if event_id and hasattr(q, "subscribe_with_snapshot") else (event, data)' in put_body, ( + assert 'queue_item = (event, data, event_id) if hasattr(q, "subscribe_with_snapshot") else (event, data)' in put_body, ( "Gateway live events must carry their own event_id for StreamChannel " "subscribers while preserving legacy queue compatibility" )