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 1/5] 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 From 5cbc5f7106864a7c2d88275b2cd865369039e136 Mon Sep 17 00:00:00 2001 From: laitekin <2625707287@qq.com> Date: Tue, 25 Aug 2026 18:13:38 +0800 Subject: [PATCH 2/5] fix(run-journal): preserve cursor ids for unjournaled metering --- api/gateway_chat.py | 2 +- api/routes.py | 8 +-- api/streaming.py | 2 +- tests/test_run_journal_routes.py | 55 +++++++++++++++++++ tests/test_run_journal_streaming_static.py | 2 +- tests/test_stage364_opus_live_sse_event_id.py | 4 +- 6 files changed, 64 insertions(+), 9 deletions(-) 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" ) From 87338e8098611532fc3da6dde1f8a513b2ee1003 Mon Sep 17 00:00:00 2001 From: laitekin <2625707287@qq.com> Date: Mon, 21 Sep 2026 22:33:14 +0800 Subject: [PATCH 3/5] test(run-journal): bind replay fixture to live stream registry --- tests/test_issue4812_session_sse_stream.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py index dc3c2c3f..27f9c771 100644 --- a/tests/test_issue4812_session_sse_stream.py +++ b/tests/test_issue4812_session_sse_stream.py @@ -610,6 +610,8 @@ def test_session_route_replay_skips_metering_rows(monkeypatch): ) 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( From 9b3c749302e883fc603c9fbe0bbc8e7639f9eef8 Mon Sep 17 00:00:00 2001 From: laitekin <2625707287@qq.com> Date: Mon, 21 Sep 2026 23:03:24 +0800 Subject: [PATCH 4/5] docs(run-journal): specify live-only metering and optional writes (#7272) --- api/run_journal.py | 2 +- docs/rfcs/session-sse-contract-v1.md | 27 ++++++++++++++++++++++++++- 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/api/run_journal.py b/api/run_journal.py index 677876eb..a088c13d 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -454,7 +454,7 @@ 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 diff --git a/docs/rfcs/session-sse-contract-v1.md b/docs/rfcs/session-sse-contract-v1.md index 65b13366..e6248e03 100644 --- a/docs/rfcs/session-sse-contract-v1.md +++ b/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 From dd747b2f20f4131cefd0c37e413a915ed2e957c5 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Tue, 22 Sep 2026 04:07:06 +0000 Subject: [PATCH 5/5] docs(changelog): stamp #7291 run-journal metering fix --- CHANGELOG.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ac7e397..6e30cbcf 100644 --- a/CHANGELOG.md +++ b/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