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

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, 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:

// 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:

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

The sendAgentCommand utility, defined in 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, 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:

// 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:

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 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:

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 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
• 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 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 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.

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