fix(ui): enforce owner-routed event streams

coding-agent/docs/adr/0015-enforce-owner-routed-http-event-streams.md

@@ -0,0 +1,50 @@
+# ADR-0015: Enforce owner-routed HTTP event streams for Phase 2 sticky routing
+
+**Status**: Proposed
+**Date**: 2026-04-21
+
+## Context
+
+ADR-0013 defines Phase 2 multi-instance ownership as an owner/lease/fencing layer on top of PostgreSQL-backed session durability. That ADR already says runtime execution, approval handling, and event streaming are owner-routed operations, and it explicitly allows sticky routing as the first event-routing strategy.
+
+The current HTTP implementation does not yet enforce that rule for live SSE streams. `GET /sessions/{session_id}/events` only checks that the session exists, then registers an in-memory event queue and starts a keepalive loop. In a multi-instance deployment, a non-owner instance can still attach a live event stream for a session it does not own, and an already-open stream on the old owner is not forced to stop when ownership changes.
+
+That gap matters because Phase 2 is already using owner checks to fence runtime-sensitive mutations (`run_agent`, checkpoint restore, close, approval submission). Leaving `/events` unfenced means the repository still allows a stale instance to appear live even after the owner has changed. This does not violate at-most-once execution directly, but it does violate the owner-routed contract that ADR-0013 and the PG phase 2 checklist already claim.
+
+## Decision
+
+Adopt owner-routed sticky routing as the first concrete Phase 2 policy for HTTP event streams.
+
+Specifically:
+
+- `GET /sessions/{session_id}/events` must reject non-owner instances before registering a live event queue.
+- The rejection contract for a non-owner or stale owner is `409 Conflict` with the existing `SessionOwnershipConflictError` detail, matching the current Phase 2 HTTP conflict surface.
+- Once a stream is open, the keepalive loop must re-check ownership before emitting keepalive pings. If ownership is lost, the generator must terminate and allow queue cleanup to run.
+- The first cut remains sticky-routing only. It does not add brokered event routing, redirect payloads, or cross-instance queue transfer.
+- The owner check should be exposed through public `SessionManager` methods for event-stream authorization rather than calling `_assert_owner` directly from HTTP handlers.
+
+This ADR intentionally keeps approval metadata persistence out of scope. It closes the narrowest, clearest routing gap first: only the current owner may serve the live SSE stream for a session.
+
+## Alternatives Rejected
+
+- Keep `/events` unfenced and rely on load-balancer stickiness alone — rejected because it leaves stale instances free to attach live streams even when PostgreSQL ownership already says they are not authoritative.
+- Add brokered event routing now — rejected because it expands Phase 2 into cross-instance fan-out before the repository has fully closed the simpler owner-routed sticky-routing contract.
+- Return a redirect payload instead of `409 Conflict` — rejected for the first cut because the HTTP API already uses `409` for Phase 2 ownership conflicts and there is no canonical owner-address discovery mechanism yet.
+- Continue serving already-open streams after owner loss until disconnect — rejected because it weakens the owner-routed contract and lets stale instances keep presenting themselves as authoritative.
+
+## Acceptance Criteria
+
+- [ ] `test_get_events_rejects_stale_owner_before_stream_registration`
+- [ ] `test_get_events_stops_stream_after_owner_change`
+- [ ] `test_get_events_keeps_stream_alive_for_current_owner`
+- [ ] `test_get_events_returns_404_before_owner_check_for_missing_session`
+- [ ] `uv run pytest tests/ui/test_http_server_failover.py -v`
+- [ ] `uv run pytest tests/ui/test_http_server.py tests/ui/test_http_server_failover.py tests/ui/test_session_manager_owner_checks.py -k "events or owner or failover" -v`
+
+## References
+
+- `docs/adr/0013-define-phase2-multi-instance-session-ownership-for-pg-http-sessions.md`
+- `docs/pg-phase2-multi-instance-checklist.md`
+- `src/coding_agent/ui/http_server.py`
+- `src/coding_agent/ui/session_manager.py`
+- `tests/ui/test_http_server.py`

coding-agent/src/coding_agent/ui/http_server.py

@@ -652,25 +652,44 @@ async def get_events(
     api_key: str | None = Depends(verify_api_key),
 ) -> EventSourceResponse:
     """Persistent SSE event stream (fan-out supported)."""
-    if not await session_manager.has_session_async(session_id):
-        raise HTTPException(status_code=404, detail="Session not found")
+    try:
+        await session_manager.get_session_async(session_id)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail=_key_error_detail(exc)) from exc
+
+    try:
+        await session_manager.authorize_event_stream(session_id)
+    except SessionOwnershipConflictError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
 
-    session = await session_manager.get_session_async(session_id)
     queue: asyncio.Queue[dict[str, str]] = asyncio.Queue(maxsize=100)
-    await session_manager.add_event_queue_async(session_id, queue)
+    try:
+        await session_manager.register_owned_event_queue_async(session_id, queue)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail=_key_error_detail(exc)) from exc
+    except SessionOwnershipConflictError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
 
     async def event_generator() -> AsyncIterator[dict[str, str]]:
         """Generate events from queue."""
         try:
             while True:
                 try:
                     event = await asyncio.wait_for(queue.get(), timeout=30.0)
+                    try:
+                        await session_manager.verify_event_stream_ownership(session_id)
+                    except SessionOwnershipConflictError:
+                        break
                     yield event
                     if event.get("event") == "SessionClosed":
                         break
                 except asyncio.TimeoutError:
                     if not await session_manager.has_session_async(session_id):
                         break
+                    try:
+                        await session_manager.verify_event_stream_ownership(session_id)
+                    except SessionOwnershipConflictError:
+                        break
                     try:
                         if not await session_manager.has_event_queue_async(
                             session_id, queue

coding-agent/src/coding_agent/ui/session_manager.py

@@ -516,6 +516,12 @@ async def _assert_owner(self, session_id: str) -> None:
         ):
             raise SessionOwnershipConflictError("stale owner or fencing token rejected")
 
+    async def authorize_event_stream(self, session_id: str) -> None:
+        await self._assert_owner(session_id)
+
+    async def verify_event_stream_ownership(self, session_id: str) -> None:
+        await self._assert_owner(session_id)
+
     async def _run_store_io(self, func: Callable[..., T], /, *args: object) -> T:
         async with self._store_io_guard:
             loop = asyncio.get_running_loop()
@@ -571,6 +577,22 @@ async def add_event_queue_async(
         session = await self.get_session_async(session_id)
         session.event_queues.append(queue)
 
+    async def register_owned_event_queue_async(
+        self,
+        session_id: str,
+        queue: asyncio.Queue[dict[str, Any]],
+    ) -> None:
+        async with self._lock:
+            await self._assert_owner(session_id)
+            session = await self.get_session_async(session_id)
+            session.event_queues.append(queue)
+            try:
+                await self._assert_owner(session_id)
+            except (Exception, asyncio.CancelledError):
+                if queue in session.event_queues:
+                    session.event_queues.remove(queue)
+                raise
+
     async def remove_event_queue_async(
         self,
         session_id: str,
@@ -591,14 +613,18 @@ async def _close_resource_async(self, resource: object) -> None:
         if isawaitable(close_result):
             await close_result
 
-    async def remove_session_async(self, session_id: str) -> None:
+    async def _remove_session_async_no_lock(self, session_id: str) -> None:
         session = await self.get_session_async(session_id)
         await self._close_runtime(session)
         self._session_cache.pop(session_id, None)
         await self._run_store_io(self._store.delete, session_id)
         self._approval_stores.pop(session_id, None)
         self._session_turn_locks.pop(session_id, None)
 
+    async def remove_session_async(self, session_id: str) -> None:
+        async with self._lock:
+            await self._remove_session_async_no_lock(session_id)
+
     async def _restore_tape(self, tape_id: str | None) -> Tape | None:
         if tape_id is None:
             return None
@@ -983,7 +1009,7 @@ async def close_session(self, session_id: str) -> None:
                         f"Session task for {session_id} did not stop after cancellation"
                     )
 
-            await self.remove_session_async(session_id)
+            await self._remove_session_async_no_lock(session_id)
 
         logger.info(f"Closed session: {session_id}")
 

coding-agent/tests/ui/test_http_server.py

@@ -866,7 +866,7 @@ async def test_event_generator_uses_public_session_apis_for_keepalive_exit(
 
         has_session_calls: list[str] = []
         get_session_calls: list[str] = []
-        has_session_results = iter([True, True, False])
+        has_session_results = iter([True, False])
 
         class FakeEventSourceResponse:
             def __init__(self, body_iterator):
@@ -920,7 +920,7 @@ async def fake_wait_for(awaitable, timeout):
         with pytest.raises(StopAsyncIteration):
             await anext(event_generator)
 
-        assert has_session_calls == [session_id, session_id, session_id]
+        assert has_session_calls == [session_id, session_id]
         assert len(get_session_calls) >= 2
         assert all(
             call_session_id == session_id for call_session_id in get_session_calls