fix(api): revalidate on a ready event, not on the connection opening

Both watch clients revalidated from `onopen`, which fires as soon as the response headers
arrive. Starlette flushes those BEFORE it starts iterating the stream generator, so the
sequence could be: headers out, client refetches the record log, a change lands and publishes,
and only then does the generator's `subscribe` complete. That change reaches neither the
refetch nor the stream, and on an idle screen the fallback poll is zero, so the transcript
stays stale until something else wakes it.

The endpoint now emits a `ready` event once the subscription is live, right behind the
`retry:` preamble, and both clients revalidate on that instead. `onopen` keeps only what it can
honestly report: the connection is up, so reset the backoff.

The new test drives the generator and asserts the channel is already subscribed by the time
`ready` is yielded, which is the whole invariant.

Author: ardaerzin

api/oss/src/apis/fastapi/sessions/watch.py

@@ -13,6 +13,7 @@
 from oss.src.dbs.redis.sessions.contract import (
     WATCH_EVENT_INTERACTION,
     WATCH_EVENT_LIFECYCLE,
+    WATCH_EVENT_READY,
     WATCH_EVENT_RECORDS_CHANGED,
 )
 from oss.src.utils.logging import get_module_logger
@@ -27,6 +28,11 @@ def retry_frame(retry_milliseconds: int) -> str:
     return f"retry: {retry_milliseconds}\n\n"
 
 
+def ready_frame() -> str:
+    """Emitted once the Redis subscription is live: the client's cue to revalidate."""
+    return "event: " + WATCH_EVENT_READY + "\ndata: {}\n\n"
+
+
 _KNOWN_EVENTS = {
     WATCH_EVENT_RECORDS_CHANGED,
     WATCH_EVENT_LIFECYCLE,
@@ -62,8 +68,15 @@ async def watch_event_stream(
 
     The first frame is a ``retry:`` preamble: it pins the client's built-in
     auto-reconnect delay (implementation-defined otherwise) so a server-side
-    drop — an API restart, a deploy — cannot reconnect-storm us. It also flushes
-    the response headers, so the client sees ``open`` before the first event.
+    drop — an API restart, a deploy — cannot reconnect-storm us.
+
+    The second is a ``ready`` event, and that is what a client revalidates on.
+    ``onopen`` fires as soon as the response headers arrive, and Starlette flushes
+    those BEFORE it starts iterating this generator — so a revalidation driven by
+    ``onopen`` can read the record log, and a change can land and publish, all before
+    the ``subscribe`` below completes. That change would reach neither the refetch nor
+    the stream. ``ready`` is emitted once the subscription is live, so a revalidation
+    keyed on it cannot straddle the gap.
 
     The subscription is torn down in ``finally`` — a client disconnect cancels
     the generator (GeneratorExit/CancelledError), which is exactly the cleanup
@@ -73,6 +86,7 @@ async def watch_event_stream(
     try:
         await pubsub.subscribe(channel)
         yield retry_frame(retry_milliseconds)
+        yield ready_frame()
         while True:
             message = await pubsub.get_message(
                 ignore_subscribe_messages=True,

api/oss/src/dbs/redis/sessions/contract.py

@@ -100,6 +100,10 @@ def make_displacement_payload(*, by: str) -> dict:
 WATCH_EVENT_RECORDS_CHANGED = "records-changed"
 WATCH_EVENT_LIFECYCLE = "lifecycle"
 WATCH_EVENT_INTERACTION = "interaction"
+# Emitted by the SSE endpoint itself, never published: it marks the point where the Redis
+# subscription is live, so a client can revalidate without racing the events it is about to
+# start receiving.
+WATCH_EVENT_READY = "ready"
 
 WATCH_LIFECYCLE_RUNNING = "running"
 WATCH_LIFECYCLE_ENDED = "ended"

api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py

@@ -18,6 +18,7 @@
 from oss.src.apis.fastapi.sessions.watch import (
     HEARTBEAT_FRAME,
     format_watch_frame,
+    ready_frame,
     retry_frame,
     watch_event_stream,
 )
@@ -73,22 +74,24 @@ async def test_stream_yields_event_frames_then_heartbeats():
     frames = []
     async for frame in stream:
         frames.append(frame)
-        if len(frames) == 5:
+        if len(frames) == 6:
             await stream.aclose()
             break
 
-    # The preamble pins the client's built-in auto-reconnect delay.
+    # The preamble pins the client's built-in auto-reconnect delay, then `ready` marks the
+    # point where the subscription is live and a client may safely revalidate.
     assert frames[0] == retry_frame(5000)
-    assert frames[1].startswith("event: records-changed\n")
-    assert json.loads(frames[1].split("data: ")[1]) == {
+    assert frames[1] == ready_frame()
+    assert frames[2].startswith("event: records-changed\n")
+    assert json.loads(frames[2].split("data: ")[1]) == {
         "type": "records-changed",
         "session_id": "s1",
     }
-    assert frames[2].startswith("event: lifecycle\n")
-    assert '"state": "running"' in frames[2]
-    assert frames[3].startswith("event: interaction\n")
+    assert frames[3].startswith("event: lifecycle\n")
+    assert '"state": "running"' in frames[3]
+    assert frames[4].startswith("event: interaction\n")
     # Queue drained -> the idle path emits keep-alive comments.
-    assert frames[4] == HEARTBEAT_FRAME
+    assert frames[5] == HEARTBEAT_FRAME
     assert pubsub.subscribed == ["watch:p:session:s1"]
 
 
@@ -103,6 +106,7 @@ async def test_stream_cleans_up_subscription_on_close():
     )
     # Take the preamble + one heartbeat, then simulate the client disconnecting.
     assert await stream.__anext__() == retry_frame(5000)
+    assert await stream.__anext__() == ready_frame()
     assert await stream.__anext__() == HEARTBEAT_FRAME
     await stream.aclose()
 
@@ -127,6 +131,7 @@ async def test_stream_skips_malformed_and_unknown_payloads():
         retry_milliseconds=5000,
     )
     assert await stream.__anext__() == retry_frame(5000)
+    assert await stream.__anext__() == ready_frame()
     frame = await stream.__anext__()
     await stream.aclose()
     # The three junk messages are dropped; the first frame is the real event.
@@ -189,6 +194,7 @@ async def test_stream_delivers_publisher_events_end_to_end():
     )
     # Preamble, then a heartbeat — proves the subscription is live before publishing.
     assert await stream.__anext__() == retry_frame(5000)
+    assert await stream.__anext__() == ready_frame()
     assert await stream.__anext__() == HEARTBEAT_FRAME
 
     publisher = SessionsWatchPublisher(redis_client=redis)
@@ -295,3 +301,26 @@ async def test_watch_endpoint_streams_the_configured_retry_preamble():
         await response.body_iterator.aclose()
 
     assert first == "retry: 9000\n\n"
+
+
+@pytest.mark.asyncio
+async def test_ready_is_not_emitted_before_the_subscription_is_live():
+    """The whole point of the `ready` event. A client revalidating on `onopen` races the
+    subscription: Starlette flushes the response headers before it iterates this generator, so a
+    change can land and publish in between and reach neither the refetch nor the stream. `ready`
+    is only reachable after `subscribe` has returned."""
+    pubsub = _FakePubSub([])
+    stream = watch_event_stream(
+        channel="watch:p:session:s1",
+        pubsub_factory=lambda: pubsub,
+        heartbeat_seconds=0.01,
+        retry_milliseconds=5000,
+    )
+
+    assert await stream.__anext__() == retry_frame(5000)
+    assert await stream.__anext__() == ready_frame()
+    assert pubsub.subscribed == ["watch:p:session:s1"], (
+        "ready reached the client before the channel was subscribed"
+    )
+
+    await stream.aclose()

web/mobile/src/features/chat/useSessionWatch.ts

@@ -96,11 +96,15 @@ export const useSessionWatch = ({
             es.onopen = () => {
                 setConnected(true)
                 attempt = 0
-                // Missed-event coverage: one revalidation per (re)connect replaces
-                // any replay/cursor semantics on the server.
+            }
+            // Missed-event coverage: one revalidation per (re)connect replaces any
+            // replay/cursor semantics on the server. Keyed on `ready`, not `onopen`:
+            // headers reach us before the server's Redis subscription is live, so a
+            // change landing in that window would miss both this refetch and the stream.
+            es.addEventListener("ready", () => {
                 notifyOnConnect()
                 invalidateBadges()
-            }
+            })
             es.addEventListener("records-changed", () => onRecordsChangedRef.current())
             es.addEventListener("lifecycle", invalidateBadges)
             es.addEventListener("interaction", invalidateBadges)

web/oss/src/components/AgentChatSlice/hooks/useSessionRecordsWatch.ts

@@ -109,12 +109,14 @@ export const useSessionRecordsWatch = ({
                 withCredentials: true,
             })
             source = es
-            // One revalidation per (re)connect covers events missed while disconnected — the
-            // server has no replay/cursor semantics.
             es.onopen = () => {
                 attempt = 0
-                notify()
             }
+            // One revalidation per (re)connect covers events missed while disconnected — the
+            // server has no replay/cursor semantics. Keyed on `ready` rather than `onopen`:
+            // headers arrive before the server's Redis subscription is live, so a change
+            // landing in that window would miss both this refetch and the stream.
+            es.addEventListener("ready", notify)
             es.addEventListener("records-changed", notify)
             es.onerror = () => {
                 // CONNECTING = built-in auto-reconnect; only a fatal CLOSED needs us.