# Streaming & tool events URL: /docs/agent/concepts/streaming Observe Pi's text output and tool calls in real time using AgentStream and AgentEvent. `agent.prompt()` and `agent.bash()` return an `AgentStream` — an `AsyncIterable`. Each event is a discriminated union that tells you whether Pi is writing text or using a tool. ## AgentEvent [#agentevent] ```ts type AgentEvent = | { type: "text"; text: string } | { type: "tool_start"; toolCallId: string; toolName: string; args: unknown } | { type: "tool_update"; toolCallId: string; toolName: string; partialResult: unknown } | { type: "tool_end"; toolCallId: string; toolName: string; result: unknown; isError: boolean } | { type: "extension_ui"; method: string; params: unknown; isDialog: boolean; requestId?: string } | { type: "permission_request"; requestId: string; tool: string; target: string; title: string } | { type: "permission_resolved"; requestId: string; decision: PermissionDecision } | { type: "auto_retry_start"; attempt: number; maxAttempts: number; delayMs: number; errorMessage: string; } | { type: "auto_retry_end"; success: boolean; attempt: number; finalError?: string } | { type: "agent_start" } | { type: "agent_end"; messages: unknown[] } | { type: "turn_start"; turnIndex: number; timestamp: number } | { type: "turn_end"; turnIndex: number; message: unknown; toolResults: unknown[] } | { type: "message_start"; message: unknown } | { type: "message_update"; message: unknown; delta: unknown } | { type: "message_end"; message: unknown } | { type: "queue_update"; steering: string[]; followUp: string[] } | { type: "compaction_start"; reason: "manual" | "threshold" | "overflow" } | { type: "compaction_end"; reason: "manual" | "threshold" | "overflow"; result: { summary: string; firstKeptEntryId: string; tokensBefore: number; estimatedTokensAfter: number; details: unknown; } | null; aborted: boolean; willRetry: boolean; } | { type: "extension_error"; extensionPath: string; event: string; error: string }; ``` | Event | When it fires | | --------------------- | ---------------------------------------------------------------------------------------------------------------------------------- | | `text` | Pi is writing a text chunk | | `tool_start` | Pi has invoked a tool (e.g. `bash`, `write_file`) | | `tool_update` | Partial output from a long-running tool | | `tool_end` | Tool completed — includes the full result and whether it errored | | `extension_ui` | A Pi extension requested UI interaction (dialog auto-cancelled; forwarded for observability) | | `permission_request` | The [permission gate](/docs/agent/concepts/permissions) paused a tool call for approval — resolve with `agent.resolvePermission()` | | `permission_resolved` | A `permission_request` was answered (by a caller, a batched decision, or the timeout) | | `auto_retry_start` | Pi is retrying after a transient error (429, 5xx) | | `auto_retry_end` | Retry sequence completed — `success` indicates whether it recovered | | `agent_start` | Pi began processing the prompt | | `agent_end` | Pi finished the full agent run (all turns complete); includes all messages | | `turn_start` | A new LLM turn began; includes `turnIndex` and `timestamp` | | `turn_end` | A turn completed with its assistant message and tool results | | `message_start` | A new assistant message began streaming | | `message_update` | Streaming delta from an in-flight message; `delta` is the raw Pi event (text, thinking, tool call delta, etc.) | | `message_end` | An assistant message completed | | `queue_update` | The steering/follow-up queue changed (e.g. after `followUp()`) | | `compaction_start` | Pi began compacting context (manual or automatic) | | `compaction_end` | Context compaction finished; `result` has token counts, `aborted` if it was cancelled | | `extension_error` | A Pi extension threw an error | ## Text-only with textOnly() [#text-only-with-textonly] If you only care about the text output, use the `textOnly()` helper to filter the stream: ```ts import { Alineo, textOnly } from "alineo"; for await (const chunk of textOnly(agent.prompt("Summarise this repo."))) { process.stdout.write(chunk); } ``` `textOnly()` is a thin generator that passes through `text` events and drops everything else. It returns `AsyncIterable` — the same interface as the old `PromptStream`. ## Observing tool calls [#observing-tool-calls] Iterate the raw `AgentStream` to see every tool Pi invokes: ```ts for await (const ev of agent.prompt("Run /workspace/script.py with python3.")) { switch (ev.type) { case "text": process.stdout.write(ev.text); break; case "tool_start": console.log(`\n[tool] ${ev.toolName} args=${JSON.stringify(ev.args)}`); break; case "tool_update": // Partial output from a running tool (e.g. streaming bash stdout) process.stdout.write(`[partial: ${ev.toolName}]`); break; case "tool_end": console.log(`[tool] ${ev.toolName} done isError=${ev.isError}`); break; } } ``` ## Collecting tool events [#collecting-tool-events] You can collect tool events alongside the text response: ```ts const toolEvents: AgentEvent[] = []; let text = ""; for await (const ev of agent.prompt("Write and run a hello world script.")) { if (ev.type === "text") { text += ev.text; } else { toolEvents.push(ev); } } const toolNames = toolEvents .filter((e) => e.type === "tool_start") .map((e) => (e as Extract).toolName); console.log("Tools used:", [...new Set(toolNames)].join(", ")); ``` ## bash() [#bash] `agent.bash()` runs a shell command inside Pi's working context and returns the same `AgentStream` type — but unlike `prompt()`, it's not incrementally streamed. Pi returns bash output synchronously, so the full output arrives as a single `text` event once the command completes; no `tool_start`/`tool_update`/`tool_end` events are emitted for it: ```ts for await (const chunk of textOnly(agent.bash("ls -la /workspace"))) { process.stdout.write(chunk); } ``` ## Lifecycle events [#lifecycle-events] Every `prompt()` call fires a sequence of lifecycle events around the text and tool events. These are useful for measuring per-turn latency, building progress indicators, or capturing the full structured response. ```ts for await (const ev of agent.prompt("Refactor this module.")) { switch (ev.type) { case "agent_start": console.time("agent"); break; case "turn_start": console.log(`turn ${ev.turnIndex} started at ${new Date(ev.timestamp).toISOString()}`); break; case "text": process.stdout.write(ev.text); break; case "turn_end": // ev.message is the full assistant message; ev.toolResults holds tool results from this turn console.log(`\nturn ${ev.turnIndex} done — ${ev.toolResults.length} tool(s) used`); break; case "agent_end": // ev.messages contains every message generated across all turns console.timeEnd("agent"); console.log(`total messages in run: ${ev.messages.length}`); break; } } ``` The event order within a single turn is always: ``` agent_start turn_start (turnIndex: 0) message_start message_update ← repeated for each streaming delta text ← emitted alongside message_update for text deltas message_end tool_start / tool_update / tool_end ← if Pi used tools turn_end turn_start (turnIndex: 1) ← if Pi needed another turn ... agent_end ``` ## Observing thinking and tool-call deltas [#observing-thinking-and-tool-call-deltas] `message_update` carries the raw Pi delta in its `delta` field. In addition to `text_delta` (which the bridge also surfaces as a `text` event), it can carry thinking and tool-call deltas from models that support extended thinking: ```ts for await (const ev of agent.prompt("Solve this step by step.")) { if (ev.type === "message_update") { const delta = ev.delta as { type: string; delta?: string; thinking?: string }; if (delta.type === "thinking_delta") { process.stdout.write(`[thinking] ${delta.thinking ?? ""}`); } // delta.type === "text_delta" is also emitted as a "text" event — no need to handle it twice } if (ev.type === "text") { process.stdout.write(ev.text); } } ``` ## Compaction events [#compaction-events] Auto-compaction fires between turns when Pi's context fills past a threshold. The `compaction_start` and `compaction_end` events let you surface this to the user instead of silently pausing: ```ts for await (const ev of agent.prompt("Analyse the entire repo.")) { switch (ev.type) { case "text": process.stdout.write(ev.text); break; case "compaction_start": console.log(`\n[compacting — ${ev.reason}]`); break; case "compaction_end": if (ev.result) { const saved = ev.result.tokensBefore - ev.result.estimatedTokensAfter; console.log(`[compaction done — freed ~${saved} tokens]`); } if (ev.aborted) { console.warn("[compaction aborted]"); } break; } } ``` `reason` is one of `"manual"` (triggered by `agent.compact()`), `"threshold"` (auto, approaching context limit), or `"overflow"` (auto, context was full). ## Queue events [#queue-events] `queue_update` fires when the steering or follow-up queue changes — for example, immediately after `agent.followUp()` is called while a prompt is in flight: ```ts const stream = agent.prompt("Write a long analysis..."); setTimeout(async () => { await agent.followUp("Now summarize that in one paragraph."); }, 1000); for await (const ev of stream) { if (ev.type === "text") process.stdout.write(ev.text); if (ev.type === "queue_update") { console.log(`\n[queued follow-ups: ${ev.followUp.length}]`); } } ```