fix(run-journal): skip metering telemetry on write and replay (#7272)

Metering telemetry is journaled at ~10 Hz during active turns, so marathon
reasoning runs balloon the durable run journal (12.4 MB of a single 18 MB run
was metering). On a mid-run refresh the reconnect path replays the entire
journal as one unpaced SSE burst, OOM-crashing the browser tab in a
crash-refresh loop while the server stays healthy.

- Write time: RunJournalWriter.append_sse_event() skips metering rows via
  REPLAY_SKIPPED_SSE_EVENTS, returning None so caller journal-id plumbing is
  untouched and remaining journaled seqs stay contiguous.
- Replay time: _replay_run_journal() and emit_replay() filter metering rows
  through journal_replay_visible(), so legacy journals never stream the
  backlog either.
- Readers (read_run_events / read_session_run_events) deliberately keep every
  row: cursor math and the offline-gap coverage check count seqs.

Refs #7272

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",
@@ -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:

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"
@@ -447,6 +455,15 @@ class RunJournalWriter:
         self._lock = _lock_for(self._path)
 
     def append_sse_event(self, event_name: str, payload=None) -> dict:
+        # 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,

tests/test_issue4812_session_sse_stream.py

@@ -583,6 +583,62 @@ 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})
+    # 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,42 @@ 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_active_stream_replay_keeps_items_for_new_run_with_same_seq_range(monkeypatch):
     import api.routes as routes