[feat] Stream session changes and fix session liveness (9/12)

api/entrypoints/routers.py

@@ -25,6 +25,7 @@
     get_cache_engine,
     get_streams_engine,
 )
+from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher
 
 from oss.databases.postgres.migrations.core.utils import (
     check_for_new_migrations as check_for_new_core_migrations,
@@ -642,9 +643,14 @@ async def lifespan(*args, **kwargs):
 
 _lock_engine = get_lock_engine()
 
+# M3 live relay: lifecycle/interaction change notifications for the SSE watch
+# endpoint, published fire-and-forget on the durable plane.
+_sessions_watch_publisher = SessionsWatchPublisher()
+
 session_streams_service = SessionStreamsService(
     streams_dao=session_streams_dao,
     lock_engine=_lock_engine,
+    watch_publisher=_sessions_watch_publisher,
 )
 
 session_turns_service = SessionTurnsService(
@@ -819,6 +825,7 @@ async def lifespan(*args, **kwargs):
 
 interactions_service = SessionInteractionsService(
     interactions_dao=interactions_dao,
+    watch_publisher=_sessions_watch_publisher,
 )
 
 triggers_service = TriggersService(

api/entrypoints/worker_queues.py

@@ -69,6 +69,7 @@
 )
 from oss.src.dbs.postgres.sessions.interactions.dao import SessionInteractionsDAO
 from oss.src.dbs.postgres.sessions.records.dao import RecordsDAO
+from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher
 from oss.src.dbs.postgres.shared.engine import (
     get_analytics_engine,
     get_transactions_engine,
@@ -218,7 +219,12 @@ def _build_interactions_broker() -> tuple[AsyncBroker, int]:
     workflows_service.embeds_service = embeds_service
     environments_service.embeds_service = embeds_service
 
-    interactions_service = SessionInteractionsService(interactions_dao=interactions_dao)
+    interactions_service = SessionInteractionsService(
+        interactions_dao=interactions_dao,
+        # M3 live relay: approval resolutions land here (worker process), so this
+        # composition publishes watch notifications too.
+        watch_publisher=SessionsWatchPublisher(),
+    )
     # Approval answers replay the session's durable records into the resume conversation;
     # records live on the analytics engine (same as the API composition in routers.py).
     records_service = RecordsService(

api/entrypoints/worker_streams.py

@@ -32,6 +32,7 @@
 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
+from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher
 from oss.src.tasks.asyncio.events.worker import EventsWorker
 from oss.src.tasks.asyncio.sessions.records_worker import RecordsWorker
 from oss.src.tasks.asyncio.shared.consumer import StreamConsumer
@@ -83,6 +84,9 @@ async def _build_records_worker(redis_client: Redis) -> StreamConsumer:
         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),
     )
 
 

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

@@ -28,16 +28,21 @@
     Response,
     status,
 )
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse, StreamingResponse
 
 # FastAPI route params need fastapi.UploadFile; request.form() yields starlette's base class.
 from fastapi import UploadFile as FastAPIUploadFile
 from starlette.datastructures import UploadFile
 from typing import Any, Optional, Union
 
+from oss.src.utils.env import env
 from oss.src.utils.exceptions import intercept_exceptions
 from oss.src.utils.logging import get_module_logger
 
+from oss.src.dbs.redis.sessions.contract import watch_channel
+from oss.src.dbs.redis.shared.engine import get_streams_engine
+from oss.src.apis.fastapi.sessions.watch import watch_event_stream
+
 from oss.src.core.access.permissions.types import Permission
 from oss.src.core.access.permissions.service import check_action_access
 from oss.src.apis.fastapi.shared.exceptions import FORBIDDEN_EXCEPTION
@@ -335,6 +340,15 @@ def __init__(
             tags=["Sessions"],
         )
 
+        self.router.add_api_route(
+            "/sessions/streams/watch",
+            self.watch_session_stream,
+            methods=["GET"],
+            operation_id="watch_session_stream",
+            tags=["Sessions"],
+            response_model=None,
+        )
+
     @intercept_exceptions()
     @_handle_session_exceptions()
     async def set_session_stream(
@@ -520,6 +534,65 @@ async def set_session_stream_header(
         )
         return SessionStreamResponse(stream=stream)
 
+    @intercept_exceptions()
+    @_handle_session_exceptions()
+    async def watch_session_stream(
+        self,
+        request: Request,
+        session_id: str = Query(...),
+    ) -> StreamingResponse:
+        """Server-sent events relay for one session (M3 live relay).
+
+        Emits change notifications only — never record payloads; clients
+        revalidate through the regular query endpoints on each event:
+
+        - ``event: records-changed`` — ``{"session_id"}``; new/updated rows
+          landed in the record log (published post-DB-commit).
+        - ``event: lifecycle`` — ``{"session_id", "state": "running"|"ended"}``.
+        - ``event: interaction`` — ``{"session_id", "status": "pending"|"resolved"}``.
+        - ``: heartbeat`` comment frames while idle (keep-alive).
+
+        Auth is the standard middleware (cookie ``sAccessToken``, ApiKey, or
+        Bearer) evaluated once at connect; scope is the credential's project.
+        The stream has no replay/cursor semantics — ``EventSource`` reconnects
+        and clients revalidate once on every ``open``, which covers any missed
+        notifications.
+
+        NOTE (spec surface): this route appears in OpenAPI for documentation,
+        but Fern does not model SSE — consume it with a native ``EventSource``
+        (same-origin ``/api`` + cookie auth needs no custom headers), not the
+        generated client.
+        """
+        _validate_session_id_http(session_id)
+        project_id = request.state.project_id
+        user_id = request.state.user_id
+
+        has_permission = await check_action_access(
+            user_uid=str(user_id),
+            project_id=str(project_id),
+            permission=Permission.VIEW_SESSIONS,
+        )
+        if not has_permission:
+            raise FORBIDDEN_EXCEPTION
+
+        stream = watch_event_stream(
+            channel=watch_channel(str(project_id), session_id),
+            # One pubsub connection per SSE connection (v1 — simplest correct
+            # 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,
+        )
+        return StreamingResponse(
+            stream,
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                # Disable proxy buffering so frames flush immediately.
+                "X-Accel-Buffering": "no",
+            },
+        )
+
 
 class RecordsRouter:
     """Records sub-router — /sessions/records/*"""

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

@@ -0,0 +1,94 @@
+"""SSE frame generator for ``GET /sessions/streams/watch`` (M3 live relay).
+
+Bridges one Redis pub/sub subscription (durable plane, one per SSE connection)
+into `text/event-stream` frames. Events carry TYPE + minimal metadata only —
+clients revalidate through their existing query paths; no record payloads ride
+the wire. Idle periods emit ``: heartbeat`` comment frames so proxies and
+clients never see a silent connection.
+"""
+
+import json
+from typing import Any, AsyncIterator, Callable, Optional
+
+from oss.src.dbs.redis.sessions.contract import (
+    WATCH_EVENT_INTERACTION,
+    WATCH_EVENT_LIFECYCLE,
+    WATCH_EVENT_RECORDS_CHANGED,
+)
+from oss.src.utils.logging import get_module_logger
+
+log = get_module_logger(__name__)
+
+HEARTBEAT_FRAME = ": heartbeat\n\n"
+
+_KNOWN_EVENTS = {
+    WATCH_EVENT_RECORDS_CHANGED,
+    WATCH_EVENT_LIFECYCLE,
+    WATCH_EVENT_INTERACTION,
+}
+
+
+def format_watch_frame(raw: Any) -> Optional[str]:
+    """One published payload -> one SSE frame; None for anything malformed/unknown."""
+    try:
+        payload = json.loads(raw)
+    except (ValueError, TypeError):
+        return None
+    if not isinstance(payload, dict):
+        return None
+    event = payload.get("type")
+    # `_KNOWN_EVENTS` is a set, so an unhashable `type` (a list, a dict) would raise
+    # TypeError out of the membership test and tear down the whole stream. A malformed
+    # publish must drop its own frame, never the connection.
+    if not isinstance(event, str) or event not in _KNOWN_EVENTS:
+        return None
+    return f"event: {event}\ndata: {json.dumps(payload)}\n\n"
+
+
+async def watch_event_stream(
+    *,
+    channel: str,
+    pubsub_factory: Callable[[], Any],
+    heartbeat_seconds: float,
+) -> AsyncIterator[str]:
+    """Subscribe to the session's watch channel and yield SSE frames forever.
+
+    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)
+        while True:
+            message = await pubsub.get_message(
+                ignore_subscribe_messages=True,
+                timeout=heartbeat_seconds,
+            )
+            if message is None:
+                yield HEARTBEAT_FRAME
+                continue
+            if message.get("type") != "message":
+                continue
+            frame = format_watch_frame(message.get("data"))
+            if frame is not None:
+                yield frame
+    finally:
+        # Two independent attempts: a failing unsubscribe must not skip the close, or the
+        # connection outlives the disconnected client.
+        try:
+            await pubsub.unsubscribe(channel)
+        except Exception as exc:  # pragma: no cover — teardown is best-effort
+            log.warning(
+                "[WATCH] pubsub unsubscribe failed",
+                channel=channel,
+                error=repr(exc),
+            )
+        try:
+            await pubsub.aclose()
+        except Exception as exc:  # pragma: no cover — teardown is best-effort
+            log.warning(
+                "[WATCH] pubsub close failed",
+                channel=channel,
+                error=repr(exc),
+            )

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

@@ -12,11 +12,33 @@
 )
 from oss.src.core.sessions.interactions.types import InteractionNotFound
 from oss.src.core.shared.dtos import Windowing
+from oss.src.dbs.redis.sessions.contract import (
+    WATCH_INTERACTION_PENDING,
+    WATCH_INTERACTION_RESOLVED,
+)
+from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher
 
 
 class SessionInteractionsService:
-    def __init__(self, *, interactions_dao: SessionInteractionsDAOInterface) -> None:
+    def __init__(
+        self,
+        *,
+        interactions_dao: SessionInteractionsDAOInterface,
+        watch_publisher: Optional[SessionsWatchPublisher] = None,
+    ) -> None:
         self.interactions_dao = interactions_dao
+        self._watch = watch_publisher
+
+    async def _publish_interaction(
+        self, *, project_id: UUID, session_id: str, status: str
+    ) -> None:
+        # Fire-and-forget relay notification; the publisher never raises.
+        if self._watch is not None:
+            await self._watch.interaction(
+                project_id=str(project_id),
+                session_id=session_id,
+                status=status,
+            )
 
     async def create_interaction(
         self,
@@ -26,11 +48,17 @@ async def create_interaction(
         #
         interaction: SessionInteractionCreate,
     ) -> SessionInteraction:
-        return await self.interactions_dao.create_interaction(
+        created = await self.interactions_dao.create_interaction(
             project_id=project_id,
             user_id=user_id,
             interaction=interaction,
         )
+        await self._publish_interaction(
+            project_id=project_id,
+            session_id=interaction.session_id,
+            status=WATCH_INTERACTION_PENDING,
+        )
+        return created
 
     async def fetch_interaction(
         self,
@@ -59,6 +87,11 @@ async def transition_interaction(
             raise InteractionNotFound(
                 f"Interaction with token {transition.token!r} not found or already terminal"
             )
+        await self._publish_interaction(
+            project_id=transition.project_id,
+            session_id=transition.session_id,
+            status=WATCH_INTERACTION_RESOLVED,
+        )
         return result
 
     async def cancel_session_pending(
@@ -69,12 +102,19 @@ async def cancel_session_pending(
         except_turn_id: Optional[str] = None,
         except_tokens: Optional[List[str]] = None,
     ) -> int:
-        return await self.interactions_dao.cancel_session_pending(
+        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,
         )
+        if cancelled:
+            await self._publish_interaction(
+                project_id=project_id,
+                session_id=session_id,
+                status=WATCH_INTERACTION_RESOLVED,
+            )
+        return cancelled
 
     async def query_interactions(
         self,