# How Pi Web's Architecture Handles User Messages: A Complete Pipeline Guide

> Explore Pi Web's architecture and discover how user messages flow through a multi-layer pipeline from UI to AI responses using SSE and optimistic UI.

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

---

**Pi Web processes user messages through a multi-layer pipeline that moves from the React UI via Next.js API routes into an `AgentSession` wrapper, streams AI responses via Server-Sent Events (SSE), and updates the chat interface using optimistic UI patterns.**

The **agegr/pi-web** repository implements a Next.js application that bridges user chat interactions with an underlying AI agent. Understanding how Pi Web's architecture handles user messages reveals a sophisticated separation between the presentation layer and agent logic, utilizing a thin RPC layer to manage complex streaming interactions.

## The Message Pipeline Overview

When a user submits a message, the request traverses five distinct architectural layers:

1. **UI Layer** – React components capture input and manage local state
2. **Hook Layer** – `useAgentSession` orchestrates optimistic updates and API communication
3. **API Layer** – Next.js routes handle HTTP commands and SSE streaming
4. **RPC Layer** – `AgentSessionWrapper` manages the Pi SDK session lifecycle
5. **Streaming Layer** – Server-Sent Events deliver real-time updates back to the browser

This design isolates the UI from agent complexity, allowing the frontend to work with high-level commands while the backend handles model inference and state management.

## Step 1: Capturing Input and Optimistic UI Updates

The process begins in [`components/ChatInput.tsx`](https://github.com/agegr/pi-web/blob/main/components/ChatInput.tsx), which calls `useAgentSession.handleSend` when the user submits a message.

The hook immediately creates a **user-role `AgentMessage`** and appends it to the local message list before any network request occurs. This **optimistic UI** pattern ensures the interface feels responsive:

```typescript
// In hooks/useAgentSession.ts
const userMsg: AgentMessage = {
  role: "user",
  content: message,
  timestamp: Date.now(),
};
setMessages(prev => [...prev, userMsg]);

```

The hook stores a temporary key for this optimistic update, allowing the UI to reconcile states if the network request fails.

## Step 2: Dispatching Commands to the API Layer

After updating the local state, `handleSend` invokes `sendAgentCommand` to communicate with the backend. This function constructs a `POST` request to `/api/agent/[id]` containing a typed command object:

```typescript
await sendAgentCommand(sessionId, {
  type: "prompt",
  message,
  // optional images parameter for multimodal input
});

```

The `sendAgentCommand` utility, defined in [`lib/agent-client.ts`](https://github.com/agegr/pi-web/blob/main/lib/agent-client.ts), provides a thin typed wrapper around `fetch` that standardizes communication with the agent API routes.

## Step 3: Processing Commands in the RPC Manager

The Next.js route at `app/api/agent/[id]/route.ts` receives the command and forwards it to the `AgentSessionWrapper` via its `send` method. This wrapper, implemented in [`lib/rpc-manager.ts`](https://github.com/agegr/pi-web/blob/main/lib/rpc-manager.ts), serves as the bridge between the HTTP API and the Pi SDK's `AgentSessionLike` interface.

When processing a `"prompt"` command, the wrapper performs validation and delegates to the underlying session:

```typescript
// In lib/rpc-manager.ts
case "prompt": {
  if (this.inner.isBashRunning) {
    throw new Error("Cannot send a prompt while a shell command is running");
  }
  this.promptRunning = true;
  this.inner.prompt(command.message as string, {
    source: "rpc",
  }).then(() => {
    this.promptRunning = false;
    this.emit({ type: "prompt_done" });
  }).catch(error => {
    this.promptRunning = false;
    this.emit({
      type: "prompt_error",
      errorMessage: error instanceof Error ? error.message : String(error),
    });
  });
  return null;
}

```

The `this.inner` property represents the actual `AgentSession` from the Pi SDK. Its `prompt` method initiates model inference and prepares to stream partial results as discrete events.

## Step 4: Streaming Responses via SSE

Once the model begins generating content, the `AgentSessionWrapper` subscribes to the underlying session's events using `inner.subscribe`. The wrapper rebroadcasts these through its own `emit` method, which triggers `notifyRunningChange` to update sidebar badges indicating live activity.

The server-side event stream endpoint at `app/api/agent/[id]/events/route.ts` pipes these events to the browser using **Server-Sent Events (SSE)**. On the client side, `useAgentSession.connectEvents` establishes the connection:

```typescript
const es = new EventSource(`/api/agent/${encodeURIComponent(sid)}/events`);
es.onmessage = e => {
  const event = JSON.parse(e.data);
  if (event.type === "connected") settle("connected");
  handleAgentEventRef.current?.(event);
};

```

The wrapper's `start` method manages subscription registration and maintains a keep-alive timer to prevent connection timeouts during long-running inference tasks.

## Step 5: Updating the UI with Streaming Data

The `handleAgentEvent` callback in [`useAgentSession.ts`](https://github.com/agegr/pi-web/blob/main/useAgentSession.ts) processes incoming events and updates the React state accordingly. The system handles three primary message event types:

- **`message_start`** – Initializes a new assistant message in the streaming state
- **`message_update`** – Appends incremental content to the partial message
- **`message_end`** – Finalizes the assistant message and resets streaming state

For incremental updates, the hook dispatches to a streaming reducer:

```typescript
case "message_update": {
  if (!agentRunningRef.current) break;
  const msg = event.message as Partial<AgentMessage>;
  dispatch({ type: "update", message: normalizeToolCalls(msg) });
  break;
}

```

When `"agent_end"` arrives, the hook clears running flags, executes the `onAgentEnd` callback, and refreshes the full session data via `loadSession` to ensure consistency between the local state and server-side session file.

## Session Forking and Branching

Pi Web's architecture also supports conversation branching through the **fork** command. When the user clicks the Fork button, the UI calls `handleFork`, which sends a `"fork"` command to the same API endpoint.

The [`rpc-manager.ts`](https://github.com/agegr/pi-web/blob/main/rpc-manager.ts) implementation handles this by creating a new `.jsonl` session file (or a new empty session before the first message), destroying the current wrapper instance, and returning the new session ID. The UI then navigates to the new session, allowing users to explore alternative conversation paths without losing context.

## Summary

- **Pi Web isolates UI and agent logic** through a clean RPC layer implemented in [`lib/rpc-manager.ts`](https://github.com/agegr/pi-web/blob/main/lib/rpc-manager.ts)
- **Optimistic updates** provide immediate feedback while `useAgentSession` manages the underlying asynchronous operations
- **Server-Sent Events** enable real-time streaming of AI responses without WebSocket complexity
- **Event-driven architecture** uses typed events (`message_start`, `message_update`, `message_end`, `agent_end`) to synchronize state across layers
- **Session forking** creates new conversation branches by managing discrete `.jsonl` session files on the server

## Frequently Asked Questions

### How does Pi Web prevent message sending during active shell commands?

The `AgentSessionWrapper` in [`lib/rpc-manager.ts`](https://github.com/agegr/pi-web/blob/main/lib/rpc-manager.ts) maintains an `isBashRunning` flag on the underlying session. When processing a `"prompt"` command, the wrapper explicitly checks this flag and throws an error if a shell command is active, preventing input corruption or context mixing.

### What happens if the SSE connection drops during stream generation?

The `AgentSessionWrapper.start` method implements a keep-alive timer that maintains the connection during long inference tasks. If the browser disconnects, the `EventSource` in `useAgentSession.connectEvents` can automatically reconnect via the browser's built-in SSE reconnection logic, and the server continues emitting events through the persistent wrapper instance.

### How does the optimistic UI handle send failures?

The `handleSend` function in [`useAgentSession.ts`](https://github.com/agegr/pi-web/blob/main/useAgentSession.ts) adds the user message to local state immediately using `setMessages`, but tracks the temporary key. If the `sendAgentCommand` promise rejects, the error handling logic can roll back the optimistic addition by filtering out messages matching that temporary key, ensuring the UI reflects the actual server state.

### Can multiple clients connect to the same agent session simultaneously?

While the architecture supports multiple SSE connections to `/api/agent/[id]/events`, the `AgentSessionWrapper` maintains a single `inner` session instance. All connected clients receive the same event stream through the wrapper's `emit` method, though race conditions when sending commands are mitigated by the RPC layer's sequential command processing.