feat: SSE keepalive, reconnection, and is_processing flag

Prevents SSE streams from being killed by reverse proxies during long
operations and lets the frontend recover gracefully after sleep/disconnect.

Backend:
- SSE keepalive pings every 15s via `: keepalive` comments
- New GET /api/events/{session_id} subscribe-only endpoint for reconnection
- is_processing flag on AgentSession, exposed in session info API

Frontend:
- reconnectToStream() re-attaches to a running session via /api/events
- Visibility change re-hydrates ALL sessions from backend on wake
- Background sessions also re-sync (removed isActive guard on listener)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

backend/models.py

@@ -75,6 +75,7 @@ class SessionInfo(BaseModel):
     session_id: str
     created_at: str
     is_active: bool
+    is_processing: bool = False
     message_count: int
     user_id: str = "dev"
     pending_approval: list[PendingApprovalTool] | None = None

backend/routes/agent.py

@@ -4,6 +4,7 @@ All routes (except /health) require authentication via the get_current_user
 dependency. In dev mode (no OAUTH_CLIENT_ID), auth is bypassed automatically.
 """
 
+import asyncio
 import json
 import logging
 from typing import Any
@@ -336,17 +337,34 @@ async def chat_sse(
         broadcaster.unsubscribe(sub_id)
         raise
 
-    # Terminal events that end the SSE stream
-    TERMINAL_EVENTS = {"turn_complete", "approval_required", "error", "interrupted", "shutdown"}
+    return _sse_response(broadcaster, event_queue, sub_id)
+
+
+# ---------------------------------------------------------------------------
+# Shared SSE helpers
+# ---------------------------------------------------------------------------
+_TERMINAL_EVENTS = {"turn_complete", "approval_required", "error", "interrupted", "shutdown"}
+_SSE_KEEPALIVE_SECONDS = 15
+
+
+def _sse_response(broadcaster, event_queue, sub_id) -> StreamingResponse:
+    """Build a StreamingResponse that drains *event_queue* as SSE,
+    sending keepalive comments every 15 s to prevent proxy timeouts."""
 
     async def event_generator():
         try:
             while True:
-                msg = await event_queue.get()
+                try:
+                    msg = await asyncio.wait_for(
+                        event_queue.get(), timeout=_SSE_KEEPALIVE_SECONDS
+                    )
+                except asyncio.TimeoutError:
+                    # SSE comment — ignored by parsers, keeps connection alive
+                    yield ": keepalive\n\n"
+                    continue
                 event_type = msg.get("event_type", "")
-                # Format as SSE
                 yield f"data: {json.dumps(msg)}\n\n"
-                if event_type in TERMINAL_EVENTS:
+                if event_type in _TERMINAL_EVENTS:
                     break
         finally:
             broadcaster.unsubscribe(sub_id)
@@ -362,6 +380,27 @@ async def chat_sse(
     )
 
 
+@router.get("/events/{session_id}")
+async def subscribe_events(
+    session_id: str,
+    user: dict = Depends(get_current_user),
+) -> StreamingResponse:
+    """Subscribe to events for a running session without submitting new input.
+
+    Used by the frontend to re-attach after a connection drop (e.g. screen
+    sleep).  Returns 404 if the session isn't active or isn't processing.
+    """
+    _check_session_access(session_id, user)
+
+    agent_session = session_manager.sessions.get(session_id)
+    if not agent_session or not agent_session.is_active:
+        raise HTTPException(status_code=404, detail="Session not found or inactive")
+
+    broadcaster = agent_session.broadcaster
+    sub_id, event_queue = broadcaster.subscribe()
+    return _sse_response(broadcaster, event_queue, sub_id)
+
+
 @router.post("/interrupt/{session_id}")
 async def interrupt_session(
     session_id: str, user: dict = Depends(get_current_user)

backend/session_manager.py

@@ -89,6 +89,7 @@ class AgentSession:
     task: asyncio.Task | None = None
     created_at: datetime = field(default_factory=datetime.utcnow)
     is_active: bool = True
+    is_processing: bool = False  # True while a submission is being executed
     broadcaster: Any = None
 
 
@@ -245,7 +246,11 @@ class SessionManager:
                         submission = await asyncio.wait_for(
                             submission_queue.get(), timeout=1.0
                         )
-                        should_continue = await process_submission(session, submission)
+                        agent_session.is_processing = True
+                        try:
+                            should_continue = await process_submission(session, submission)
+                        finally:
+                            agent_session.is_processing = False
                         if not should_continue:
                             break
                     except asyncio.TimeoutError:
@@ -405,6 +410,7 @@ class SessionManager:
             "session_id": session_id,
             "created_at": agent_session.created_at.isoformat(),
             "is_active": agent_session.is_active,
+            "is_processing": agent_session.is_processing,
             "message_count": len(agent_session.session.context_manager.items),
             "user_id": agent_session.user_id,
             "pending_approval": pending_approval,

frontend/src/components/SessionChat.tsx

@@ -44,10 +44,10 @@ export default function SessionChat({ sessionId, isActive, onSessionDead }: Sess
 
   // Re-sync state when the browser tab regains focus (Chrome throttles
   // timers in background tabs which can stall the AI SDK's update flushing).
+  // Fires for ALL sessions so background sessions also recover after sleep.
   useEffect(() => {
-    if (!isActive) return;
     const onVisible = () => {
-      if (document.visibilityState === 'visible') {
+      if (document.visibilityState === 'visible' && isActive) {
         useAgentStore.getState().switchActiveSession(sessionId);
       }
     };

frontend/src/hooks/useAgentChat.ts

@@ -339,6 +339,53 @@ export function useAgentChat({ sessionId, isActive, onReady, onError, onSessionD
     return () => { cancelled = true; };
   }, [sessionId]); // eslint-disable-line react-hooks/exhaustive-deps
 
+  // -- Re-hydrate on wake from sleep (SSE stream may have died) -----------
+  const rehydratingRef = useRef(false);
+  useEffect(() => {
+    const onVisible = async () => {
+      if (document.visibilityState !== 'visible') return;
+      if (rehydratingRef.current) return;
+      rehydratingRef.current = true;
+      try {
+        const [msgsRes, infoRes] = await Promise.all([
+          apiFetch(`/api/session/${sessionId}/messages`),
+          apiFetch(`/api/session/${sessionId}`),
+        ]);
+        if (!msgsRes.ok || !infoRes.ok) return;
+        const info = await infoRes.json();
+        const data = await msgsRes.json();
+        if (!Array.isArray(data) || data.length === 0) return;
+
+        // Rebuild pending-approval set
+        let pendingIds: Set<string> | undefined;
+        if (info.pending_approval && Array.isArray(info.pending_approval)) {
+          pendingIds = new Set(
+            info.pending_approval.map((t: { tool_call_id: string }) => t.tool_call_id)
+          );
+          if (pendingIds.size > 0) setNeedsAttention(sessionId, true);
+        }
+
+        const uiMsgs = llmMessagesToUIMessages(data, pendingIds);
+        if (uiMsgs.length > 0) {
+          chat.setMessages(uiMsgs);
+          saveMessages(sessionId, uiMsgs);
+        }
+
+        // If the backend is still processing but we lost the SSE stream,
+        // mark the UI as busy so the chat input stays disabled.
+        if (info.is_processing) {
+          updateSession(sessionId, { isProcessing: true, activityStatus: { type: 'thinking' } });
+        }
+      } catch {
+        /* ignore — backend may be briefly unreachable */
+      } finally {
+        rehydratingRef.current = false;
+      }
+    };
+    document.addEventListener('visibilitychange', onVisible);
+    return () => document.removeEventListener('visibilitychange', onVisible);
+  }, [sessionId]); // eslint-disable-line react-hooks/exhaustive-deps
+
   // -- Persist messages ---------------------------------------------------
   const prevLenRef = useRef(initialMessages.length);
   useEffect(() => {

frontend/src/lib/sse-chat-transport.ts

@@ -360,7 +360,29 @@ export class SSEChatTransport implements ChatTransport<UIMessage> {
   }
 
   async reconnectToStream(): Promise<ReadableStream<UIMessageChunk> | null> {
-    // Not needed — each turn is a separate request
-    return null;
+    // Check if the backend session is still processing a turn.
+    // If so, subscribe to its event stream so we can resume live updates
+    // (e.g. after page refresh or wake-from-sleep reconnection).
+    try {
+      const infoRes = await apiFetch(`/api/session/${this.sessionId}`);
+      if (!infoRes.ok) return null;
+      const info = await infoRes.json();
+      if (!info.is_processing) return null;
+
+      // Session is mid-turn — subscribe to its event broadcast.
+      const response = await apiFetch(`/api/events/${this.sessionId}`, {
+        headers: { 'Accept': 'text/event-stream' },
+      });
+      if (!response.ok || !response.body) return null;
+
+      this.sideChannel.onProcessing();
+
+      return response.body
+        .pipeThrough(new TextDecoderStream())
+        .pipeThrough(createSSEParserStream())
+        .pipeThrough(createEventToChunkStream(this.sideChannel));
+    } catch {
+      return null;
+    }
   }
 }