The agent loop

pi-agent-core is about 2,500 lines that turn Chapter 1's five-line pseudocode into a production loop: two nested loops that drain steering and follow-up queues, a uniform event stream for UIs, tool calls that are validated, hooked, executed in parallel and reported in a stable order, and cancellation that flows through every await. agent-loop.ts holds the pure loop; agent.ts wraps it in a stateful Agent class.

Learning goals. Trace the event sequence of a run with tool calls; explain the inner and outer loops and the difference between steering and follow-up messages; explain how parallel tool execution keeps result order deterministic; describe how abort propagates; list the extension points (convertToLlm, transformContext, beforeToolCall, afterToolCall, prepareRequest, prepareNextTurn, finishTurn) and when each runs.

6.1 Agent messages: the model's messages plus your own#

The loop works on AgentMessage, not on pi-ai's Message:

export interface CustomAgentMessages {
	// Empty by default - apps extend via declaration merging
}
export type AgentMessage = Message | CustomAgentMessages[keyof CustomAgentMessages];

An application adds its own message kinds (a UI notification, a compaction summary, an extension's custom record) through declaration merging: TypeScript lets you re-open an interface in another module and add members, so declare module "@earendil-works/pi-agent-core" { interface CustomAgentMessages { notification: NotificationMessage } } widens the union everywhere without changing the library. The module header states the boundary rule: "Agent loop that works with AgentMessage throughout. Transforms to Message[] only at the LLM call boundary." Two configurable functions do that transformation before every request:

• transformContext(messages, signal) works on agent messages ("Context window management (pruning old messages)", "Injecting context from external sources").
• convertToLlm(messages) maps each agent message to zero or more LLM messages and drops UI-only ones. The default keeps only the four standard roles.

Both carry the contract "must not throw or reject. Return a safe fallback value instead." The loop is written so that every failure becomes data on the event stream (Chapter 3's rule, one level up); a hook that throws would break the event sequence.

6.2 Events#

The loop reports progress as an AgentEvent union with three nested lifecycles:

export type AgentEvent =
	| { type: "agent_start" }
	| { type: "agent_end"; messages: AgentMessage[] }
	| { type: "turn_start" }
	| { type: "turn_end"; message: AgentMessage; toolResults: ToolResultMessage[] }
	| { type: "message_start"; message: AgentMessage }
	| { type: "message_update"; message: AgentMessage; assistantMessageEvent: AssistantMessageEvent }
	| { type: "message_end"; message: AgentMessage }
	| { type: "tool_execution_start"; toolCallId: string; toolName: string; args: any }
	| { type: "tool_execution_update"; toolCallId: string; toolName: string; args: any; partialResult: any }
	| { type: "tool_execution_end"; toolCallId: string; toolName: string; result: any; isError: boolean; durationMs?: number };

A run with one tool call produces, in order:

agent_start
turn_start
  message_start/end (user prompt)
  message_start (assistant, partial)
  message_update × n (text_delta, toolcall_delta, …)
  message_end (assistant, stopReason toolUse)
  tool_execution_start → tool_execution_update × k → tool_execution_end
  message_start/end (toolResult)
turn_end
turn_start
  message_start … message_end (assistant, stopReason stop)
turn_end
agent_end

A turn is one assistant response plus the results of its tool calls. Every message — user, assistant, tool result, injected system message — gets a message_start/message_end pair, so a UI or a persistence layer can treat all messages uniformly. Only assistant messages get message_update, which wraps the underlying pi-ai event. Each emitted message is a shallow copy ({ ...partialMessage }) so subscribers do not observe later mutations.

The loop does not push events into a stream directly; it calls an injected emit: AgentEventSink, and awaits it. agentLoop() wires emit to an EventStream<AgentEvent, AgentMessage[]> for callers that want to iterate; the Agent class wires it to its own reducer and listeners. Because emit is awaited, a slow subscriber slows the loop: this is deliberate backpressure — a session writer can finish persisting a message before the next one is produced.

6.3 The two loops#

runLoop() is the heart of the package. Stripped to its control flow:

let pendingMessages = (await config.getSteeringMessages?.()) || [];
while (true) {                                   // outer: follow-ups
	let hasMoreToolCalls = true;
	while (hasMoreToolCalls || pendingMessages.length > 0) {   // inner: turns
		// (from the second turn) prepareNextTurn, poll steering again, emit turn_start
		// append pending messages (with message_start/end), then prepareRequest
		const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFunction);
		if (message.stopReason === "error" || message.stopReason === "aborted") {
			// finishTurn, turn_end, agent_end; return
		}
		// execute tool calls (or fail them all if stopReason is "length")
		// finishTurn decides end / continue; emit turn_end
		pendingMessages = (await config.getSteeringMessages?.()) || [];
	}
	const followUpMessages = (await config.getFollowUpMessages?.()) || [];
	if (followUpMessages.length > 0) { pendingMessages = followUpMessages; continue; }
	if (explicitContinuation) { explicitContinuation = false; continue; }
	break;
}
await emit({ type: "agent_end", messages: newMessages });

The inner loop runs turns while there are tool results to answer or steering messages to inject. The outer loop runs when the agent would otherwise stop, checking for follow-ups. This gives the two queue semantics their meaning:

• Steering (agent.steer(msg)): "Queue a message to be injected after the current assistant turn finishes." The model sees it at the next request, while it is still working. Tool calls already requested are not skipped.
• Follow-up (agent.followUp(msg)): "Queue a message to run only after the agent would otherwise stop." Useful for a backlog of independent requests.

Each queue has a QueueMode: "all" drains everything at a drain point, "one-at-a-time" (the default for both) drains only the oldest, leaving the rest for later points. Note the careful re-poll after prepareNextTurn: "Preparation can be long-running (for example, compaction). Pick up steering queued while it ran. Only poll again if the earlier poll returned nothing; otherwise one-at-a-time mode would deliver two messages in this turn."

Termination needs care because the loop must never stop with a tool call unanswered and never spin forever. The exits are: an error or aborted response ("hard exits"); finishTurn returning { action: "end" }; a tool batch whose every result sets terminate: true; or no tool calls, no steering, no follow-ups and no explicit continuation.

runLoop · inner turn loop and outer follow-up loop

The inner loop repeats while tool results or steering messages need a response; the outer loop restarts it only when a follow-up is queued. Error and abort exit immediately.

6.4 Streaming one response#

streamAssistantResponse() is the boundary with pi-ai. It applies transformContext, then convertToLlm, then normalizeContext, resolves the API key per request through getApiKey ("important for expiring tokens" — a long tool phase can outlive an OAuth token), and calls the injected streamFunction. As events arrive it maintains the invariant that the last element of context.messages is the partial assistant message: on start it pushes event.partial; on each delta it replaces the last element; on done/error it replaces it with the final message from response.result().

The stream function is injected (StreamFn), defaulting to getDefaultStreamFn(). That single seam lets the same loop run against pi-ai directly, against a proxy server, or against the faux provider in tests.

6.5 Executing tool calls#

Tool calls go through a fixed pipeline, implemented once and reused by both execution modes and by runToolCall() (which lets a tool call other tools "so the hooks (for example permission checks) apply to those calls too").

1. Lookup. Unknown name → immediate error result "Tool X not found".
2. Prepare. Optional tool.prepareArguments(args), a compatibility shim for old argument shapes.
3. Validate. validateToolArguments() from pi-ai (Chapter 3). Failure → error result carrying the validation message.
4. beforeToolCall. May return { block: true, reason } → error result with the reason. This is where an extension implements permission prompts.
5. Execute. tool.execute(toolCallId, args, signal, onUpdate), timed with the monotonic clock. A thrown error becomes an error result. Updates are accepted only while the tool runs ("Calls made after the tool promise settles are ignored"), and all update emissions are awaited before finishing, so tool_execution_update never arrives after tool_execution_end.
6. afterToolCall. May override content, details, isError, usage or terminate, field by field with no deep merge.

The preparation step returns a small discriminated union, { kind: "immediate", result, isError } or { kind: "prepared", toolCall, tool, args }. "Immediate" outcomes skip execution. This separates deciding from doing, which is exactly what parallel execution needs.

Parallel mode (the default) prepares calls one by one — so validation and permission hooks run in a deterministic order and a user is never shown two permission prompts at once — and turns each prepared call into a thunk (a zero-argument function that will run it). Then:

const orderedFinalizedCalls = await Promise.all(
	finalizedCalls.map((entry) => (typeof entry === "function" ? entry() : Promise.resolve(entry))),
);
const messages: ToolResultMessage[] = [];
for (const finalized of orderedFinalizedCalls) {
	const toolResultMessage = createToolResultMessage(finalized);
	await emitToolResultMessage(toolResultMessage, emit);
	messages.push(toolResultMessage);
}

Promise.all runs the thunks concurrently and returns results in input order, regardless of completion order. Hence the documented split: "tool_execution_end is emitted in tool completion order after each tool is finalized, while tool-result message artifacts are emitted later in assistant source order." Live UI feedback is as fast as possible; the transcript is deterministic, which matters for caching and replay.

Sequential mode runs prepare-execute-finalize for each call before the next. It is used when the config says so or when any tool in the batch declares executionMode: "sequential" — a per-tool escape hatch for tools that cannot safely overlap.

One special case shows defensive thinking. If the assistant message stopped with length, the output was cut off, and "streamed tool-call arguments are finalized with a best-effort JSON salvage parser, so a truncated message can yield tool calls whose arguments parse and validate but are silently incomplete." failToolCallsFromTruncatedMessage() therefore fails every call with an instruction to re-issue it. A write tool executed with a truncated content argument would silently corrupt a file.

6.6 Tool changes as transcript events#

context.tools is what the runtime can execute; system messages in the transcript declare what the model may call. Before each request declareToolChanges() compares the two (getToolStateChanges(getCurrentTools(…), context.tools.map(toToolDeclaration))) and, if they differ, inserts a system message with toolsAdded/toolsRemoved. The transcript thus records when each tool appeared, and replaying it yields exactly the executable set. Adding a tool mid-session appends to the prompt instead of rewriting it, which preserves the provider's prompt cache (Chapter 4).

6.7 The Agent class: state, queues and a run lifecycle#

Agent owns a mutable state (messages, tools, model, thinkingLevel, plus read-only isStreaming, streamingMessage, pendingToolCalls, errorMessage) and exposes prompt(), continue(), steer(), followUp(), abort(), waitForIdle() and subscribe().

State changes happen in one place, processEvents(), a reducer: for each event it updates state (message_end appends the message, tool_execution_start adds the id to pendingToolCalls, …) and then awaits each listener in subscription order. Note that pendingToolCalls is replaced with a new Set rather than mutated, so a UI holding the old reference can compare identities to detect change.

Runs are mutually exclusive. runWithLifecycle() refuses to start if activeRun exists ("Use steer() or followUp() to queue messages, or wait for completion"), creates a fresh AbortController, executes the loop, converts any unexpected exception into a synthetic failed assistant message with a full event sequence (message_start, message_end, turn_end, agent_end), and in finally clears runtime state and resolves the promise that waitForIdle() returns. The run is idle only after the agent_end listeners finish, so code like "save the session, then exit" can await it.

State arrays are protected by defensive copying: assigning state.messages or state.tools copies the array, and createContextSnapshot() hands the loop copies, so the loop's working context and the agent's committed transcript cannot alias each other.

6.8 Cancellation#

JavaScript has no way to kill a running function. Cancellation is cooperative: an AbortController produces an AbortSignal, and every long operation checks signal.aborted or listens for its abort event. Agent.abort() aborts the run's controller. The signal flows into the stream function (adapters pass it to fetch, which tears down the HTTP connection), into beforeToolCall/afterToolCall ("The hook receives the agent abort signal and is responsible for honoring it"), and into each tool.execute (the bash tool kills its process tree, Chapter 8). The loop checks it between steps: sequential execution stops after the current tool; parallel thunks that have not started return "Operation aborted" results. The adapter then ends the response with stopReason: "aborted", a hard exit, and the agent becomes idle.

6.9 Running the loop through a server#

proxy.ts provides a StreamFn for apps whose server holds the API keys. A subtle optimization: every pi-ai event carries the full partial message, so forwarding events verbatim would send the whole response-so-far with every token — O(n²) bytes for an n-token response. The proxy protocol strips partial ("server sends these with partial field stripped to reduce bandwidth") and the client rebuilds the message from deltas, using parseStreamingJson for tool arguments. This is delta encoding: send changes, reconstruct state at the receiver.

6.10 Concepts in this chapter#

• Event-driven architecture and the observer pattern. Producers emit events; any number of subscribers react, decoupled from the producer.
• Reducers. State changes expressed as a function of (state, event); one place mutates state.
• Producer/consumer queues and drain points. Work arriving asynchronously is queued and consumed at well-defined points in a control loop.
• Cooperative cancellation. Cancellation tokens (AbortSignal) are passed down and checked; nothing is forcibly killed mid-instruction.
• Fork-join concurrency. Start independent tasks, wait for all (Promise.all), combine results in a fixed order.
• Thunks / deferred computation. Wrap work in a function to separate planning from execution.
• Inversion of control and dependency injection. The loop calls hooks and an injected stream function; callers customize behaviour without subclassing.
• Defensive copying and immutability at boundaries. Copy on assignment and on hand-off to avoid aliasing bugs.
• Mutual exclusion. One active run at a time, enforced by a guard.
• Delta encoding. Transmit differences, not full state.
• Declaration merging / open unions. Extend a library's types from outside it.

Exercises#

1. Write the full event sequence for a run where the first response contains two parallel tool calls, the second finishes first, and a steering message arrives during tool execution.
2. Why are tool preparations sequential even in parallel mode? Give a concrete user-visible problem the alternative would cause.
3. Implement a beforeToolCall hook that blocks bash commands containing rm -rf and sets terminate: true. When exactly does the run stop?
4. In executePreparedToolCall, what bug would appear if acceptingUpdates were not set to false before awaiting updateEvents?
5. Explain the difference between prompt() and continue(), and why continue() refuses when the last message is an assistant message with no queued messages.
6. A listener on message_end writes to disk and takes 50 ms. How does that affect the loop, and why is that the desired behaviour?
7. Estimate the bytes a non-optimized proxy would send for a 2,000-token reply if each token is 4 bytes and each event carries the partial message.