fix(streaming): preserve recovery after ambiguous profile 404s (#7299)

docs/rfcs/session-sse-contract-v1.md

@@ -67,19 +67,29 @@ The browser closes its persistent per-session SSE while hidden and uses
 subject to browser timer throttling. An active stream can be attached through
 the existing replay path; successful attachment stops the poll.
 
-HTTP `404` and `410` are terminal for the owning hidden-session poll. The
-response stops the interval and clears the matching hidden-resume session ID,
-so returning to the visible tab does not reopen SSE through that stale owner.
-A delayed response for session A must not stop session B's replacement poll or
-clear B's resume owner. This cleanup affects browser observation state only;
+HTTP `404` is ambiguous: the profile-visibility guard can return it for a live
+session after another tab changes the browser-wide active-profile cookie.
+A single `404` therefore keeps polling and retains the hidden-resume owner.
+After three consecutive `404` responses, the poll pauses to bound repeated
+missing-session requests, but retains that owner so returning to the visible
+tab can reopen SSE. Any other response or network error resets this budget;
+a newly started poll also starts with a fresh budget.
+
+HTTP `410` is terminal: it stops the interval and clears the matching
+hidden-resume session ID. Returning to the visible tab then does not reopen
+SSE through that owner. Responses and queued ticks belong to one poll timer,
+not merely a session ID: they cannot stop or attach a replacement poll, even
+when the replacement observes the same session. The budget counts responses
+received by the current poll. Cleanup affects browser observation state only;
 it does not delete a session or cancel an agent run.
 
 Successful idle responses with no `active_stream_id`, network failures, and
 other non-success HTTP responses (including `401`, `403`, `429`, and `5xx`)
 remain retryable. Visibility return normally restores per-session SSE when a
-resume owner still exists; a terminal missing-session response removes that
-automatic recovery path. Explicit later session selection follows normal
-session loading. Behavior coverage lives in
+resume owner still exists, including after repeated `404` responses. Only a
+terminal `410` removes that automatic recovery path. A profile restored before
+the three-miss limit can recover through the next hidden poll; after the limit,
+recovery waits for visibility return or explicit session loading. Behavior coverage lives in
 `tests/test_hidden_tab_server_initiated_turn.py`.
 
 ### Heartbeat

static/messages.js

@@ -8120,29 +8120,37 @@ function _startHiddenActiveStreamPoll(sid) {
   if (!sid) return;
   _stopHiddenActiveStreamPoll();
   _sessionStreamHiddenPollSid = sid;
+  let notFoundCount = 0;
+  let pollTimer = null;
+  const ownsPoll = () => _sessionStreamHiddenPollSid === sid &&
+    _sessionStreamHiddenPollTimer === pollTimer;
   const tick = () => {
+    // A queued tick/response must not mutate a replacement, even for the same sid.
+    if (!ownsPoll()) return;
     // Stop conditions: tab became visible (real SSE takes over), session
     // switched, or we're already rendering a stream.
     if (typeof document !== 'undefined' && !document.hidden) { _stopHiddenActiveStreamPoll(); return; }
-    if (_sessionStreamHiddenPollSid !== sid) { _stopHiddenActiveStreamPoll(); return; }
     if (S.activeStreamId) return; // already rendering; wait it out
     try {
       fetch(_apiUrl('api/session/status?session_id=' + encodeURIComponent(sid)), {credentials: 'same-origin'})
         .then(r => {
-          if (r && (r.status === 404 || r.status === 410)) {
-            // The session is permanently gone. Stop only when this response
-            // still belongs to the active poll so a late 404 from an older
-            // session cannot tear down its replacement.
-            if (_sessionStreamHiddenPollSid === sid) {
-              if (_sessionStreamHiddenSid === sid) _sessionStreamHiddenSid = null;
-              _stopHiddenActiveStreamPoll();
-            }
+          if (!ownsPoll()) return null;
+          if (r && r.status === 404) {
+            // Profile visibility also returns 404. Bound repeated misses, but
+            // retain the resume owner so a profile flip back can self-heal.
+            if (++notFoundCount >= 3) _stopHiddenActiveStreamPoll();
+            return null;
+          }
+          notFoundCount = 0;
+          if (r && r.status === 410) {
+            if (_sessionStreamHiddenSid === sid) _sessionStreamHiddenSid = null;
+            _stopHiddenActiveStreamPoll();
             return null;
           }
           return r && r.ok ? r.json() : null;
         })
         .then(d => {
-          if (!d || _sessionStreamHiddenPollSid !== sid) return;
+          if (!d || !ownsPoll()) return;
           const streamId = d.active_stream_id;
           if (streamId && S.activeStreamId !== String(streamId)) {
             // Server-initiated turn in flight while hidden → attach as replay.
@@ -8175,10 +8183,11 @@ function _startHiddenActiveStreamPoll(sid) {
             }
           }
         })
-        .catch(() => {});
-    } catch (_) {}
+        .catch(() => { notFoundCount = 0; });
+    } catch (_) { notFoundCount = 0; }
   };
-  _sessionStreamHiddenPollTimer = setInterval(tick, 6000);
+  pollTimer = setInterval(tick, 6000);
+  _sessionStreamHiddenPollTimer = pollTimer;
   // Fire one immediately so a turn already running when we go hidden is caught
   // without waiting a full interval.
   tick();

tests/test_hidden_tab_server_initiated_turn.py

@@ -95,7 +95,7 @@ def test_hidden_poll_hits_session_status_and_attaches_as_replay():
     """
     start = MESSAGES_JS.find("function _startHiddenActiveStreamPoll(sid)")
     assert start != -1
-    body = MESSAGES_JS[start:start + 3200]
+    body = MESSAGES_JS[start:MESSAGES_JS.index("function _stopHiddenActiveStreamPoll()", start)]
     assert "api/session/status?session_id=" in body
     assert "d.active_stream_id" in body
     # attaches as replay (recovered=true) — turn is already mid-flight
@@ -178,9 +178,7 @@ def test_poll_stops_only_when_attach_succeeds():
     """
     start = MESSAGES_JS.find("function _startHiddenActiveStreamPoll(sid)")
     assert start != -1, "_startHiddenActiveStreamPoll signature not found"
-    # Window 2400: the multi-pane follow-up adds an explanatory comment block
-    # before the attach call, pushing it past a narrower slice.
-    body = MESSAGES_JS[start:start + 3200]
+    body = MESSAGES_JS[start:MESSAGES_JS.index("function _stopHiddenActiveStreamPoll()", start)]
     assert "const attached = _attachServerInitiatedStream(sid, streamId, true)" in body
     # Stop the poll only on a true attach (the false branch keeps polling within
     # the bounded-retry budget rather than stopping).
@@ -191,14 +189,16 @@ def test_poll_stops_only_when_attach_succeeds():
     assert "_SESSION_STREAM_HIDDEN_POLL_MAX_FALSE" in body
 
 
-# ── Terminal missing-session responses stop the hidden poll ────────────────
+# ── Missing-session responses preserve recoverable profile ownership ──────
 
 NODE = shutil.which("node")
 
 
-@pytest.mark.skipif(NODE is None, reason="node not available")
-def test_hidden_poll_stops_only_for_owned_missing_sessions():
-    """Terminal responses stop only their owner and cannot reopen on visibility."""
+@pytest.fixture(scope="module")
+def hidden_poll_results():
+    """Execute the real poll and visibility handler with deterministic responses."""
+    if NODE is None:
+        pytest.skip("node not available")
     start = MESSAGES_JS.index("function _startHiddenActiveStreamPoll(sid)")
     poll_end = MESSAGES_JS.index("function _chatStreamActiveForSession(sid)", start)
     stream_end = MESSAGES_JS.index("function stopSessionStream()", poll_end)
@@ -232,7 +232,8 @@ def test_hidden_poll_stops_only_for_owned_missing_sessions():
           session: {session_id: 'session-a', message_count: 0},
         };
         const _apiUrl = value => value;
-        const _attachServerInitiatedStream = () => true;
+        let attachCalls = 0;
+        const _attachServerInitiatedStream = () => { attachCalls += 1; return true; };
         let intervalFn = null;
         let intervalSeq = 0;
         let fetchCalls = 0;
@@ -284,6 +285,7 @@ def test_hidden_poll_stops_only_for_owned_missing_sessions():
           intervalFn = null;
           fetchCalls = 0;
           eventSourceCalls = 0;
+          attachCalls = 0;
         }
 
         async function runStatus(status, reject = false) {
@@ -308,22 +310,26 @@ def test_hidden_poll_stops_only_for_owned_missing_sessions():
           };
         }
 
-        async function runStaleResponse() {
+        async function runStaleResponse(status, sameSession = false) {
           reset();
           let resolveA;
           globalThis.fetch = url => {
             fetchCalls += 1;
-            if (String(url).includes('session-a')) {
+            if (fetchCalls === 1) {
               return new Promise(resolve => { resolveA = resolve; });
             }
             return Promise.resolve(response(200));
           };
           _sessionStreamHiddenSid = 'session-a';
           _startHiddenActiveStreamPoll('session-a');
-          _sessionStreamHiddenSid = 'session-b';
-          _startHiddenActiveStreamPoll('session-b');
+          const oldTick = intervalFn;
+          const replacementSid = sameSession ? 'session-a' : 'session-b';
+          _sessionStreamHiddenSid = replacementSid;
+          _startHiddenActiveStreamPoll(replacementSid);
           await settle();
-          resolveA(response(404));
+          resolveA(response(status));
+          await settle();
+          oldTick();
           await settle();
           return {
             running: intervalFn !== null,
@@ -352,6 +358,28 @@ def test_hidden_poll_stops_only_for_owned_missing_sessions():
           };
         }
 
+        async function runSequence(statuses) {
+          reset();
+          const states = [];
+          globalThis.fetch = () => {
+            const status = statuses[fetchCalls++];
+            if (status === -1) return Promise.reject(new Error('offline'));
+            if (status === 'active') return Promise.resolve({ok: true, status: 200,
+              json: async () => ({active_stream_id: 'live-a'})});
+            return Promise.resolve(response(status));
+          };
+          startSessionStream('session-a');
+          for (let i = 0; i < statuses.length; i++) {
+            if (i && intervalFn) intervalFn();
+            await settle();
+            states.push({running: intervalFn !== null, hiddenSid: _sessionStreamHiddenSid});
+          }
+          document.hidden = false;
+          visibilityChange();
+          await settle();
+          return {states, fetchCalls, eventSourceCalls, attachCalls};
+        }
+
         (async () => {
           const result = {
             missing404: await runStatus(404),
@@ -359,9 +387,16 @@ def test_hidden_poll_stops_only_for_owned_missing_sessions():
             server500: await runStatus(500),
             offline: await runStatus(0, true),
             idle200: await runStatus(200),
-            stale404: await runStaleResponse(),
+            stale404: await runStaleResponse(404),
+            stale410: await runStaleResponse(410),
+            sameSession410: await runStaleResponse(410, true),
             visible404: await runVisibilityRecovery(404),
             visible410: await runVisibilityRecovery(410),
+            repeated404: await runSequence([404, 404, 404, 404]),
+            recoveredProfile: await runSequence([404, 'active']),
+            reset200: await runSequence([404, 404, 200, 404, 404, 404]),
+            reset500: await runSequence([404, 404, 500, 404, 404, 404]),
+            resetOffline: await runSequence([404, 404, -1, 404, 404, 404]),
           };
           process.stdout.write(JSON.stringify(result));
         })().catch(error => {
@@ -378,32 +413,64 @@ def test_hidden_poll_stops_only_for_owned_missing_sessions():
         timeout=10,
     )
     assert proc.returncode == 0, proc.stderr
-    result = json.loads(proc.stdout)
+    return json.loads(proc.stdout)
 
-    for key in ("visible404", "visible410"):
-        assert result[key] == {
+
+def test_hidden_poll_transient_404_preserves_visibility_recovery(hidden_poll_results):
+    result = hidden_poll_results
+    assert result["visible404"]["eventSourceCalls"] == 1
+    assert result["missing404"] == {
+        "fetchCalls": 2, "running": True, "pollSid": "session-a", "hiddenSid": "session-a",
+    }
+    assert result["recoveredProfile"]["attachCalls"] == 1
+
+
+def test_hidden_poll_repeated_404_is_bounded_but_can_resume(hidden_poll_results):
+    result = hidden_poll_results["repeated404"]
+    assert [s["running"] for s in result["states"]] == [True, True, False, False]
+    assert all(s["hiddenSid"] == "session-a" for s in result["states"])
+    assert result["fetchCalls"] == 3
+    assert result["eventSourceCalls"] == 1
+
+
+@pytest.mark.parametrize("key", ["reset200", "reset500", "resetOffline"])
+def test_hidden_poll_404_budget_requires_consecutive_responses(hidden_poll_results, key):
+    result = hidden_poll_results[key]
+    assert [s["running"] for s in result["states"]] == [True] * 5 + [False]
+    assert result["eventSourceCalls"] == 1
+
+
+def test_hidden_poll_410_stops_and_clears_resume_owner(hidden_poll_results):
+    result = hidden_poll_results
+    assert result["visible410"] == {
             "eventSourceCalls": 0,
             "running": False,
             "pollSid": None,
             "hiddenSid": None,
-        }
-
-    for key in ("missing404", "missing410"):
-        assert result[key] == {
+    }
+    assert result["missing410"] == {
             "fetchCalls": 1,
             "running": False,
             "pollSid": None,
             "hiddenSid": None,
-        }
+    }
 
+
+def test_hidden_poll_transient_failures_and_idle_remain_retryable(hidden_poll_results):
+    result = hidden_poll_results
     for key in ("server500", "offline", "idle200"):
         assert result[key]["fetchCalls"] == 2
         assert result[key]["running"] is True
         assert result[key]["pollSid"] == "session-a"
         assert result[key]["hiddenSid"] == "session-a"
 
-    assert result["stale404"] == {
+
+@pytest.mark.parametrize("key,sid", [
+    ("stale404", "session-b"), ("stale410", "session-b"), ("sameSession410", "session-a"),
+])
+def test_hidden_poll_stale_response_and_tick_cannot_stop_replacement(hidden_poll_results, key, sid):
+    assert hidden_poll_results[key] == {
         "running": True,
-        "pollSid": "session-b",
-        "hiddenSid": "session-b",
+        "pollSid": sid,
+        "hiddenSid": sid,
     }