"""Tests for the persistent per-session SSE channel (Option X). Verifies the SessionChannel registry + reaper + dual-emit + endpoint wiring that bridges the cross-turn bg_task_complete delivery gap (between agent turns STREAMS is torn down, so the session-scoped channel is the only live surface). Companion to t_98368bd0 implementation plan. Structural (source-grep) checks plus pure-function tests for the SessionChannel class and reaper logic. """ from __future__ import annotations import time from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parents[1] def _js_function_decl(src: str, name: str) -> str: marker = f"function {name}(" start = src.find(marker) assert start != -1, f"{name}() not found" brace = src.find("){", start) assert brace != -1, f"{name}() body not found" brace += 1 depth = 1 i = brace + 1 in_str = None escaped = False while i < len(src) and depth: ch = src[i] if in_str: if escaped: escaped = False elif ch == "\\": escaped = True elif ch == in_str: in_str = None elif ch in ('"', "'", "`"): in_str = ch elif ch == "{": depth += 1 elif ch == "}": depth -= 1 i += 1 assert depth == 0, f"{name}() body did not close" return src[start:i] # --------------------------------------------------------------------------- # Module surface # --------------------------------------------------------------------------- def test_background_process_exports_session_channel_api(): from api import background_process as bp for name in ( "SessionChannel", "SESSION_CHANNELS", "SESSION_CHANNELS_LOCK", "get_or_create_session_channel", "subscribe_to_session_channel", "get_session_channel", "start_session_channel_reaper", "stop_session_channel_reaper", ): assert hasattr(bp, name), f"missing: {name}" def test_config_exports_session_channel_ttl_constants(): from api import config as cfg assert isinstance(cfg.SESSION_CHANNEL_IDLE_TTL_SECS, int) assert cfg.SESSION_CHANNEL_IDLE_TTL_SECS == 14400 # 4 hours per spec assert isinstance(cfg.SESSION_CHANNEL_SUBSCRIBER_GRACE_SECS, int) assert cfg.SESSION_CHANNEL_SUBSCRIBER_GRACE_SECS == 60 # --------------------------------------------------------------------------- # SessionChannel: subscribe / emit / unsubscribe lifecycle # --------------------------------------------------------------------------- def test_session_channel_subscriber_lifecycle(): """subscribe → emit → unsubscribe with no leak.""" from api.background_process import SessionChannel ch = SessionChannel("sess-1") q1 = ch.subscribe() q2 = ch.subscribe() assert ch.subscriber_count() == 2 delivered = ch.emit("bg_task_complete", {"hello": 1}) assert delivered == 2 assert q1.get_nowait() == ("bg_task_complete", {"hello": 1}) assert q2.get_nowait() == ("bg_task_complete", {"hello": 1}) ch.unsubscribe(q1) assert ch.subscriber_count() == 1 # Re-emit goes to remaining sub only ch.emit("bg_task_complete", {"hello": 2}) assert q2.get_nowait() == ("bg_task_complete", {"hello": 2}) ch.unsubscribe(q2) assert ch.subscriber_count() == 0 def test_session_channel_emit_with_full_buffer_drops_silently(): """A slow tab whose queue is full doesn't block other subscribers.""" from api.background_process import SessionChannel ch = SessionChannel("sess-full") q_slow = ch.subscribe(maxsize=1) q_fast = ch.subscribe(maxsize=16) # Fill the slow queue q_slow.put_nowait(("filler", {})) delivered = ch.emit("bg_task_complete", {"x": 1}) # fast receives, slow dropped assert delivered == 1 assert q_fast.get_nowait() == ("bg_task_complete", {"x": 1}) # --------------------------------------------------------------------------- # Reaper: subscribers-empty grace + idle TTL cap # --------------------------------------------------------------------------- def test_session_channel_reaper_keeps_live_subscriber(): """A channel with at least one subscriber must NEVER be collected.""" from api.background_process import SessionChannel ch = SessionChannel("sess-live") ch.subscribe() # Far in the future, with live subscriber → not collected assert ch.reaper_should_collect(time.time() + 1_000_000) is False def test_session_channel_reaper_grace_period(): """No subscribers + grace expired → collect.""" from api.background_process import SessionChannel from api import config as cfg ch = SessionChannel("sess-grace") q = ch.subscribe() ch.unsubscribe(q) # Just dropped — within grace → keep assert ch.reaper_should_collect(time.time()) is False # Past grace → collect later = time.time() + cfg.SESSION_CHANNEL_SUBSCRIBER_GRACE_SECS + 1 assert ch.reaper_should_collect(later) is True def test_session_channel_reaper_idle_ttl_cap(): """Channel older than idle TTL with no subscribers → collect.""" from api.background_process import SessionChannel from api import config as cfg ch = SessionChannel("sess-zombie") # Force created_at far in the past, no subscriber drop tracked ch.created_at = time.time() - (cfg.SESSION_CHANNEL_IDLE_TTL_SECS + 100) ch.last_subscriber_drop_at = None assert ch.subscriber_count() == 0 assert ch.reaper_should_collect(time.time()) is True def test_session_channel_reaper_idle_ttl_held_by_live_subscriber(): """Even past idle TTL, a live subscriber keeps the channel alive.""" from api.background_process import SessionChannel from api import config as cfg ch = SessionChannel("sess-zombie-live") ch.created_at = time.time() - (cfg.SESSION_CHANNEL_IDLE_TTL_SECS + 100) ch.subscribe() # someone IS listening assert ch.reaper_should_collect(time.time()) is False def test_reaper_collects_via_registry(): """The reaper loop iterates SESSION_CHANNELS and removes collected entries.""" from api import background_process as bp, config as cfg sid = "sess-reaper-integration" ch = bp.get_or_create_session_channel(sid) q = ch.subscribe() ch.unsubscribe(q) # Push the drop time far enough back to trigger grace collection ch.last_subscriber_drop_at = time.time() - (cfg.SESSION_CHANNEL_SUBSCRIBER_GRACE_SECS + 5) # Drive one iteration of the reaper's body directly now = time.time() with bp.SESSION_CHANNELS_LOCK: for k, channel in list(bp.SESSION_CHANNELS.items()): if channel.reaper_should_collect(now): bp.SESSION_CHANNELS.pop(k, None) assert bp.get_session_channel(sid) is None def test_subscribe_to_session_channel_is_atomic_get_create_subscribe(): """The atomic helper returns a registered channel with the subscriber already attached, in one SESSION_CHANNELS_LOCK critical section. First call creates; second call reuses the same instance. """ from api import background_process as bp sid = "sess-atomic-subscribe" with bp.SESSION_CHANNELS_LOCK: bp.SESSION_CHANNELS.pop(sid, None) try: ch, q = bp.subscribe_to_session_channel(sid) # Channel is registered and the slot is already counted. assert bp.get_session_channel(sid) is ch assert ch.subscriber_count() == 1 # An emit reaches our queue immediately — proves we're on the live channel. ch.emit("bg_task_complete", {"n": 1}) assert q.get_nowait() == ("bg_task_complete", {"n": 1}) # Second call reuses the same instance and adds a second subscriber. ch2, q2 = bp.subscribe_to_session_channel(sid) assert ch2 is ch assert ch.subscriber_count() == 2 ch.unsubscribe(q2) finally: with bp.SESSION_CHANNELS_LOCK: bp.SESSION_CHANNELS.pop(sid, None) def test_subscribe_to_session_channel_survives_concurrent_reaper(): """Regression for PR #2971 Greptile P1 (background_process.py:215). Reproduces the reaper TOCTOU: an idle, past-grace channel exists in the registry; a subscriber arrives while the reaper sweeps concurrently. With the old split ``get_or_create_session_channel()`` + ``ch.subscribe()`` the reaper could collect the channel in the gap, orphaning the subscriber so later emits never reach its queue. The atomic helper holds SESSION_CHANNELS_LOCK across both steps, so the post-subscribe registry entry must be the EXACT channel the subscriber is attached to, and a subsequent emit must be delivered. """ import threading from api import background_process as bp, config as cfg sid = "sess-reaper-toctou" # Seed an idle channel that is already eligible for collection (no subs, # drop time pushed well past the grace window) — the dangerous precondition. with bp.SESSION_CHANNELS_LOCK: bp.SESSION_CHANNELS.pop(sid, None) seed = bp.SessionChannel(sid) seed.last_subscriber_drop_at = ( time.time() - (cfg.SESSION_CHANNEL_SUBSCRIBER_GRACE_SECS + 5) ) bp.SESSION_CHANNELS[sid] = seed stop = threading.Event() def _reaper_spin(): # Hammer the exact reaper critical section concurrently. while not stop.is_set(): now = time.time() with bp.SESSION_CHANNELS_LOCK: for k, channel in list(bp.SESSION_CHANNELS.items()): if channel.reaper_should_collect(now): bp.SESSION_CHANNELS.pop(k, None) t = threading.Thread(target=_reaper_spin, daemon=True) t.start() try: for _ in range(200): ch, q = bp.subscribe_to_session_channel(sid) # Post-condition: the channel we're subscribed to is the one in the # registry (atomicity guarantee). The reaper cannot have evicted it # between create and subscribe because both ran under the lock and # the channel now has a live subscriber. assert bp.get_session_channel(sid) is ch assert ch.subscriber_count() >= 1 # And an emit on the registry-resolved channel reaches our queue — # i.e. we are NOT orphaned on a collected channel. resolved = bp.get_session_channel(sid) resolved.emit("bg_task_complete", {"ping": 1}) assert q.get(timeout=1.0) == ("bg_task_complete", {"ping": 1}) ch.unsubscribe(q) finally: stop.set() t.join(timeout=2.0) with bp.SESSION_CHANNELS_LOCK: bp.SESSION_CHANNELS.pop(sid, None) # --------------------------------------------------------------------------- # Dual emit: STREAMS empty + SESSION_CHANNELS has subscriber → delivered # --------------------------------------------------------------------------- def test_emit_when_idle_between_turns_session_channel_delivers(): """No STREAMS but a SESSION_CHANNELS subscriber → event reaches the channel.""" from api import background_process as bp, config as cfg sid = "sess-between-turns" ch = bp.get_or_create_session_channel(sid) q = ch.subscribe() try: # Make sure STREAMS has no relevant entry with cfg.STREAMS_LOCK: assert all( (cfg.ACTIVE_RUNS.get(stream_id) or {}).get("session_id") != sid for stream_id in cfg.STREAMS ) evt = { "type": "completion", "session_id": "proc-99", "session_key": sid, "command": "sleep 1", "exit_code": 0, "output": "ok", } bp.register_process_session(sid, sid) try: bp._process_one(evt) event_name, data = q.get(timeout=2.0) assert event_name == "bg_task_complete" assert data["session_id"] == sid assert data["task_id"] == "proc-99" assert data.get("event_id"), "emitter must stamp event_id (Q4 contract)" finally: bp.unregister_process_session(sid) cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid) cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None) finally: ch.unsubscribe(q) with bp.SESSION_CHANNELS_LOCK: bp.SESSION_CHANNELS.pop(sid, None) def test_emit_during_busy_turn_dual_emits_to_both(): """STREAMS + SESSION_CHANNELS both subscribed → both receive.""" from api import background_process as bp, config as cfg sid = "sess-busy-dual" stream_id = "stream-busy-dual" streams_received: list = [] class _FakeStreamChannel: def put_nowait(self, item): streams_received.append(item) with cfg.STREAMS_LOCK: cfg.STREAMS[stream_id] = _FakeStreamChannel() cfg.ACTIVE_RUNS[stream_id] = {"session_id": sid} ch = bp.get_or_create_session_channel(sid) q = ch.subscribe() bp.register_process_session(sid, sid) try: evt = { "type": "completion", "session_id": "proc-busy-1", "session_key": sid, "command": "sleep 1", "exit_code": 0, "output": "ok", } bp._process_one(evt) # Both surfaces received the event (frontend will dedupe by process_id) assert streams_received, "STREAMS subscriber must receive in-turn delivery" event_name, data = streams_received[0] assert event_name == "bg_task_complete" assert data["task_id"] == "proc-busy-1" assert data.get("event_id"), "emitter must stamp event_id (Q4 contract)" ev2, data2 = q.get(timeout=2.0) assert ev2 == "bg_task_complete" assert data2["task_id"] == "proc-busy-1" assert data2.get("event_id"), "emitter must stamp event_id (Q4 contract)" finally: ch.unsubscribe(q) with bp.SESSION_CHANNELS_LOCK: bp.SESSION_CHANNELS.pop(sid, None) with cfg.STREAMS_LOCK: cfg.STREAMS.pop(stream_id, None) cfg.ACTIVE_RUNS.pop(stream_id, None) bp.unregister_process_session(sid) cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid) cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None) # --------------------------------------------------------------------------- # Route + frontend wiring (source-grep) # --------------------------------------------------------------------------- def test_routes_registers_session_stream_endpoint(): src = (REPO_ROOT / "api" / "routes.py").read_text() assert "/api/session/stream" in src assert "_handle_session_sse_stream" in src def test_routes_session_sse_uses_session_channel_subscribe(): src = (REPO_ROOT / "api" / "routes.py").read_text() # The handler must use the atomic get-or-create+subscribe helper (closes # the PR #2971 reaper TOCTOU race) and release the slot on every exit path. assert "subscribe_to_session_channel" in src assert "ch.unsubscribe(q)" in src # The split get-then-subscribe call pair must NOT come back — it reopens # the race the atomic helper exists to close. assert "ch = get_or_create_session_channel(sid)" not in src def test_server_starts_session_channel_reaper(): src = (REPO_ROOT / "server.py").read_text() assert "start_session_channel_reaper" in src assert "stop_session_channel_reaper" in src def test_frontend_opens_session_stream(): js = (REPO_ROOT / "static" / "messages.js").read_text() assert "api/session/stream?session_id=" in js assert "startSessionStream" in js assert "stopSessionStream" in js def test_session_stream_pauses_while_chat_stream_is_active(): """The active turn's chat SSE already carries live events for that session. Keeping /api/session/stream open for the same sid at the same time burns a Chrome same-origin connection slot and can starve ordinary /api/session fetches on HTTP/1.1. """ js = (REPO_ROOT / "static" / "messages.js").read_text() assert "function _chatStreamActiveForSession(sid)" in js assert "function _suspendSessionStreamForLiveChat(sid)" in js assert "function _resumeSessionStreamAfterLiveChat(sid)" in js start_src = _js_function_decl(js, "startSessionStream") assert "if (_chatStreamActiveForSession(sid))" in start_src assert "_sessionStreamHiddenSid = sid;" in start_src assert "return;" in start_src assert start_src.index("if (_chatStreamActiveForSession(sid))") < start_src.index("new EventSource(") attach_ix = js.index("function attachLiveStream") attach_src = js[attach_ix:js.index("function transcript()", attach_ix)] assert "_suspendSessionStreamForLiveChat(activeSid);" in attach_src assert attach_src.index("_suspendSessionStreamForLiveChat(activeSid);") < attach_src.index("new EventSource(") resume_src = _js_function_decl(js, "_resumeSessionStreamAfterLiveChat") assert "S.session.session_id !== sid" in resume_src assert "if (_chatStreamActiveForSession(sid)) return;" in resume_src assert "_sessionStreamHiddenSid = null;" in resume_src assert "startSessionStream(sid);" in resume_src suspend_src = _js_function_decl(js, "_suspendSessionStreamForLiveChat") assert "if (_sessionStreamSessionId !== sid) return;" in suspend_src assert "stopSessionStream();" in suspend_src def test_session_stream_resume_rearms_when_live_stream_registry_clears(): """The done event can arrive before stream_end closes /api/chat/stream. If the first resume attempt sees LIVE_STREAMS[sid] still present, the stream_end teardown must re-attempt resume after deleting that owner entry. """ js = (REPO_ROOT / "static" / "messages.js").read_text() close_src = _js_function_decl(js, "closeLiveStream") delete_idx = close_src.index("delete LIVE_STREAMS[sessionId];") resume_idx = close_src.index("_resumeSessionStreamAfterLiveChat(sessionId);") assert delete_idx < resume_idx, ( "closeLiveStream must retry session-stream resume after LIVE_STREAMS is " "cleared, otherwise done-before-stream_end can leave /api/session/stream closed" ) resume_src = _js_function_decl(js, "_resumeSessionStreamAfterLiveChat") assert "S.session.session_id !== sid" in resume_src, ( "the closeLiveStream retry relies on the visible-session guard to avoid " "reopening a session stream for a background session-switch teardown" ) assert "if (_chatStreamActiveForSession(sid)) return;" in resume_src def test_frontend_busy_race_gate_obsoleted_by_option_z_pivot(): """Per the Option Z PIVOT note baked into the handler body, the browser is no longer in the wakeup path at all — the server-side drain owns starting the next turn. The original ``if (S.busy)`` busy-race gate inside the handler was paired with the now-removed re-POST of ``/api/chat/stream``; once the re-POST went away (Option Z), the gate became moot. We assert the pivot documentation is in place so a future refactor doesn't silently re-introduce the gate without re-introducing the re-POST as well.""" js = (REPO_ROOT / "static" / "messages.js").read_text() fn_ix = js.index("function _handleBgTaskCompleteEvent") fn_src = js[fn_ix:fn_ix + 2400] assert "Option Z PIVOT" in fn_src assert "drain thread" in fn_src # The legacy re-POST and busy-race gate must NOT be inside the handler. assert "if (S.busy)" not in fn_src assert "/api/chat/stream" not in fn_src def test_frontend_shared_handler_dedupes_across_paths(): """Module-scope dedupe ring buffer (Map+TTL keyed (sid, event_id)) is what makes dual-emit safe.""" js = (REPO_ROOT / "static" / "messages.js").read_text() # Module-scope Map+TTL declaration outside any `function () { ... }` body assert "const _bgTaskCompleteSeenIds = new Map();" in js assert "const _BG_TASK_COMPLETE_TTL_MS = 60000;" in js assert "const _BG_TASK_COMPLETE_CAP = 256;" in js assert "_bgTaskCompleteRingBufferAdd" in js assert "_handleBgTaskCompleteEvent" in js # The handler must require event_id (server contract surface per Q4). assert "if (!evt_id) return;" in js or "if(!evt_id) return;" in js def test_sessions_js_starts_and_stops_session_stream_on_mount_unmount(): js = (REPO_ROOT / "static" / "sessions.js").read_text() assert "startSessionStream(S.session.session_id)" in js assert "stopSessionStream" in js # --------------------------------------------------------------------------- # Regression: notify_on_complete event shape carries NO session_key # --------------------------------------------------------------------------- # # Root cause of "zero ack POSTs ever" (t_0f447014): # tools.process_registry.ProcessRegistry._move_to_finished() enqueues the # completion event for a notify_on_complete process WITHOUT a "session_key" # field (only the watch_match enqueue includes one). _process_one then does # `session_key = evt.get("session_key") or process_id`, so it falls back to # the process id ("proc_xxxx"), which is NEVER a key in # PROCESS_SESSION_INDEX (only webui_session_id -> webui_session_id is # registered). The lookup misses → silent debug drop → no SSE emit ever. # # The existing happy-path test (test_emit_when_idle_between_turns_...) masks # this because it hand-builds evt WITH "session_key": sid — a shape the real # completion enqueue never produces. This test drives the event through the # REAL process_registry.completion_queue using the EXACT dict shape # _move_to_finished() produces, so it fails on the unfixed code and passes # after the fix. def test_real_completion_event_shape_routes_to_session_channel(): """A completion event in the real _move_to_finished() shape (no session_key) must still route to the registered WebUI session. Faithfully reproduces production: the terminal tool spawns a background process with session_key == webui_session_id (captured synchronously at spawn while the turn env is live). On exit, ProcessRegistry._move_to_ finished() (1) moves the ProcessSession into _finished — which retains session_key — then (2) enqueues a completion event that DOES NOT carry a "session_key" field. The drain (_process_one) must recover the session_key from the still-tracked ProcessSession in the registry. """ import time as _t from api import background_process as bp, config as cfg pytest.importorskip("tools.process_registry", reason="hermes-agent not installed") from tools.process_registry import process_registry, ProcessSession webui_sid = "sess-real-completion-shape" proc_id = "proc_realshape0001" ch = bp.get_or_create_session_channel(webui_sid) q = ch.subscribe() # streaming.py binds key == webui_session_id (register_process_session # called with (session_id, session_id)). HERMES_SESSION_KEY for the # spawned child therefore equals webui_sid, and the terminal tool stamps # that onto ProcessSession.session_key at spawn time. bp.register_process_session(webui_sid, webui_sid) # Simulate the finished process the registry retains in _finished after # _move_to_finished(): it carries the spawn-time session_key. finished = ProcessSession( id=proc_id, command="pytest -q", session_key=webui_sid, started_at=_t.time(), exited=True, exit_code=0, notify_on_complete=True, ) with process_registry._lock: process_registry._finished[proc_id] = finished # Build the event EXACTLY as ProcessRegistry._move_to_finished() enqueues # it for notify_on_complete: type/session_id/command/exit_code/output. # Crucially: NO "session_key" key. This is the real wire shape that the # existing happy-path test (test_emit_when_idle_between_turns_...) never # exercises because it hand-injects session_key. evt = { "type": "completion", "session_id": proc_id, "command": "pytest -q", "exit_code": 0, "output": "1197 passed", } process_registry.completion_queue.put(evt) drained = process_registry.completion_queue.get(timeout=2.0) try: bp._process_one(drained) event_name, data = q.get(timeout=2.0) assert event_name == "bg_task_complete" assert data["session_id"] == webui_sid assert data["task_id"] == proc_id assert data.get("event_id"), "emitter must stamp event_id (Q4 contract)" # Per the Q1 minimal-payload trim settled in #2242, the SSE payload # no longer carries ``wakeup_prompt`` (or ``command`` / ``exit_code``). # The optional ``summary`` field is now the only human-readable # surface; when the synthetic wakeup body is available the emitter # derives a short first-line summary from it. assert "wakeup_prompt" not in data assert "command" not in data assert "exit_code" not in data summary = data.get("summary") if summary is not None: assert isinstance(summary, str) assert "IMPORTANT" in summary or "Background process" in summary finally: ch.unsubscribe(q) with bp.SESSION_CHANNELS_LOCK: bp.SESSION_CHANNELS.pop(webui_sid, None) bp.unregister_process_session(webui_sid) with process_registry._lock: process_registry._finished.pop(proc_id, None) cfg.PENDING_BG_TASK_COMPLETIONS.discard(webui_sid) cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(webui_sid, None) # =========================================================================== # Option Z (PIVOT): server-side wakeup is the PRIMARY mechanism. # # The drain thread starts the agent turn directly server-side # (api/background_process._start_server_side_wakeup_turn → # api.routes.start_session_turn) with NO browser round-trip. The per-session # SSE channel is demoted to a pure live-view layer. These tests prove: # 1. closed-tab (no SSE subscriber at all) STILL starts a server-side turn # 2. active-turn defers (no double-start; PR #2279 next-turn drain handles it) # 3. one wakeup per process_id (dedupe) # 4. open tab still sees the live SSE frame (live-view unchanged) # =========================================================================== def _wait_for_wakeup(holder, timeout=3.0): """Thin wrapper preserving the legacy local name; the body lives in ``tests/_wakeup_helpers.py`` and is shared with ``test_wakeup_defer_race.py`` (Copilot PR #2971 r3305700944). """ from tests._wakeup_helpers import wait_for_wakeup as _impl return _impl(holder, timeout=timeout) def _install_fake_start_session_turn(monkeypatch, *, status=200): """Thin wrapper preserving the legacy local name; the body lives in ``tests/_wakeup_helpers.py`` and is shared with ``test_wakeup_defer_race.py`` (Copilot PR #2971 r3305700944). """ from tests._wakeup_helpers import install_fake_start_session_turn as _impl return _impl(monkeypatch, status=status) def test_server_side_wakeup_when_idle_no_tab(monkeypatch): """THE headline test: no SSE subscriber at all (closed tab / never opened). Pushing a completion must still start a server-side turn for the session. This is the closed-tab case browser-mediated wakeup could never serve. """ from api import background_process as bp, config as cfg sid = "sess-optz-idle-notab" proc_id = "proc-optz-idle-1" holder = _install_fake_start_session_turn(monkeypatch) bp.register_process_session(sid, sid) try: # Deliberately NO ch.subscribe() and NO STREAMS entry: nobody is # listening. ACTIVE_RUNS has no row for sid → session is idle. assert bp.get_session_channel(sid) is None assert bp._session_has_active_turn(sid) is False evt = { "type": "completion", "session_id": proc_id, "session_key": sid, "command": "sleep 8", "exit_code": 0, "output": "done", } bp._process_one(evt) assert _wait_for_wakeup(holder), ( "server-side wakeup turn was NOT started for an idle session with " "no tab — closed-tab case is broken" ) assert len(holder["calls"]) == 1 call = holder["calls"][0] assert call["session_id"] == sid assert call["source"] == "process_wakeup" assert call["message"].startswith("[IMPORTANT: Background process") finally: bp.unregister_process_session(sid) cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid) cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None) def test_server_side_wakeup_deferred_when_turn_active(monkeypatch): """A foreground turn is active (ACTIVE_RUNS has a row for the session) → the drain must NOT start a second turn. The PENDING_BG_TASK_COMPLETIONS marker is left for PR #2279's next-turn drain. """ from api import background_process as bp, config as cfg sid = "sess-optz-active-defer" proc_id = "proc-optz-active-1" stream_id = "stream-optz-active-1" holder = _install_fake_start_session_turn(monkeypatch) bp.register_process_session(sid, sid) cfg.ACTIVE_RUNS[stream_id] = {"session_id": sid} try: assert bp._session_has_active_turn(sid) is True evt = { "type": "completion", "session_id": proc_id, "session_key": sid, "command": "sleep 8", "exit_code": 0, "output": "done", } bp._process_one(evt) # Give any (incorrectly spawned) runner thread a chance to fire. fired = holder["event"].wait(timeout=1.0) assert fired is False, ( "server-side wakeup must DEFER when a turn is active — it " "double-started a turn" ) assert holder["calls"] == [] # Marker must remain so the next-turn drain delivers it. assert sid in cfg.PENDING_BG_TASK_COMPLETIONS finally: cfg.ACTIVE_RUNS.pop(stream_id, None) bp.unregister_process_session(sid) cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid) cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None) def test_wakeup_dedupe_once_per_process(monkeypatch): """The same process_id delivered twice (kill_process racing the reader thread) must wake the agent at most once. """ from api import background_process as bp, config as cfg sid = "sess-optz-dedupe" proc_id = "proc-optz-dedupe-1" holder = _install_fake_start_session_turn(monkeypatch) bp.register_process_session(sid, sid) try: evt = { "type": "completion", "session_id": proc_id, "session_key": sid, "command": "sleep 8", "exit_code": 0, "output": "done", } bp._process_one(evt) assert _wait_for_wakeup(holder) # Second delivery of the SAME process_id — must be deduped before the # server-side wakeup branch. bp._process_one(dict(evt)) time.sleep(0.5) assert len(holder["calls"]) == 1, ( "duplicate completion for the same process_id woke the agent twice" ) finally: bp.unregister_process_session(sid) cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid) cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None) def test_open_tab_sees_live_stream(monkeypatch): """Live-view still works: with a subscribed per-session SSE channel, the bg_task_complete frame is still delivered to the tab (so the open tab can render the server-initiated turn live). Server-side wakeup is additive — it does not remove the SSE emit. """ from api import background_process as bp, config as cfg sid = "sess-optz-liveview" proc_id = "proc-optz-liveview-1" holder = _install_fake_start_session_turn(monkeypatch) ch = bp.get_or_create_session_channel(sid) q = ch.subscribe() bp.register_process_session(sid, sid) try: evt = { "type": "completion", "session_id": proc_id, "session_key": sid, "command": "sleep 8", "exit_code": 0, "output": "done", } bp._process_one(evt) # Live-view: the open tab still receives the SSE frame. event_name, data = q.get(timeout=2.0) assert event_name == "bg_task_complete" assert data["session_id"] == sid assert data["task_id"] == proc_id assert data.get("event_id"), "emitter must stamp event_id (Q4 contract)" # AND the server-side wakeup still started (no active turn here). assert _wait_for_wakeup(holder) assert holder["calls"][0]["session_id"] == sid finally: ch.unsubscribe(q) with bp.SESSION_CHANNELS_LOCK: bp.SESSION_CHANNELS.pop(sid, None) bp.unregister_process_session(sid) cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid) cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None) # --------------------------------------------------------------------------- # event_id contract surface — backend emitter must stamp every event # --------------------------------------------------------------------------- def test_backend_emitter_stamps_event_id_on_every_bg_task_complete(): """Per the #2242 Q4 reply: every bg_task_complete emit carries an event_id; the consumer's ring-buffer dedupe is keyed on it. Source-grep the payload builder to confirm event_id is stamped.""" src = (REPO_ROOT / "api" / "background_process.py").read_text() # Locate the canonical payload builder and confirm event_id is in the dict. fn_ix = src.index("def _build_payload") fn_src = src[fn_ix:fn_ix + 4000] assert '"event_id"' in fn_src or "'event_id'" in fn_src, ( "payload builder must stamp event_id on every bg_task_complete payload" ) assert "uuid.uuid4().hex" in fn_src, ( "event_id should be a per-emit uuid hex (R2 §Q1)" ) # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Regression (Greptile P1 r3371970195): a CLOSED (readyState===2) session # EventSource must trigger a FRESH reconnect, not silently die. # --------------------------------------------------------------------------- # # Bug: startSessionStream's top guard is # `if (_sessionStreamSessionId === sid && _sessionEventSource) return;` # When onerror fired with readyState === 2 (permanent close: server 4xx/204, # browser retry-exhaustion), the old code scheduled the reconnect timer WITHOUT # clearing _sessionEventSource. The still-non-null (but CLOSED) object made the # guard short-circuit, so stopSessionStream() was never reached and no new # EventSource was ever created — the session stream stayed dead until the user # navigated away and back, dropping all bg_task_complete notifications meanwhile. # # Fix: in onerror, when es.readyState === 2 and es is still the active source, # close it and null _sessionEventSource BEFORE arming the reconnect timer, so # the deferred startSessionStream() passes its guard, runs stopSessionStream(), # and builds a fresh EventSource. The `_sessionEventSource === es` identity # check prevents a stale onerror from a superseded stream stomping a newer live # connection. def test_session_stream_onerror_clears_closed_source_so_reconnect_proceeds(): js = (REPO_ROOT / "static" / "messages.js").read_text() # Isolate the onerror handler body within startSessionStream. fn_ix = js.index("function startSessionStream") err_ix = js.index("es.onerror", fn_ix) onerror_src = js[err_ix:err_ix + 1600] # Must only act on the permanently-CLOSED state. assert "es.readyState === 2" in onerror_src # Identity-guard so a stale onerror can't stomp a newer live connection. assert "_sessionEventSource === es" in onerror_src # Must drop the dead reference (and close it) BEFORE arming the timer so # startSessionStream's "already connected" guard no longer short-circuits. assert "_sessionEventSource = null;" in onerror_src assert "es.close()" in onerror_src # Ordering: the null-out must precede the setTimeout that re-opens. null_pos = onerror_src.index("_sessionEventSource = null;") timer_pos = onerror_src.index("setTimeout(") assert null_pos < timer_pos, ( "must null the closed EventSource BEFORE arming the reconnect timer, " "else startSessionStream's guard short-circuits and the stream stays dead" ) # --------------------------------------------------------------------------- # Regression (Greptile P1 r3377162160): session stream must NOT stay dead # after a failed / early-returned loadSession (sessions.js:754 re-arm). # --------------------------------------------------------------------------- # # Bug: loadSession() tears down the live per-session SSE unconditionally at the # top — `if (typeof stopSessionStream==='function') stopSessionStream();` — but # only RE-arms it on the success path (`startSessionStream(S.session.session_id)` # near the end). Every early-return exit leaves the session the user is still # viewing with _sessionEventSource === null and no path back to a live stream: # - fetch error (network / non-404) → returns after stopSessionStream # - api() returned undefined (401 redirect) → returns # - stale-load race (_loadingSessionId !== sid after data) → returns # - same-session no-op guard (currentSid===sid && !forceReload) → returns # BEFORE the teardown, but a *prior* failed load already nulled the source, # and re-selecting the same session would otherwise no-op forever. # Net effect: bg_task_complete delivery silently dies until a full page reload. # # Fix: an idempotent helper `_rearmActiveSessionStream()` calls # startSessionStream(S.session.session_id) for whatever session is actually on # screen, invoked on the early-return paths AND the same-session no-op guard. # startSessionStream() is idempotent (its top guard # `_sessionStreamSessionId === sid && _sessionEventSource` no-ops when already # live) so the success path is never double-armed. The fetch-error path keeps # its own pre-existing guarded restart (`_selfHealedCurrent` check) instead of # the helper, because only there can the current session have just self-healed # away — re-arming a 404'd/deleted session_id would spin the SSE reconnect loop # against a dead session. Mirrors the #2979 messages.js reconnect fix. def test_load_session_rearms_stream_on_every_early_return(): js = (REPO_ROOT / "static" / "sessions.js").read_text() # The idempotent re-arm helper must exist and arm the on-screen session. assert "function _rearmActiveSessionStream(" in js, ( "expected a dedicated idempotent re-arm helper" ) helper_ix = js.index("function _rearmActiveSessionStream(") helper_src = js[helper_ix:helper_ix + 400] assert "S.session" in helper_src and "startSessionStream(" in helper_src, ( "helper must (re)arm startSessionStream for the currently-shown S.session" ) # Isolate the loadSession body. Widened window: the #4946 visit-ack helpers # added inside loadSession pushed the fetch-error catch's stream restart past # the old 14000-char cutoff. fn_ix = js.index("async function loadSession(") body = js[fn_ix:fn_ix + 16000] # The unconditional teardown must still be there (this is what creates the # dead-stream window the re-arm closes). assert "stopSessionStream()" in body # Post-teardown early-return paths must re-arm. The helper covers the # same-session guard, the undefined-data (401) exit, and the stale-response # exit — 3 helper call sites is the floor. (The fetch-error path uses its # own `_selfHealedCurrent`-guarded restart, asserted separately below; the # rapid-switch post-draft handoff is owned by the newer load's own arming.) assert js.count("_rearmActiveSessionStream()") >= 3, ( "each failed/early-return loadSession exit after stopSessionStream() " "must re-arm the on-screen session's stream, else bg_task_complete " "delivery dies until a page reload (Greptile P1 r3377162160)" ) # Specifically: the same-session no-op guard must be PRECEDED by a re-arm # so re-selecting a session whose stream a prior failed load killed revives # it. The re-arm sits before the guard (not inside a wrapping block), with # the updated authoritative-load check that still permits the same-session # guard when no different in-flight load is running; it's idempotent so # the real-switch path is unaffected. guard_ix = body.index("currentSid===sid && !forceReload && (!_loadingSessionId || _loadingSessionId===sid)") pre_guard = body[max(0, guard_ix - 600):guard_ix] assert "_rearmActiveSessionStream()" in pre_guard, ( "a re-arm must run before the same-session no-op guard so a " "previously-killed stream is revived on re-selecting the session" ) # The fetch-error catch must restart the stream for the on-screen session, # but guarded against the self-healed-current (404'd) case so it never # spins the reconnect loop against a dead session_id. catch_ix = body.index("const _selfHealedCurrent") catch_src = body[catch_ix:catch_ix + 2200] assert "!_selfHealedCurrent" in catch_src and "startSessionStream(currentSid)" in catch_src, ( "fetch-error path must restart the on-screen stream, guarded against " "the self-healed-current (deleted/404) session" ) def test_session_sse_stream_unsubscribes_on_header_write_failure(): """Deep-review fix (Codex): in _handle_session_sse_stream the subscriber slot is acquired by subscribe_to_session_channel BEFORE the SSE headers are written. The header writes (send_response/send_header/end_headers) touch the socket and can raise a client-disconnect error. If that happened OUTSIDE the try/finally, ch.unsubscribe(q) would be skipped and — because reaper_should_collect refuses to collect a channel with sub_count>0 — the channel would zombie forever. Pin that the subscribe and the header setup both sit inside the single try whose finally unsubscribes. """ from pathlib import Path src = Path(__file__).resolve().parents[1].joinpath("api", "routes.py").read_text(encoding="utf-8") i = src.find("def _handle_session_sse_stream(") assert i != -1, "handler not found" j = src.find("\ndef ", i + 1) body = src[i:j] sub_ix = body.find("subscribe_to_session_channel(") assert sub_ix != -1, "subscribe call not found" try_ix = body.find("try:", sub_ix) # SSE handlers finish their headers via end_sse_headers() (api/sse_chunked), # which is exactly handler.end_headers() unless chunked framing is opted in. end_headers_ix = body.find("end_sse_headers(handler)", sub_ix) finally_ix = body.find("finally:", sub_ix) unsub_ix = body.find("ch.unsubscribe(q)", finally_ix if finally_ix != -1 else sub_ix) # Order must be: subscribe → try → end_headers (inside try) → finally → unsubscribe. assert try_ix != -1 and finally_ix != -1 and unsub_ix != -1 assert sub_ix < try_ix < end_headers_ix < finally_ix < unsub_ix, ( "header setup must run INSIDE the try/finally that unsubscribes — " "a header-write disconnect must not leak a SessionChannel subscriber" )