diff --git a/CHANGELOG.md b/CHANGELOG.md index 5c0d6b11..ea29cc24 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,14 @@ ## [Unreleased] +## [v0.51.170] — 2026-05-30 — Release EP (stage-batch52 — run-aware SSE replay cursors) + +### Fixed + +- SSE run-journal replay cursors are now run-aware: a stale `after_seq` from an interrupted prior stream can no longer suppress replay events in a newer stream whose sequence numbers reset from 1. The reconnect cursor now carries a run-scoped `after_event_id` (`run_id:seq`) and the server ignores it when the run id differs, falling back to same-run `after_seq` dedupe (#3124). + +## [v0.51.169] — 2026-05-30 — Release EO (stage-batch51 — skill-toggle profile scoping + update-tag filter + Docker docs) + ### Fixed - Docker docs now explain host-localhost URLs (`host.docker.internal` / `host.containers.internal`) and the `sudo docker compose` `$HOME=/root` bind-mount pitfall for users whose WebUI cannot reach host APIs or see `~/.hermes` (#3012, #3006). diff --git a/api/routes.py b/api/routes.py index 0ce2c25d..d448a382 100644 --- a/api/routes.py +++ b/api/routes.py @@ -7321,7 +7321,27 @@ def _sse_with_id(handler, event, data, event_id=None): _sse(handler, event, data) -def _parse_run_journal_after_seq(qs: dict) -> int | None: +def _parse_run_journal_event_id(raw: str | None) -> tuple[str | None, int | None]: + raw = str(raw or "").strip() + if not raw: + return None, None + if ":" in raw: + run_id, tail = raw.rsplit(":", 1) + else: + run_id, tail = None, raw + try: + seq = max(0, int(tail)) + except (TypeError, ValueError): + return run_id or None, None + return run_id or None, seq + + +def _parse_run_journal_after_seq(qs: dict, stream_id: str | None = None) -> int | None: + event_run_id, event_seq = _parse_run_journal_event_id(qs.get("after_event_id", [None])[0]) + if event_run_id: + if stream_id and event_run_id != stream_id: + return None + return event_seq raw = qs.get("after_seq", [None])[0] if raw in (None, ""): return None @@ -7376,7 +7396,7 @@ def _handle_sse_stream(handler, parsed): handler.send_header("Connection", "close") handler.end_headers() try: - _replay_run_journal(handler, stream_id, _parse_run_journal_after_seq(qs)) + _replay_run_journal(handler, stream_id, _parse_run_journal_after_seq(qs, stream_id)) except _CLIENT_DISCONNECT_ERRORS: pass return True diff --git a/static/messages.js b/static/messages.js index 5af08e2f..d9209cce 100644 --- a/static/messages.js +++ b/static/messages.js @@ -946,6 +946,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ let _streamFadeReduceMotion=false; let _streamFadeReduceMotionOnChange=null; let _lastRunJournalSeq=0; + let _lastRunJournalEventId=''; const _STREAM_FADE_MS=200; const _STREAM_FADE_MAX_MS=350; const _STREAM_FADE_STAGGER_MS=16; @@ -1462,7 +1463,10 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ if(!raw) return; const tail=raw.includes(':')?raw.slice(raw.lastIndexOf(':')+1):raw; const seq=Number.parseInt(tail,10); - if(Number.isFinite(seq)&&seq>_lastRunJournalSeq) _lastRunJournalSeq=seq; + if(Number.isFinite(seq)&&seq>_lastRunJournalSeq){ + _lastRunJournalSeq=seq; + _lastRunJournalEventId=raw; + } } function _runJournalReplayAfterSeq(){ return Math.max(0,_lastRunJournalSeq||0); @@ -1470,8 +1474,11 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ function _runJournalReplayParams(){ // `replay=1` documents frontend intent. The server selects replay when the // stream id no longer has a live worker; `after_seq` prevents duplicated - // journal events after this EventSource has already rendered part of a run. - return `&replay=1&after_seq=${encodeURIComponent(String(_runJournalReplayAfterSeq()))}`; + // journal events after this EventSource has already rendered part of the + // same run. `after_event_id` keeps that cursor run-aware so a stale cursor + // from an earlier interrupted stream cannot suppress a newer stream whose + // sequence numbers started over from 1. + return `&replay=1&after_seq=${encodeURIComponent(String(_runJournalReplayAfterSeq()))}&after_event_id=${encodeURIComponent(_lastRunJournalEventId||'')}`; } let _lastRenderMs=0; diff --git a/tests/test_run_journal_routes.py b/tests/test_run_journal_routes.py index 187541cf..860a48a9 100644 --- a/tests/test_run_journal_routes.py +++ b/tests/test_run_journal_routes.py @@ -24,7 +24,7 @@ def test_dead_stream_sse_replays_journal_before_404_fallback(): assert "find_run_summary(stream_id)" in block assert "stream not found" in block assert "_replay_run_journal" in block - assert "_parse_run_journal_after_seq" in block + assert "_parse_run_journal_after_seq(qs, stream_id)" in block assert 'Content-Type", "text/event-stream; charset=utf-8"' in block @@ -160,3 +160,24 @@ def test_replay_run_journal_honors_after_seq_cursor(monkeypatch): body = handler.wfile.getvalue().decode("utf-8") assert "id: run_1:4\n" in body assert "event: done\n" in body + + +def test_parse_run_journal_after_seq_is_run_aware(): + import api.routes as routes + + assert routes._parse_run_journal_after_seq( + {"after_event_id": ["new-run:7"], "after_seq": ["7"]}, + "new-run", + ) == 7 + assert routes._parse_run_journal_after_seq( + {"after_event_id": ["old-run:7"], "after_seq": ["7"]}, + "new-run", + ) is None + assert routes._parse_run_journal_after_seq({"after_seq": ["3"]}, "new-run") == 3 + + +def test_frontend_sends_run_aware_replay_cursor(): + messages_src = (ROOT / "static" / "messages.js").read_text(encoding="utf-8") + assert "let _lastRunJournalEventId=''" in messages_src + assert "_lastRunJournalEventId=raw" in messages_src + assert "after_event_id=${encodeURIComponent(_lastRunJournalEventId||'')}" in messages_src