Pi Web Agent SSE Events: Complete Reference for Real-Time Streaming

Pi Web agents emit nine distinct Server-Sent Events (SSE) through a single GET /api/agent/[id]/events endpoint to communicate run lifecycle, compaction status, tool execution, and streaming message chunks to the front-end.

The Pi Web repository (agegr/pi-web) implements a real-time event streaming system that keeps the UI synchronized with agent execution. This guide catalogs every SSE event type, its payload structure, and the source files where each is defined.

Complete List of SSE Event Types

All events share a common JSON structure with a type discriminator field. The following events are defined in lib/agent-event-wire.ts unless otherwise noted.

Agent Lifecycle Events

These events bracket a single agent run from creation to cleanup.

Event	Trigger	Payload
agent_start	New AgentSession created or forked; run officially begins	{runId: string, sessionId: string, model: string, thinkingLevel?: number}
prompt_done	LLM completion finishes; UI stops "thinking" spinner	{runId: string, tokenCount: number, finishReason: string}
agent_end	Run terminates (completion, error, or user cancellation)	{runId: string, success: boolean, error?: string}
agent_settled	Post-run cleanup complete (tool result processing, etc.)	{runId: string}

Compaction Events

Pi Web automatically compresses conversation history to manage context window limits. Two naming conventions exist:

Event	Version	Trigger	Payload
compaction_start	v2.0+	Session compaction about to begin	{runId: string, startTime: number}
compaction_end	v2.0+	Compaction finished; message list updated	{runId: string, summary: string, tokensSaved: number}
auto_compaction_start	legacy	Backward-compatible alias	—
auto_compaction_end	legacy	Backward-compatible alias	—

Tool Execution Events

Event	Trigger	Payload
tool_call	Model requests tool execution	{runId: string, toolName: string, arguments: object, callId: string}
tool_result	Tool execution completes and returns to model	{runId: string, callId: string, result: any}

Streaming Message Events

Defined in lib/agent-event-stream.ts:

Event	Trigger	Payload
message	Token chunk streamed during generation	{runId: string, role: "assistant" | "user", content: string, isPartial: boolean}

Implementation Architecture

The SSE pipeline spans four interconnected layers.

1. Event Definition Layer

lib/agent-event-wire.ts centralizes type definitions and serialization logic. This file exports TypeScript discriminated unions for type-safe event handling and the writeEvent(res, event) helper that formats JSON with SSE protocol newlines.

2. Streaming Layer

lib/agent-event-stream.ts manages low-level token streaming. It transforms raw LLM output into message events with isPartial: true for intermediate chunks and isPartial: false for the final token.

3. Connection Management Layer

lib/agent-event-connection.ts implements:

• Connection lifecycle (open, heartbeat, close)
• Automatic reconnection with exponential backoff
• Run-ID validation to ignore stale events from previous sessions

4. Client Integration Layer

hooks/useAgentSession.ts exposes the React hook consumed by UI components. It opens the SSE connection, dispatches events to state reducers, and handles cleanup on unmount.

API Route Entry Point

The Next.js App Router endpoint at app/api/agent/[id]/events/route.ts binds the HTTP response stream to the event wire:

// app/api/agent/[id]/events/route.ts
import { NextRequest } from "next/server";
import { createEventStream } from "@/lib/agent-event-connection";

export async function GET(
  request: NextRequest,
  { params }: { params: { id: string } }
) {
  const sessionId = params.id;
  const stream = await createEventStream(sessionId);
  
  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      "Connection": "keep-alive",
    },
  });
}

Client-Side Consumption Examples

Basic Event Listener Hook

// hooks/useAgentEvents.ts
import { useEffect, useRef } from "react";

type AgentEvent =
  | { type: "agent_start"; runId: string; model: string }
  | { type: "prompt_done"; runId: string; tokenCount: number }
  | { type: "compaction_start"; runId: string; startTime: number }
  | { type: "compaction_end"; runId: string; summary: string; tokensSaved: number }
  | { type: "message"; runId: string; role: "assistant"; content: string; isPartial: boolean };

export function useAgentEvents(
  sessionId: string,
  onEvent: (event: AgentEvent) => void
) {
  const eventSourceRef = useRef<EventSource | null>(null);

  useEffect(() => {
    const url = `/api/agent/${sessionId}/events`;
    const es = new EventSource(url);
    eventSourceRef.current = es;

    es.onmessage = (e) => {
      try {
        const parsed: AgentEvent = JSON.parse(e.data);
        onEvent(parsed);
      } catch (err) {
        console.error("Failed to parse SSE payload:", err);
      }
    };

    es.onerror = () => {
      // Connection will auto-retry per browser EventSource behavior
      console.warn("SSE connection error");
    };

    return () => {
      es.close();
      eventSourceRef.current = null;
    };
  }, [sessionId, onEvent]);
}

Full Session Integration with State Machine

// components/AgentSession.tsx
import { useReducer } from "react";
import { useAgentEvents } from "@/hooks/useAgentEvents";

type State = {
  runId: string | null;
  status: "idle" | "running" | "compacting" | "done";
  messages: Array<{ role: string; content: string }>;
  tokensUsed: number;
};

function reducer(state: State, event: AgentEvent): State {
  switch (event.type) {
    case "agent_start":
      return {
        ...state,
        runId: event.runId,
        status: "running",
        messages: [],
      };
    
    case "message":
      if (event.isPartial && state.messages.length > 0) {
        // Append to last assistant message
        const last = state.messages.at(-1)!;
        return {
          ...state,
          messages: [
            ...state.messages.slice(0, -1),
            { ...last, content: last.content + event.content },
          ],
        };
      }
      return {
        ...state,
        messages: [...state.messages, { role: event.role, content: event.content }],
      };
    
    case "compaction_start":
      return { ...state, status: "compacting" };
    
    case "compaction_end":
      return { ...state, status: "running" };
    
    case "prompt_done":
      return { ...state, status: "done", tokensUsed: event.tokenCount };
    
    default:
      return state;
  }
}

export function AgentSession({ sessionId }: { sessionId: string }) {
  const [state, dispatch] = useReducer(reducer, {
    runId: null,
    status: "idle",
    messages: [],
    tokensUsed: 0,
  });

  useAgentEvents(sessionId, dispatch);

  return (
    <div>
      <div>Status: {state.status}</div>
      <div>Tokens: {state.tokensUsed}</div>
      {state.messages.map((m, i) => (
        <p key={i} className={m.role}>{m.content}</p>
      ))}
    </div>
  );
}

Sending Messages to Trigger Events

// lib/agent-client.ts
export async function sendUserMessage(
  sessionId: string,
  content: string
): Promise<void> {
  const response = await fetch(`/api/agent/${sessionId}`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      cmd: "send",
      payload: { role: "user", content },
    }),
  });

  if (!response.ok) {
    throw new Error(`Failed to send message: ${response.statusText}`);
  }
  // SSE events will follow: agent_start → message chunks → prompt_done → agent_end
}

File Reference Map

File Path	Responsibility
lib/agent-event-wire.ts	Event type definitions, serialization, writeEvent() helper
lib/agent-event-stream.ts	Token streaming, message event generation
lib/agent-event-connection.ts	Connection lifecycle, reconnection, run-ID validation
hooks/useAgentSession.ts	React hook for UI state synchronization
app/api/agent/[id]/events/route.ts	HTTP endpoint wiring

Summary

• Pi Web agent SSE events travel through a single endpoint with nine distinct event types covering lifecycle, compaction, tools, and streaming.
• Modern compaction events use compaction_start/compaction_end; legacy names remain for backward compatibility.
• Type safety is enforced through discriminated unions in lib/agent-event-wire.ts.
• Resilience is built into lib/agent-event-connection.ts with automatic reconnection and stale event filtering.
• Client integration is simplified by hooks/useAgentSession.ts which manages EventSource lifecycle and state updates.

Frequently Asked Questions

How do I distinguish between a running agent and one that is compacting?

Check the type field of incoming SSE events. When compaction_start or auto_compaction_start arrives, the agent is compacting conversation history. The status returns to active generation when compaction_end or auto_compaction_end is received.

Can I receive events from multiple concurrent runs on the same session?

No. Pi Web's lib/agent-event-connection.ts validates that the runId in each event matches the current active run. Events from stale runs are silently discarded to prevent UI state corruption.

What happens if my SSE connection drops during a long generation?

The browser's native EventSource automatically attempts reconnection with exponential backoff. hooks/useAgentSession.ts additionally re-fetches session state on visibilitychange events to restore synchronization after extended disconnections.

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