From bcafbd69655ff232f7d9d24267c1e32f9da243ac Mon Sep 17 00:00:00 2001 From: laitekin <2625707287@qq.com> Date: Tue, 25 Aug 2026 16:19:11 +0800 Subject: [PATCH] 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 | 5 ++ api/run_journal.py | 33 +++++++++++++ tests/test_issue4812_session_sse_stream.py | 56 ++++++++++++++++++++++ tests/test_run_journal.py | 55 +++++++++++++++++++++ tests/test_run_journal_routes.py | 36 ++++++++++++++ 5 files changed, 185 insertions(+) diff --git a/api/routes.py b/api/routes.py index 7bfa4347..062a3772 100644 --- a/api/routes.py +++ b/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: diff --git a/api/run_journal.py b/api/run_journal.py index ab8d0a7e..677876eb 100644 --- a/api/run_journal.py +++ b/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, diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py index 68970c4e..dc3c2c3f 100644 --- a/tests/test_issue4812_session_sse_stream.py +++ b/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 diff --git a/tests/test_run_journal.py b/tests/test_run_journal.py index cce7186f..49133434 100644 --- a/tests/test_run_journal.py +++ b/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) diff --git a/tests/test_run_journal_routes.py b/tests/test_run_journal_routes.py index a0bd82d3..95077d2d 100644 --- a/tests/test_run_journal_routes.py +++ b/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