fix(run-journal): preserve cursor ids for unjournaled metering

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")

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 = (

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")

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
 

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
 
 
 

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"
     )