fix: make SSE replay cursors run-aware
This commit is contained in:
1 parent
d37e57f1ea
commit
5cb82f9da4
4 files changed
+62
-6
No files matched your search
@@ -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).
|
||||
|
||||
+22
-2
@@ -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
|
||||
|
||||
+10
-3
@@ -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;
|
||||
|
||||
@@ -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
|
||||
Reference in new issue
Block a user