How Pi Web Implements Running State SSE Broadcast to Avoid Polling

Pi Web eliminates client-side polling for session status updates by implementing a server-sent events (SSE) broadcaster that pushes the set of running session IDs to connected clients only when the state actually changes.

The agegr/pi-web repository solves the real-time sidebar synchronization problem—showing which agent sessions are currently executing—through a push-based architecture. Instead of browsers repeatedly asking the server for updates, the server maintains a persistent connection and streams changes instantly, reducing network overhead and latency.

Global Registry and Broadcaster

The core mechanism resides in lib/rpc-manager.ts, which implements a global pub/sub system that survives Next.js hot-reloads. This module tracks which RPC sessions are actively running and manages a registry of listeners waiting for state changes.

The Listener Management System

To ensure the broadcaster persists across module reloads during development, Pi Web stores the listener set on globalThis.__piRunningListeners. The subscribeRunningSessions() function adds callbacks to this global Set and returns an unsubscribe function for cleanup.

// lib/rpc-manager.ts – Listener registration
export function subscribeRunningSessions(listener: (ids: string[]) => void): () => void {
  const listeners = getRunningListeners();   // global singleton Set
  listeners.add(listener);
  return () => { listeners.delete(listener); };
}

Delta Detection and Notification

The notifyRunningChange() function implements intelligent deduplication. It recomputes the running session IDs via getRunningRpcSessionIds(), serializes them to a JSON snapshot, and compares against lastRunningSnapshot. Only when the sorted ID list actually differs does it broadcast to all registered listeners.

// lib/rpc-manager.ts – Broadcasting logic
export function notifyRunningChange(): void {
  const ids = getRunningRpcSessionIds();      // collect IDs where session.isRunning()
  const snapshot = JSON.stringify([...ids].sort());
  if (snapshot === lastRunningSnapshot) return;   // dedupe
  lastRunningSnapshot = snapshot;
  for (const listener of getRunningListeners()) {
    try { listener(ids); } catch { /* ignore */ }
  }
}

Each AgentSessionWrapper invokes notifyRunningChange() whenever its isRunning() state toggles—whether on session start, completion, or cancellation—ensuring the broadcast fires immediately upon state transition.

The SSE Endpoint

The API route at app/api/agent/running/events/route.ts creates an EventSource-compatible HTTP response that bridges the global broadcaster to the client. This endpoint handles the Server-Sent Events protocol, including proper headers and stream formatting.

Stream Initialization

When a client connects, the route immediately sends the current snapshot of running session IDs, then registers the connection's controller as a listener via subscribeRunningSessions(). This ensures the client receives the initial state before waiting for updates.

// app/api/agent/running/events/route.ts – SSE handler
import { getRunningRpcSessionIds, subscribeRunningSessions } from "@/lib/rpc-manager";

export async function GET() {
  const headers = new Headers({
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    "Connection": "keep-alive",
  });

  const stream = new ReadableStream({
    start(controller) {
      // Send initial snapshot
      controller.enqueue(`data:${JSON.stringify(getRunningRpcSessionIds())}\n\n`);

      // Register listener to push updates
      const unsubscribe = subscribeRunningSessions((ids) => {
        controller.enqueue(`data:${JSON.stringify(ids)}\n\n`);
      });

      // Cleanup when client disconnects
      controller.signal.addEventListener("abort", unsubscribe);
    },
  });

  return new Response(stream, { status: 200, headers });
}

Connection Lifecycle Management

The route leverages the AbortSignal available on the ReadableStream controller to handle client disconnects gracefully. When the browser closes the connection, the abort event fires, triggering the unsubscribe function returned by subscribeRunningSessions() and preventing memory leaks.

Client-Side Integration

The sidebar component consumes this SSE stream through a React hook. The client opens a single persistent HTTP connection via EventSource and updates local state only when messages arrive, eliminating the need for setInterval or repeated fetch() calls.

// Client-side subscription (React hook)
import { useEffect, useState } from "react";

export function useRunningSessions() {
  const [runningIds, setRunningIds] = useState<string[]>([]);

  useEffect(() => {
    const es = new EventSource("/api/agent/running/events");
    es.onmessage = (e) => setRunningIds(JSON.parse(e.data));
    return () => es.close();
  }, []);

  return runningIds;
}

This approach ensures the UI reflects the server state within milliseconds of a change, without the latency inherent in polling intervals.

Summary

• Push-based architecture: The server pushes updates only when the running session set changes, rather than clients polling at fixed intervals.
• Global singleton pattern: Storing listeners on globalThis.__piRunningListeners in lib/rpc-manager.ts ensures the broadcaster survives Next.js hot-reloads during development.
• Delta compression: The notifyRunningChange() function compares JSON snapshots to avoid broadcasting identical state, reducing bandwidth.
• Clean resource management: The SSE route uses controller.signal to automatically unregister listeners when clients disconnect, preventing memory leaks.
• Single persistent connection: One HTTP connection per client handles both initial state and all subsequent updates, minimizing connection overhead compared to repeated polling requests.

Frequently Asked Questions

What is SSE and why does Pi Web use it instead of polling?

Server-Sent Events (SSE) is a browser API that establishes a persistent HTTP connection for unidirectional server-to-client streaming. Pi Web uses SSE instead of polling because it provides lower latency (updates arrive immediately), reduces server load (no redundant requests), and automatically handles connection management through standard HTTP mechanisms. The text/event-stream content type enables the browser to parse incoming messages as discrete events.

How does Pi Web handle Next.js hot-reloads without losing the running state?

The repository stores the listener registry on globalThis.__piRunningListeners, a global object that persists across module reloads in Next.js development mode. When the code reloads, getRunningListeners() retrieves the existing Set rather than creating a new one, ensuring active SSE connections continue receiving broadcasts without interruption.

What triggers the running state broadcast to clients?

The broadcast triggers whenever an AgentSessionWrapper instance detects a change in its isRunning() status. Specifically, session lifecycle events—such as execution start, completion, or cancellation—invoke notifyRunningChange() in lib/rpc-manager.ts. This function then recomputes the full set of running session IDs and notifies all registered SSE listeners if the set differs from the previous snapshot.

How does the server detect when a client disconnects from the SSE stream?

The SSE route in app/api/agent/running/events/route.ts attaches an event listener to controller.signal for the abort event. When the client closes the browser tab, navigates away, or calls EventSource.close(), the underlying HTTP connection terminates, triggering the abort signal. This executes the unsubscribe function returned by subscribeRunningSessions(), removing the closed connection's callback from the global listener Set and freeing associated memory.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}