Merge pull request #7700 from nesquena/stage/7291-journal-metering

fix(run-journal): keep metering live-only and preserve replay cursors (#7291)

CHANGELOG.md

@@ -5,6 +5,14 @@
 
 ### Fixed
 
+- **A long-running turn no longer replays a stale token count after a reload.** The run-journal
+  recorded live metering frames, so reattaching to a stream — or reloading a tab mid-turn —
+  could replay a snapshot from earlier in the same run and briefly show token/TPS figures that
+  had already been superseded. Metering is now live-only and never journaled, and replayed
+  frames no longer carry an event id that could advance the client's resume cursor past real
+  content. Reattach and reload now show the current numbers for the turn in progress. Thanks
+  @laitekin. (#7291)
+
 - **Steering a conversation works again after the context is compressed.** When compression
   rotated `agent.session_id`, a steer could no longer find the active worker: it was either
   silently dropped or accepted and never delivered. Steers now resolve the owning worker

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

@@ -10710,6 +10710,7 @@ from api.run_journal import (
     _parse_run_journal_event_id as _shared_parse_run_journal_event_id,
     bound_run_journal_snapshot_args,
     find_run_summary,
+    journal_replay_visible,
     read_run_events,
     read_session_run_events,
     session_journal_fingerprint,
@@ -18575,6 +18576,8 @@ def _replay_run_journal(
         max_seq=max_seq,
     )
     for entry in journal.get("events") or []:
+        if not journal_replay_visible(entry):
+            continue
         _sse_with_id(
             handler,
             entry.get("event") or entry.get("type") or "message",
@@ -19028,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
@@ -19110,6 +19113,8 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_id):
 
     def emit_replay(events, stream_id, cutoff_seq):
         for entry in events:
+            if not journal_replay_visible(entry):
+                continue
             event_id = str(entry.get("event_id") or "")
             event_seq = _run_journal_same_run_seq(event_id, stream_id)
             if cutoff_seq is not None and event_seq is not None and event_seq > cutoff_seq:
@@ -19176,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/run_journal.py

@@ -45,6 +45,14 @@ TERMINAL_SSE_EVENTS = frozenset({"done", "cancel", "apperror", "error", "stream_
 SSE_RELAY_CLOSE_EVENTS = frozenset({"stream_end", "cancel", "apperror", "error"})
 # Back-compat alias used by older call sites / tests.
 _TERMINAL_SSE_EVENTS = TERMINAL_SSE_EVENTS
+# Events that are live-UI-only telemetry with no recovery value in the run
+# journal. They are skipped at WRITE time (never durably journaled, so they
+# cannot bloat the journal on marathon runs) and filtered at REPLAY time (so
+# legacy journals that already contain a backlog never stream it to a
+# reconnecting browser tab). Readers deliberately do NOT filter: cursor math
+# (``cursor_event_missing`` bound) and the offline-gap coverage check count
+# journal seqs and must keep seeing every row.
+REPLAY_SKIPPED_SSE_EVENTS = frozenset({"metering"})
 _FSYNC_MODE_ENV = "HERMES_WEBUI_RUN_JOURNAL_FSYNC"
 _FSYNC_MODE_EAGER = "eager"
 _FSYNC_MODE_TERMINAL_ONLY = "terminal-only"
@@ -446,7 +454,16 @@ class RunJournalWriter:
         self._path = _run_path(self.session_id, self.run_id, session_dir=self.session_dir)
         self._lock = _lock_for(self._path)
 
-    def append_sse_event(self, event_name: str, payload=None) -> dict:
+    def append_sse_event(self, event_name: str, payload=None) -> dict | None:
+        # Live-UI-only telemetry (metering) has no recovery value in the journal:
+        # nothing reads those rows back for recovery, and journaling them at ~10 Hz
+        # on marathon runs balloons the durable file (12+ MB of a single 18 MB run
+        # was metering). Skip the write entirely and return None so callers'
+        # journal-id plumbing (``(journaled or {}).get("event_id")``) is untouched.
+        # Not reserving a seq keeps the remaining journaled seqs contiguous, which
+        # the offline-gap coverage and replay-cursor contiguity checks rely on.
+        if str(event_name or "").strip() in REPLAY_SKIPPED_SSE_EVENTS:
+            return None
         # Draw from the shared module-level seq cache under the per-path lock so
         # this writer and any direct append_run_event() call on the same path
         # agree on one monotonic, gapless sequence.
@@ -462,6 +479,22 @@ class RunJournalWriter:
         )
 
 
+def journal_replay_visible(event) -> bool:
+    """Return True when a journal row should be streamed to a reconnecting tab.
+
+    Live-UI-only telemetry rows (see ``REPLAY_SKIPPED_SSE_EVENTS``) carry no
+    recovery value — replaying a metering backlog only re-paints a stale TPS
+    number while multiplying the reconnect burst size. Writers no longer journal
+    them, but legacy journals may already contain them, so the replay emit sites
+    filter through this predicate. Non-dict rows are passed through (visible) so
+    an unexpected shape can never silently swallow user-visible output.
+    """
+    if not isinstance(event, dict):
+        return True
+    name = str(event.get("event") or event.get("type") or "")
+    return name not in REPLAY_SKIPPED_SSE_EVENTS
+
+
 def read_run_events(
     session_id: str,
     run_id: str,

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

docs/rfcs/session-sse-contract-v1.md

@@ -183,7 +183,7 @@ when integrating with the live chat SSE relay.
 | `cancel` | Run cancelled |
 | `done` | Turn finalized (session payload); title/`stream_end` may follow |
 | `stream_end` | SSE fence — close the client EventSource |
-| `metering` | Token/cost metering snapshot |
+| `metering` | Best-effort live token/cost metering snapshot; not journal-replayed |
 | `context_status` | Context window / usage status |
 | `goal` | Goal / plan card update |
 | `goal_continue` | Goal continuation signal |
@@ -228,6 +228,31 @@ A future implementation must replay from the run journal via the existing
 `_replay_run_journal()` path (in `api/routes.py`) and fall back to the
 snapshot mechanism when journal entries are unavailable for a given cursor.
 
+### Live-only metering (implemented replay behavior)
+
+`metering` is best-effort live telemetry, not durable recovery content.
+`RunJournalWriter.append_sse_event()` returns `None` for metering without
+writing a row or allocating a sequence number. Other journaled events retain
+contiguous sequence numbers. Local-agent and gateway producers carry an
+explicit per-frame event ID, including `None`, through `StreamChannel`;
+an unjournaled metering frame emits no new SSE `id:` and must not borrow the
+previous frame's journal ID. The last-ID fallback is limited to legacy
+two-element queue entries.
+
+Existing journals are not rewritten. Direct readers (`read_run_events()` and
+`read_session_run_events()`) retain legacy metering rows for cursor validation
+and gap/coverage checks. Both run-level and per-session journal replay emitters
+apply `journal_replay_visible()` to omit those rows from reconnect delivery.
+Consequently, replayed legacy event IDs can have gaps where metering was
+filtered; clients must not infer missing durable content from those gaps.
+Metering missed while disconnected is not recovered from the journal; fresh
+live updates remain available. This exception does not remove token, reasoning,
+tool, or terminal events from durable replay.
+
+Coverage: `tests/test_run_journal.py`, `tests/test_run_journal_routes.py`,
+`tests/test_issue4812_session_sse_stream.py`, and
+`tests/test_stage364_opus_live_sse_event_id.py`.
+
 ## Snapshot fallback
 
 When the `Last-Event-ID` cursor is evicted, expired, unknown, or refers to a

tests/test_issue4812_session_sse_stream.py

@@ -583,6 +583,64 @@ def test_session_route_live_delivery_skips_replayed_active_run_items(monkeypatch
     assert stream.unsubscribed is True
 
 
+def test_session_route_replay_skips_metering_rows(monkeypatch):
+    import api.routes as routes
+
+    class _FakeStream:
+        def __init__(self):
+            self.q = queue.Queue()
+            self.q.put_nowait(("stream_end", {"status": "done"}, "run_active:2"))
+            self.unsubscribed = False
+
+        def subscribe_with_snapshot(self):
+            return self.q, {"last_event_id": "run_active:1", "offline_buffered_events": 1}
+
+        def unsubscribe(self, q):
+            self.unsubscribed = q is self.q
+
+    stream = _FakeStream()
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True)
+    monkeypatch.setattr(
+        routes,
+        "get_session",
+        lambda sid, metadata_only=False: SimpleNamespace(
+            session_id=sid,
+            compact=lambda **_kwargs: {"session_id": sid, "title": "Session"},
+        ),
+    )
+    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})
+    # Both the pre-attach replay and the post-attach reconciliation draw from the
+    # same journal; a metering row among them must never reach the SSE body.
+    monkeypatch.setattr(
+        routes,
+        "read_session_run_events",
+        lambda *_args, **_kwargs: {
+            "status": "ok",
+            "events": [
+                {"run_id": "run_prev", "seq": 1, "event": "token", "payload": {"text": "replayed"}, "event_id": "run_prev:1"},
+                {"run_id": "run_prev", "seq": 2, "event": "metering", "payload": {"tps": 47.3}, "event_id": "run_prev:2"},
+            ],
+        },
+    )
+
+    handler = _FakeHandler()
+    routes._handle_session_sse_stream_for_session(
+        handler,
+        urlparse("/api/sessions/session_1/events?after_event_id=run_prev:0"),
+        "session_1",
+    )
+
+    body = handler.wfile.getvalue().decode("utf-8")
+    assert "id: run_prev:1\n" in body
+    assert "event: token\n" in body
+    assert "event: metering\n" not in body
+    assert "id: run_prev:2\n" not in body
+    assert stream.unsubscribed is True
+
+
 def test_session_route_breaks_on_terminal_even_when_reconciliation_replayed_it(monkeypatch):
     """CORE regression (#5677 gate r3): if reconciliation replays the active run's
     TERMINAL event at the subscribe cutoff, the live queue's copy is deduped — but

tests/test_run_journal.py

@@ -2,11 +2,14 @@ import json
 from pathlib import Path
 
 from api.run_journal import (
+    REPLAY_SKIPPED_SSE_EVENTS,
     RunJournalWriter,
     append_run_event,
     find_run_summary,
+    journal_replay_visible,
     latest_run_summary,
     read_run_events,
+    read_session_run_events,
     stale_interrupted_event,
 )
 
@@ -28,6 +31,58 @@ def test_run_journal_appends_monotonic_seq_and_reads_after_cursor(tmp_path):
     assert [event["event"] for event in journal["events"]] == ["done"]
 
 
+def test_writer_skips_metering_and_keeps_remaining_seqs_contiguous(tmp_path):
+    writer = RunJournalWriter("session_1", "run_1", session_dir=tmp_path)
+
+    first = writer.append_sse_event("token", {"text": "hello"})
+    skipped = writer.append_sse_event("metering", {"tps": 47.3})
+    second = writer.append_sse_event("done", {"session": {}})
+
+    assert skipped is None
+    assert first["seq"] == 1
+    # Skipped metering must NOT consume a seq, so visible rows stay contiguous —
+    # the offline-gap coverage and replay-cursor contiguity checks depend on it.
+    assert second["seq"] == 2
+
+    journal = read_run_events("session_1", "run_1", session_dir=tmp_path)
+    assert [event["event"] for event in journal["events"]] == ["token", "done"]
+
+
+def test_writer_skip_is_payload_independent(tmp_path):
+    writer = RunJournalWriter("session_1", "run_1", session_dir=tmp_path)
+
+    assert writer.append_sse_event("metering", None) is None
+    assert writer.append_sse_event("metering", {"tps": 1.0, "active": 1}) is None
+
+    journal = read_run_events("session_1", "run_1", session_dir=tmp_path)
+    assert journal["events"] == []
+
+
+def test_journal_replay_visible_predicate():
+    assert REPLAY_SKIPPED_SSE_EVENTS == {"metering"}
+    assert journal_replay_visible({"event": "token", "payload": {"text": "x"}}) is True
+    assert journal_replay_visible({"event": "metering", "payload": {"tps": 1.0}}) is False
+    # `type` fallback: metering rows are invisible under either field.
+    assert journal_replay_visible({"type": "metering"}) is False
+    # Unknown shapes must pass through (visible) so they can never swallow output.
+    assert journal_replay_visible("not-a-dict") is True
+
+
+def test_readers_keep_metering_rows(tmp_path):
+    # Writers skip metering, but readers deliberately do NOT filter: cursor math
+    # (the ``cursor_event_missing`` bound) and the offline-gap coverage check
+    # count journal seqs and must keep seeing every row.
+    append_run_event("session_1", "run_1", "token", {"text": "hello"}, session_dir=tmp_path)
+    append_run_event("session_1", "run_1", "metering", {"tps": 47.3}, session_dir=tmp_path)
+
+    journal = read_run_events("session_1", "run_1", session_dir=tmp_path)
+    assert [event["event"] for event in journal["events"]] == ["token", "metering"]
+
+    replay = read_session_run_events("session_1", after_event_id="run_1:1", session_dir=tmp_path)
+    assert replay["status"] == "ok"
+    assert [event["event"] for event in replay["events"]] == ["metering"]
+
+
 def test_run_journal_reads_bounded_replay_window(tmp_path):
     writer = RunJournalWriter("session_1", "run_1", session_dir=tmp_path)
 

tests/test_run_journal_routes.py

@@ -761,6 +761,97 @@ def test_replay_run_journal_honors_after_seq_cursor(monkeypatch):
     assert "event: done\n" in body
 
 
+def test_replay_run_journal_skips_metering_rows(monkeypatch):
+    import api.routes as routes
+
+    handler = SimpleNamespace(wfile=io.BytesIO())
+    monkeypatch.setattr(
+        routes,
+        "find_run_summary",
+        lambda stream_id: {
+            "session_id": "session_1",
+            "run_id": stream_id,
+            "terminal": True,
+        },
+    )
+    monkeypatch.setattr(
+        routes,
+        "read_run_events",
+        lambda session_id, run_id, after_seq=None, max_seq=None: {
+            "events": [
+                {"event": "token", "payload": {"text": "hello"}, "event_id": f"{run_id}:1"},
+                {"event": "metering", "payload": {"tps": 47.3}, "event_id": f"{run_id}:2"},
+                {"event": "done", "payload": {"session": {}}, "event_id": f"{run_id}:3"},
+            ]
+        },
+    )
+
+    assert routes._replay_run_journal(handler, "run_1", 0) is True
+    body = handler.wfile.getvalue().decode("utf-8")
+    assert "id: run_1:1\n" in body
+    assert "event: token\n" in body
+    assert "id: run_1:3\n" in body
+    assert "event: done\n" in body
+    # The metering row is filtered from the replay burst.
+    assert "event: metering\n" not in body
+    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"
     )