diff --git a/CHANGELOG.md b/CHANGELOG.md index f772b933..e0b221c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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) diff --git a/api/routes.py b/api/routes.py index 0846dc5c..82deecb7 100644 --- a/api/routes.py +++ b/api/routes.py @@ -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) diff --git a/api/run_journal.py b/api/run_journal.py index 5e277d96..3749ab2c 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -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() diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py index 60dbec33..1118f65f 100644 --- a/tests/test_issue4812_session_sse_stream.py +++ b/tests/test_issue4812_session_sse_stream.py @@ -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