# How SSE Event Streaming Powers Real-Time Updates in Pi-Web's ChatWindow

> Discover how SSE event streaming powers real-time updates in Pi-Web's ChatWindow. Learn how a Next.js backend pushes JSON events to a React frontend for seamless chat experiences.

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

---

**Pi-Web's ChatWindow receives live updates through a Server-Sent Events (SSE) stream that pushes JSON-encoded events from the Next.js backend to the React frontend, where a custom hook processes message fragments and UI state changes in real time.**

The `agegr/pi-web` repository implements a robust real-time chat interface using **SSE event streaming** to deliver low-latency updates without WebSocket complexity. This architecture allows the ChatWindow component to render streaming AI responses, tool execution progress, and session status changes as they occur on the backend. By leveraging a standard HTTP connection with a `ReadableStream` on the server and the native `EventSource` API on the client, pi-web achieves reliable, server-push functionality within a modern React framework.

## The SSE Architecture: From ReadableStream to EventSource

### Backend Endpoint Implementation

In `app/api/agent/[id]/events/route.ts`, the backend establishes a **Server-Sent Events** endpoint by creating a `ReadableStream` that encapsulates the session's event emitter. The stream encodes each event as a JSON payload prefixed with `data: ` and suffixed with double newlines, conforming to the SSE protocol.

The implementation initializes the stream with a `connected` event containing the session ID, then subscribes to the `AgentSession` event emitter via `session.onEvent()`. To prevent proxy timeouts during idle periods, the code sends a heartbeat message (`:\n\n`) every 30 seconds using `setInterval`. When the client disconnects, the `abort` signal triggers cleanup logic that clears the heartbeat interval, unsubscribes from the session, and closes the controller.

```ts
// app/api/agent/[id]/events/route.ts
const stream = new ReadableStream({
  start(controller) {
    const encode = (data) => {
      const text = `data: ${JSON.stringify(data)}\n\n`;
      controller.enqueue(new TextEncoder().encode(text));
    };
    encode({ type: "connected", sessionId: id });
    const unsubscribe = session.onEvent((event) => encode(event));
    const heartbeat = setInterval(() => {
      controller.enqueue(new TextEncoder().encode(":\n\n"));
    }, 30_000);
    const cleanup = () => {
      clearInterval(heartbeat);
      unsubscribe();
      controller.close();
    };
    req.signal?.addEventListener("abort", cleanup);
  },
});

```

### Client Connection Management

On the frontend, the `useAgentSession` hook in [`hooks/useAgentSession.ts`](https://github.com/agegr/pi-web/blob/main/hooks/useAgentSession.ts) manages the SSE connection through the `connectEvents` function. This factory establishes an `EventSource` instance pointing to `/api/agent/${sid}/events` and returns a Promise that resolves once the initial `connected` event is received or rejects on timeout.

The implementation stores the `EventSource` reference in a ref for lifecycle management and attaches an `onmessage` handler that parses JSON payloads and routes them to `handleAgentEvent`. Error handling leverages the browser's native auto-reconnect behavior while implementing a custom timeout using `EVENT_STREAM_CONNECT_TIMEOUT_MS` to detect permanent connection failures.

```ts
// hooks/useAgentSession.ts
const connectEvents = useCallback((sid: string) => {
  const es = new EventSource(`/api/agent/${encodeURIComponent(sid)}/events`);
  eventSourceRef.current = es;
  return new Promise<EventStreamConnectionResult>((resolve) => {
    const timeout = setTimeout(() => resolve({ status: "timeout", source: es }), EVENT_STREAM_CONNECT_TIMEOUT_MS);
    es.onmessage = (e) => {
      const event = JSON.parse(e.data);
      if (event.type === "connected") resolve({ status: "connected", source: es });
      handleAgentEventRef.current?.(event);
    };
    es.onerror = () => { /* auto‑reconnect logic */ };
  });
}, []);

```

## Processing Real-Time Events in the React UI

### Event Dispatch and State Updates

Once connected, the `handleAgentEvent` callback processes incoming SSE payloads to mutate React state. Located in [`hooks/useAgentSession.ts`](https://github.com/agegr/pi-web/blob/main/hooks/useAgentSession.ts), this function uses a switch statement to handle distinct event types: `agent_start` sets loading flags, `message_update` dispatches partial content to a streaming reducer, and `message_end` persists completed messages to the chat history.

The reducer manages `streamState`, tracking whether a message is currently streaming and accumulating partial content from `message_update` events. This state also handles auxiliary events like `tool_execution_start` and `auto_compaction_end`, updating flags such as `isCompacting` to drive progress indicators elsewhere in the UI.

```ts
// hooks/useAgentSession.ts
const handleAgentEvent = useCallback((event) => {
  switch (event.type) {
    case "agent_start":
      setAgentRunning(true);
      dispatch({ type: "start" });
      break;
    case "message_update":
      dispatch({ type: "update", message: normalizeToolCalls(event.message) });
      break;
    case "message_end":
      setMessages((prev) => [...prev, normalizeToolCalls(event.message)]);
      dispatch({ type: "reset" });
      break;
    // …other cases for tools, compaction, etc.
  }
}, []);

```

### Rendering Streaming Messages in ChatWindow

The `ChatWindow` component consumes the `streamState` exposed by `useAgentSession` to render real-time message bubbles. When `streamState.isStreaming` is true and `streamingMessage` contains partial content, the component renders a `MessageView` with the `isStreaming` prop enabled, creating the visual effect of live typing.

Once the backend emits an `agent_end` or `prompt_done` event, the streaming state resets, and the final message moves from the transient `streamingMessage` state to the persistent `messages` array. This architectural separation ensures that partial content never pollutes the permanent chat history while providing immediate visual feedback.

```tsx
// components/ChatWindow.tsx (excerpt)
{streamState.isStreaming && streamState.streamingMessage && (
  <MessageView
    message={streamState.streamingMessage as AgentMessage}
    isStreaming
    modelNames={modelNames}
    cwd={messageCwd}
    onOpenFile={onOpenFile}
  />
)}

```

## Reliability Mechanisms for Production SSE

### Heartbeat and Connection Keep-Alive

To prevent intermediary proxies from terminating idle connections, the SSE endpoint emits a heartbeat comment (`:\n\n`) every 30 seconds. This keeps the TCP connection alive during long-running AI inference operations that may not generate immediate data events.

### Abort Handling and Resource Cleanup

The backend listens for the `abort` signal on the incoming request via `req.signal.addEventListener("abort", cleanup)`. When the client navigates away or closes the tab, this triggers immediate cleanup of the heartbeat interval and session subscription, preventing memory leaks and orphaned connections.

### Automatic Reconnection and State Reconciliation

While the browser's native `EventSource` automatically attempts to reconnect on transient network failures, pi-web implements additional safeguards. A periodic reconciliation effect (using `AGENT_STATE_RECONCILE_MS`) polls the server state when the tab returns from the background, detecting any missed SSE events that occurred while the browser throttled JavaScript execution.

### Optimistic UI Updates

To minimize perceived latency, the UI applies **optimistic updates** by immediately appending user messages to the `messages` array before server confirmation. Subsequent SSE events then confirm or augment this state, ensuring the chat interface feels responsive even before the AI begins its response stream.

## Summary

- **Backend Streaming**: The Next.js API route in `app/api/agent/[id]/events/route.ts` creates a `ReadableStream` that encodes session events as SSE-compliant JSON payloads, complete with 30-second heartbeats.
- **Client Connection**: The `useAgentSession` hook manages the `EventSource` lifecycle, resolving connection promises on the initial `connected` event and routing all subsequent messages to the event handler.
- **State Synchronization**: The `handleAgentEvent` callback dispatches actions to a `streamState` reducer, enabling the `ChatWindow` component to render live typing indicators and progress flags.
- **Production Resilience**: Abort signal handling, automatic reconnection logic, and periodic state reconciliation ensure the SSE pipeline remains stable across network interruptions and browser backgrounding.
- **Optimistic Rendering**: User messages appear instantly in the UI while the SSE stream delivers AI responses incrementally, creating a fluid conversational experience.

## Frequently Asked Questions

### What event types does the pi-web SSE stream emit?

The SSE stream emits structured events including `connected`, `message_start`, `message_update`, `message_end`, `agent_start`, `agent_end`, and `tool_execution_*` variants. Each event is JSON-encoded and prefixed with the `data:` field according to the Server-Sent Events specification, allowing the client to distinguish between streaming content, status changes, and tool execution progress.

### How does pi-web handle network interruptions in SSE connections?

The implementation relies on the browser's native `EventSource` auto-reconnect behavior for transient failures, supplemented by a custom timeout mechanism in `connectEvents` that surfaces permanent connection issues. Additionally, a background reconciliation effect using `AGENT_STATE_RECONCILE_MS` detects missed events when tabs return from the background and forces a state sync with the server.

### Why does the backend send heartbeat messages every 30 seconds?

The heartbeat message—sent as an SSE comment line (`:\n\n`) every 30 seconds—prevents HTTP proxies and load balancers from terminating idle connections. This is essential for maintaining the stream during long-running AI operations that may not produce immediate data events, ensuring the client receives the complete response when generation completes.

### How is the streaming message bubble rendered in real time?

The `handleAgentEvent` callback dispatches `message_update` events to a reducer that accumulates partial content in `streamState.streamingMessage`. The `ChatWindow` component subscribes to this state and conditionally renders a `MessageView` with `isStreaming` enabled whenever partial content exists, creating the visual effect of live typing that updates instantly as new SSE chunks arrive.