[fix] Converge the desktop session when a gate is answered elsewhere (11/12)

api/entrypoints/worker_streams.py

@@ -25,10 +25,12 @@
 
 from oss.src.core.events.service import EventsService
 from oss.src.core.secrets.services import VaultService
+from oss.src.core.sessions.interactions.service import SessionInteractionsService
 from oss.src.core.sessions.records.service import RecordsService
 from oss.src.core.tracing.service import TracingService
 from oss.src.dbs.postgres.events.dao import EventsDAO
 from oss.src.dbs.postgres.secrets.dao import SecretsDAO
+from oss.src.dbs.postgres.sessions.interactions.dao import SessionInteractionsDAO
 from oss.src.dbs.postgres.sessions.records.dao import RecordsDAO
 from oss.src.dbs.postgres.tracing.dao import TracingDAO
 from oss.src.dbs.postgres.webhooks.dao import WebhooksDAO
@@ -79,14 +81,22 @@ async def _build_spans_worker(redis_client: Redis) -> StreamConsumer:
 
 
 async def _build_records_worker(redis_client: Redis) -> StreamConsumer:
+    watch_publisher = SessionsWatchPublisher(redis_client=redis_client)
     return RecordsWorker(
         service=RecordsService(records_dao=RecordsDAO()),
         redis_client=redis_client,
         stream_name="streams:records",
         consumer_group="worker-records",
         # M3 live relay: post-append change notifications on the durable plane,
         # reusing this process's durable connection.
-        watch_publisher=SessionsWatchPublisher(redis_client=redis_client),
+        watch_publisher=watch_publisher,
+        # The gate safety net: this loop sees every turn's terminal record, so it is where a
+        # pending gate that outlived its turn gets cancelled, scoped to that turn's own gates
+        # so a newer turn's live park is never in range.
+        interactions_service=SessionInteractionsService(
+            interactions_dao=SessionInteractionsDAO(),
+            watch_publisher=watch_publisher,
+        ),
     )
 
 

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

@@ -554,6 +554,11 @@ async def watch_session_stream(
 
         Auth is the standard middleware (cookie ``sAccessToken``, ApiKey, or
         Bearer) evaluated once at connect; scope is the credential's project.
+        Browsers authenticate by cookie — ``EventSource`` cannot set headers —
+        so a connect landing on an expired access token 401s like any other
+        request. There is no interceptor to refresh-and-retry a stream, so the
+        client must refresh the session itself and reopen (see the web hooks).
+
         The stream has no replay/cursor semantics — ``EventSource`` reconnects
         and clients revalidate once on every ``open``, which covers any missed
         notifications.
@@ -581,6 +586,7 @@ async def watch_session_stream(
             # teardown story; revisit with a shared listener if counts grow).
             pubsub_factory=lambda: get_streams_engine().get_redis().pubsub(),
             heartbeat_seconds=env.sessions.watch_heartbeat_seconds,
+            retry_milliseconds=env.sessions.watch_retry_milliseconds,
         )
         return StreamingResponse(
             stream,

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
@@ -21,6 +22,17 @@
 
 HEARTBEAT_FRAME = ": heartbeat\n\n"
 
+
+def retry_frame(retry_milliseconds: int) -> str:
+    """SSE `retry:` field — sets the client's built-in auto-reconnect delay."""
+    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,
@@ -50,16 +62,31 @@ async def watch_event_stream(
     channel: str,
     pubsub_factory: Callable[[], Any],
     heartbeat_seconds: float,
+    retry_milliseconds: int,
 ) -> AsyncIterator[str]:
     """Subscribe to the session's watch channel and yield SSE frames forever.
 
+    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.
+
+    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
     path, so no Redis subscription outlives its SSE connection.
     """
     pubsub = pubsub_factory()
     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/core/sessions/interactions/interfaces.py

@@ -46,6 +46,7 @@ async def cancel_session_pending(
         session_id: str,
         except_turn_id: Optional[str] = None,
         except_tokens: Optional[List[str]] = None,
+        only_turn_id: Optional[str] = None,
     ) -> int: ...
 
     @abstractmethod

api/oss/src/core/sessions/interactions/service.py

@@ -101,12 +101,14 @@ async def cancel_session_pending(
         session_id: str,
         except_turn_id: Optional[str] = None,
         except_tokens: Optional[List[str]] = None,
+        only_turn_id: Optional[str] = None,
     ) -> int:
         cancelled = await self.interactions_dao.cancel_session_pending(
             project_id=project_id,
             session_id=session_id,
             except_turn_id=except_turn_id,
             except_tokens=except_tokens,
+            only_turn_id=only_turn_id,
         )
         if cancelled:
             await self._publish_interaction(

api/oss/src/core/sessions/streams/service.py

@@ -13,7 +13,7 @@
 """
 
 import uuid_utils.compat as uuid
-from typing import List, Optional
+from typing import Iterable, List, Optional
 from uuid import UUID
 
 from oss.src.utils.logging import get_module_logger
@@ -35,10 +35,14 @@
     force_cancel_alive,
     force_clear_owner,
     get_alive_owner,
+    get_owner,
     get_running_owner,
     get_session_liveness,
+    is_turn_superseded,
+    mark_turn_superseded,
     refresh_alive,
     refresh_running,
+    release_alive,
     release_attached,
     steal_attached,
 )
@@ -86,6 +90,57 @@ def __init__(
         self._lock = lock_engine
         self._watch = watch_publisher
 
+    async def _supersede_turns(
+        self,
+        *,
+        project_id: UUID,
+        session_id: str,
+        turn_ids: Iterable[Optional[str]],
+    ) -> None:
+        """Tombstone every turn displaced by this edit. `displaced ⇒ dead` is the invariant
+        that makes the ambiguous "`alive` held by another turn + no `running`" state safe to
+        resolve as a handover: only a turn that has never been displaced can reach it."""
+        for turn_id in {t for t in turn_ids if t}:
+            await mark_turn_superseded(
+                self._lock,
+                project_id=str(project_id),
+                session_id=session_id,
+                turn_id=turn_id,
+            )
+
+    async def _displace_turns(self, *, project_id: UUID, session_id: str) -> None:
+        """Tear alive+running off whichever turn holds them, tombstoning it first.
+
+        The order is the point. Clearing first leaves a window in which the turn being
+        displaced heartbeats, finds `alive` free and nx-acquires it straight back - a
+        cancelled session then reads as alive for a whole ALIVE_TTL. Tombstoning first makes
+        that beat refuse itself. The keys are still re-read after the clear, so a turn that
+        took them inside the window is tombstoned too.
+        """
+        await self._supersede_turns(
+            project_id=project_id,
+            session_id=session_id,
+            turn_ids=(
+                await get_alive_owner(
+                    self._lock, project_id=str(project_id), session_id=session_id
+                ),
+                await get_running_owner(
+                    self._lock, project_id=str(project_id), session_id=session_id
+                ),
+            ),
+        )
+        displaced_alive = await force_cancel_alive(
+            self._lock, project_id=str(project_id), session_id=session_id
+        )
+        displaced_running = await clear_running(
+            self._lock, project_id=str(project_id), session_id=session_id
+        )
+        await self._supersede_turns(
+            project_id=project_id,
+            session_id=session_id,
+            turn_ids=(displaced_alive, displaced_running),
+        )
+
     async def _publish_lifecycle(
         self, *, project_id: UUID, session_id: str, state: str
     ) -> None:
@@ -138,12 +193,7 @@ async def command(
             )
 
         elif mode == CommandMode.steer:
-            await force_cancel_alive(
-                self._lock, project_id=str(project_id), session_id=session_id
-            )
-            await clear_running(
-                self._lock, project_id=str(project_id), session_id=session_id
-            )
+            await self._displace_turns(project_id=project_id, session_id=session_id)
             turn_id = await self._start_turn(
                 project_id=project_id,
                 user_id=user_id,
@@ -157,12 +207,7 @@ async def command(
             )
 
         elif mode == CommandMode.cancel:
-            await force_cancel_alive(
-                self._lock, project_id=str(project_id), session_id=session_id
-            )
-            await clear_running(
-                self._lock, project_id=str(project_id), session_id=session_id
-            )
+            await self._displace_turns(project_id=project_id, session_id=session_id)
             await self._mark_stream_ended(
                 project_id=project_id,
                 user_id=user_id,
@@ -241,12 +286,7 @@ async def kill(
         whose runner replica is unreachable, is still a no-op success (best-effort teardown).
         """
         _validate_session_id(session_id)
-        await force_cancel_alive(
-            self._lock, project_id=str(project_id), session_id=session_id
-        )
-        await clear_running(
-            self._lock, project_id=str(project_id), session_id=session_id
-        )
+        await self._displace_turns(project_id=project_id, session_id=session_id)
         # Drop affinity too: claim_owner never steals, so a surviving owner key would lock
         # the session out of every other replica for the rest of OWNER_TTL_SECONDS.
         await force_clear_owner(
@@ -292,6 +332,39 @@ async def heartbeat(
     ) -> SessionHeartbeatResult:
         _validate_session_id(request.session_id)
 
+        # A turn that was already displaced (handover, cancel, steer, kill, sweep) is dead
+        # forever: refuse the beat before it touches ANY lock or the row. This is what keeps
+        # the ambiguous "`alive` held by another turn + no `running`" state safe to resolve as
+        # a handover below — the dangerous reading of that state was a zombie beat from an
+        # older turn taking the nest of a session parked awaiting approval, which then made
+        # the user's approval resume look superseded and abort. A zombie is by definition a
+        # turn that already lost the nest, so the tombstone written at the moment it lost it
+        # is the discriminator the locks alone cannot provide. Refusing early also stops the
+        # zombie's own turn-end beat from clearing the LIVE turn's `running`.
+        if request.turn_id and await is_turn_superseded(
+            self._lock,
+            project_id=str(project_id),
+            session_id=request.session_id,
+            turn_id=request.turn_id,
+        ):
+            stream = await self._dao.get_by_session_id(
+                project_id=project_id,
+                session_id=request.session_id,
+            )
+            # Read affinity, never claim it: renewing OWNER_TTL on a dead turn's beat pins the
+            # session to this replica for another full TTL, which is exactly what has to expire
+            # before another replica can take the session over.
+            owner = await get_owner(
+                self._lock,
+                project_id=str(project_id),
+                session_id=request.session_id,
+            )
+            return SessionHeartbeatResult(
+                stream=stream,
+                replica_id=owner or request.replica_id,
+                is_current_turn=False,
+            )
+
         # replica_id claims affinity without stealing from a live different owner; turn_id
         # separately refreshes the alive/running TTLs. `owner` is the actual winner (this
         # replica if it won or already held it, another replica otherwise).
@@ -337,10 +410,13 @@ async def heartbeat(
             # Acquire-then-refresh: the first heartbeat must establish the nest locks
             # itself (acquire_* is nx=True — a no-op if _start_turn already holds them).
             # A failed nx acquire is NOT by itself a takeover: nx fails whenever ANY value
-            # holds the key, and `alive` outlives its turn (release_alive has no callers, and
+            # holds the key, and `alive` outlives its turn (nothing releases it at turn end, and
             # the turn-end beat clears only `running`), so every follow-up turn on a warm
             # session sees the previous turn's key. `running` is the discriminator — a real
-            # takeover (steer/_start_turn) holds it under the usurper's turn id.
+            # takeover (steer/_start_turn) holds it under the usurper's turn id — and the
+            # supersession tombstone checked above is what makes the remaining "no running"
+            # case safe: any turn that could reach here dishonestly has already been
+            # tombstoned by whatever displaced it.
             if not await refresh_alive(
                 self._lock,
                 project_id=str(project_id),
@@ -371,18 +447,36 @@ async def heartbeat(
                         pass  # a live different turn holds the session: real takeover
                     else:
                         # Stale `alive` from this session's own previous (ended or parked)
-                        # turn — legitimate handover, not an interruption.
-                        await force_cancel_alive(
-                            self._lock,
-                            project_id=str(project_id),
-                            session_id=request.session_id,
-                        )
-                        acquired = await acquire_alive(
+                        # turn — legitimate handover, not an interruption. The displaced turn
+                        # is tombstoned so it can never beat its way back in: that is the only
+                        # thing standing between this branch and a zombie stealing the nest of
+                        # a parked session.
+                        #
+                        # Compare-and-delete against the owner read just above, never an
+                        # unconditional delete: an API `_start_turn` can land in the gap
+                        # between that read and this write, and clearing the key then would
+                        # tombstone the live turn that had just taken the session. Losing that
+                        # race leaves `acquired` False, which is the truth.
+                        displaced = alive_owner
+                        if displaced is None or await release_alive(
                             self._lock,
                             project_id=str(project_id),
                             session_id=request.session_id,
-                            turn_id=request.turn_id,
-                        )
+                            turn_id=displaced,
+                        ):
+                            if displaced and displaced != request.turn_id:
+                                await mark_turn_superseded(
+                                    self._lock,
+                                    project_id=str(project_id),
+                                    session_id=request.session_id,
+                                    turn_id=displaced,
+                                )
+                            acquired = await acquire_alive(
+                                self._lock,
+                                project_id=str(project_id),
+                                session_id=request.session_id,
+                                turn_id=request.turn_id,
+                            )
                 if not acquired or turn_was_established:
                     is_current_turn = False
             if not await refresh_running(

api/oss/src/dbs/postgres/sessions/interactions/dao.py

@@ -141,12 +141,13 @@ async def cancel_session_pending(
         session_id: str,
         except_turn_id: Optional[str] = None,
         except_tokens: Optional[List[str]] = None,
+        only_turn_id: Optional[str] = None,
     ) -> int:
         """Cancel still-pending interactions for a session. With `except_turn_id`, spare the
         current turn's own gates (used at turn start to cancel prior turns' unanswered gates;
         without it, cancel all of them, e.g. on kill). `except_tokens` spares prior-turn gates
-        the current turn answers in-band, so the resume can resolve them instead. Returns the
-        count cancelled."""
+        the current turn answers in-band, so the resume can resolve them instead. With
+        `only_turn_id`, touch nothing but that one turn's gates. Returns the count cancelled."""
         async with self.engine.session() as session:
             stmt = (
                 sa_update(SessionInteractionDBE)
@@ -160,6 +161,8 @@ async def cancel_session_pending(
                     updated_at=datetime.now(timezone.utc),
                 )
             )
+            if only_turn_id is not None:
+                stmt = stmt.where(SessionInteractionDBE.turn_id == only_turn_id)
             if except_turn_id is not None:
                 stmt = stmt.where(SessionInteractionDBE.turn_id != except_turn_id)
             if except_tokens: