# How Pi Web Implements Running State SSE Broadcast to Avoid Polling

> Discover how Pi Web uses SSE broadcast for real-time running state updates, eliminating inefficient client polling. Learn about this efficient implementation.

- Repository: [Alex Yang/pi-web](https://github.com/agegr/pi-web)
- Tags: internals
- Published: 2026-07-25

---

**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`](https://github.com/agegr/pi-web/blob/main/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.

```typescript
// 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.

```typescript
// 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`](https://github.com/agegr/pi-web/blob/main/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.

```typescript
// 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.

```tsx
// 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`](https://github.com/agegr/pi-web/blob/main/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`](https://github.com/agegr/pi-web/blob/main/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`](https://github.com/agegr/pi-web/blob/main/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.