fix(#4812): re-sync idle no-cursor SSE subscriber on run completing during keepalive wait

Fable gate found a seam Codex's 3 boundary findings didn't cover: an idle no-cursor
subscriber to /api/sessions/{id}/events misses a run that starts AND finishes inside a
single keepalive tick — the wait loop only looked for a live in-memory STREAMS entry to
attach, never noticing the journal advanced, so the run's transcript was silently missing
until manual refresh.

Add a cheap bounded session_journal_fingerprint() (file-count/max-mtime/total-size from
stat only, never parses bodies) and, inside the idle attach-wait loop, emit a session_snapshot
re-sync whenever the fingerprint advances with no live stream to attach — the same honest
recovery contract used for a failed reconciliation. Re-baselines after each re-sync so a quiet
idle connection never spams snapshots. 2 regression tests; 22/22 SSE tests green.

Co-authored-by: rodboev <rodboev@users.noreply.github.com>
This commit is contained in:
tandrodboev committed 2026-07-10 20:15:49 +00:00
1 parent 9600e1546a
commit 2829843a2a
4 files changed
+125

No files matched your search

+4
View File
@@ -3,6 +3,10 @@
## [Unreleased]
### Added
- **Resumable session event streams (server contract).** A new per-session SSE endpoint (`GET /api/sessions/{id}/events`) lets a client reconnect and resume from where its stream dropped — replaying journalled events after a cursor (standard EventSource `Last-Event-ID`, with an `after_event_id` query fallback for clients that can't set the header), deduped against live events at an atomic subscribe/snapshot boundary. Replay is bounded (per-line byte cap enforced pre-decode + an event-count limit) and *honest*: any limit, gap, or non-contiguous cursor returns a fresh `session_snapshot` recovery boundary so the client re-syncs rather than silently losing events. An idle no-cursor subscriber also re-syncs if a run starts and finishes entirely within one keepalive tick. Cross-profile access is refused (404) before any replay. This is the Phase-1 server contract; no client consumes it yet. Thanks @rodboev. (#5677, #4812)
### Fixed
- **Repeated automatic retries after your provider credentials run out no longer spam the chat with identical error bubbles.** When a session's provider credential pool is exhausted, each completed background task used to start a fresh automatic "wakeup" turn that failed the same way, stacking duplicate terminal error messages. The first failure still shows clearly, but subsequent automatic wakeups on the same model/provider are now paused (recorded as auditable suppression metadata, not shown), and the pause clears automatically on recovery — a successful run, a model/provider switch, or when the underlying credential state actually changes. Explicit (non-automatic) recovery turns are never blocked. Thanks @franksong2702. (#5732, #3929)
+15
View File
@@ -9259,6 +9259,7 @@ from api.run_journal import (
find_run_summary,
read_run_events,
read_session_run_events,
session_journal_fingerprint,
stale_interrupted_event,
)
from api.todo_state import attach_todo_state
@@ -16937,10 +16938,24 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_id):
if subscriber is None:
if replay_ok:
emit_replay(replay_events, active_stream_id, None)
# Baseline the journal so we can detect a run that starts AND finishes
# inside a single keepalive tick while we wait for a live stream. Such a
# run updates the journal but never materializes an in-memory STREAMS
# entry to attach, so without this an idle subscriber (esp. a no-cursor
# client) would silently miss that run's transcript until manual refresh.
_idle_journal_fp = session_journal_fingerprint(session_id)
while True:
subscriber, subscriber_stream, stream_snapshot, active_stream_id = attach_active_stream()
if subscriber is not None:
break
# Journal advanced with no live stream to attach → a run completed
# entirely within the wait. Re-sync via a snapshot boundary (the
# same honest-recovery contract used for a failed reconciliation),
# then re-baseline so we only re-sync on genuinely new advances.
_current_journal_fp = session_journal_fingerprint(session_id)
if _current_journal_fp != _idle_journal_fp:
_idle_journal_fp = _current_journal_fp
emit_session_snapshot(active_stream_id)
handler.wfile.write(b": keepalive\n\n")
handler.wfile.flush()
time.sleep(_SSE_HEARTBEAT_INTERVAL_SECONDS)
+34
View File
@@ -367,6 +367,40 @@ def latest_run_summary(session_id: str, run_id: str, *, session_dir: Path | None
return _summary_from_events(session_id, run_id, journal.get("events") or [])
def session_journal_fingerprint(session_id: str, *, session_dir: Path | None = None) -> tuple[int, float, int]:
"""Cheap, bounded fingerprint of a session's run journal: (file_count, max_mtime, total_size).
Reads only directory + per-file stat metadata (never parses journal bodies), so it stays
O(runs) and cannot be tipped over by a large ``done`` row. Used to detect that the journal
advanced during an idle live-subscribe wait — a run that starts AND finishes inside a single
keepalive tick leaves the journal changed but never materializes a live in-memory stream, so a
no-cursor idle subscriber would otherwise miss it until a manual refresh. Returns (0, 0.0, 0)
when the session has no journal yet. Invalid ids resolve to the empty fingerprint rather than
raising so callers can probe unconditionally.
"""
try:
sid = _validate_id(session_id, "session_id")
except ValueError:
return (0, 0.0, 0)
root = Path(session_dir) if session_dir is not None else _default_session_dir()
session_root = root / RUN_JOURNAL_DIR_NAME / sid
if not session_root.exists():
return (0, 0.0, 0)
count = 0
max_mtime = 0.0
total_size = 0
for path in session_root.glob("*.jsonl"):
try:
st = path.stat()
except OSError:
continue
count += 1
total_size += st.st_size
if st.st_mtime > max_mtime:
max_mtime = st.st_mtime
return (count, max_mtime, total_size)
def find_run_summary(run_id: str, *, session_dir: Path | None = None) -> dict | None:
rid = _validate_id(run_id, "run_id")
root = Path(session_dir) if session_dir is not None else _default_session_dir()
@@ -357,6 +357,78 @@ def test_session_route_emits_snapshot_without_id_for_missing_cursor_and_keepaliv
assert stop["count"] == 1
def test_session_route_resyncs_when_run_completes_inside_idle_wait(monkeypatch):
"""Idle no-cursor subscriber: a run that starts+finishes inside one keepalive
tick advances the journal with no live stream to attach. The handler must emit
a session_snapshot re-sync so the client catches that run instead of silently
missing it until manual refresh (Fable gate finding on #5677)."""
import api.routes as routes
stop = _stop_after_first_heartbeat(monkeypatch)
monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_a, **_k: True)
monkeypatch.setattr(
routes,
"get_session",
lambda sid, metadata_only=False: SimpleNamespace(
session_id=sid,
compact=lambda **_kwargs: {"session_id": sid, "title": "Session"},
),
)
# No live stream ever attaches (the run completed before we could subscribe).
monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_a, **_k: None)
# No cursor → no replay; the missing-cursor path returns ok/empty.
monkeypatch.setattr(routes, "read_session_run_events", lambda *_a, **_k: {"status": "ok", "events": []})
# Fingerprint advances between the pre-loop baseline and the in-loop check,
# simulating a run completing during the wait.
fps = iter([(0, 0.0, 0), (1, 123.0, 42)])
monkeypatch.setattr(routes, "session_journal_fingerprint", lambda *_a, **_k: next(fps, (1, 123.0, 42)))
handler = _FakeHandler()
routes._handle_session_sse_stream_for_session(
handler,
urlparse("/api/sessions/session_1/events"),
"session_1",
)
body = handler.wfile.getvalue().decode("utf-8")
assert "event: session_snapshot\n" in body, "idle journal advance must emit a snapshot re-sync"
assert ": keepalive\n\n" in body
assert stop["count"] == 1
def test_session_route_no_resync_when_idle_journal_unchanged(monkeypatch):
"""The idle-wait re-sync must fire ONLY on a genuine journal advance — a quiet
idle connection (fingerprint unchanged) must NOT spam snapshots every tick."""
import api.routes as routes
stop = _stop_after_first_heartbeat(monkeypatch)
monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_a, **_k: 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 *_a, **_k: None)
monkeypatch.setattr(routes, "read_session_run_events", lambda *_a, **_k: {"status": "ok", "events": []})
# Fingerprint never changes → no re-sync.
monkeypatch.setattr(routes, "session_journal_fingerprint", lambda *_a, **_k: (2, 99.0, 77))
handler = _FakeHandler()
routes._handle_session_sse_stream_for_session(
handler,
urlparse("/api/sessions/session_1/events"),
"session_1",
)
body = handler.wfile.getvalue().decode("utf-8")
assert "event: session_snapshot\n" not in body, "quiet idle wait must not emit a snapshot"
assert ": keepalive\n\n" in body
assert stop["count"] == 1
def test_session_route_blocks_hidden_sessions_before_replay_or_live_attach(monkeypatch):
import api.routes as routes