285 KiB
285 KiB
Agent Architecture: Pi Agent Core - A Comprehensive Guide for Julia Implementation
Table of Contents
- Overview
- Architecture Layers
- Core Components
- Message System
- Agent Loop
- Tool Execution
- Session Management
- Memory & Context Management
- Event System
- Hook System
- Implementation Guide for Julia
- Data Flow Diagrams
- Key Algorithms
Complete System Architecture
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ COMPLETE SYSTEM ARCHITECTURE DIAGRAM │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ USER INPUT │
│ "what product do you have in stock?" │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ AGENT CLASS (agent.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ State (_state: MutableAgentState) │ │
│ │ • systemPrompt │ │
│ │ • model │ │
│ │ • thinkingLevel │ │
│ │ • tools (accessor) │ │
│ │ • messages (accessor) │ │
│ │ • isStreaming │ │
│ │ • streamingMessage │ │
│ │ • pendingToolCalls │ │
│ │ • errorMessage │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Queues │ │
│ │ • steeringQueue (PendingMessageQueue) │ │
│ │ • followUpQueue (PendingMessageQueue) │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Hooks │ │
│ │ • convertToLlm │ │
│ │ • transformContext │ │
│ │ • beforeToolCall │ │
│ │ • afterToolCall │ │
│ │ • prepareNextTurn │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Methods │ │
│ │ • prompt(input) → normalizePromptInput() → runPromptMessages() │ │
│ │ • continue() → runContinuation() │ │
│ │ • reset() → clear state │ │
│ │ • subscribe(listener) → unsubscribe() │ │
│ │ • abort() → signal.abort() │ │
│ │ • waitForIdle() → activeRun.promise │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ AGENT LOOP (agent_loop.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ runLoop() │ │
│ │ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ Outer Loop: │ │ │
│ │ │ while (true): │ │ │
│ │ │ • Process pending messages (steering/follow-up) │ │ │
│ │ │ • streamAssistantResponse() │ │ │
│ │ │ • executeToolCalls() │ │ │
│ │ │ • emit(turn_end) │ │ │
│ │ │ • prepareNextTurn hook │ │ │
│ │ │ • shouldStopAfterTurn hook │ │ │
│ │ │ • Check steering/follow-up queues │ │ │
│ │ │ → Continue or exit │ │ │
│ │ └──────────────────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ streamAssistantResponse() │ │
│ │ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ 1. transformContext() (optional) │ │ │
│ │ │ 2. convertToLlm() │ │ │
│ │ │ 3. Build LLM context │ │ │
│ │ │ 4. Resolve API key │ │ │
│ │ │ 5. streamFunction() → LLM API │ │ │
│ │ │ 6. Stream events (start, delta*, done) │ │ │
│ │ │ 7. Commit message to context │ │ │
│ │ └──────────────────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ executeToolCalls() │ │
│ │ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ if parallel: │ │ │
│ │ │ • Preflight sequentially │ │ │
│ │ │ • Execute concurrently (Promise.all) │ │ │
│ │ │ │ │ │
│ │ │ if sequential: │ │ │
│ │ │ • Execute one-by-one │ │ │
│ │ │ │ │ │
│ │ │ For each tool: │ │ │
│ │ │ • prepareToolCall() │ │ │
│ │ │ • executePreparedToolCall() │ │ │
│ │ │ • finalizeExecutedToolCall() │ │ │
│ │ │ • createToolResultMessage() │ │ │
│ │ │ • emit(tool_execution_start/end) │ │ │
│ │ │ • emit(message_start/end) for toolResult │ │ │
│ │ └──────────────────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ PROVIDER ABSTRACTION │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ streamFunction(model, llmContext, options) │ │
│ │ • Models.streamSimple() from @earendil-works/pi-ai │ │
│ │ • Makes actual LLM API call │ │
│ │ • Returns AssistantMessageEventStream │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ SESSION PERSISTENCE (session.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Session │ │
│ │ • storage: SessionStorage │ │
│ │ • metadata: SessionMetadata │ │
│ │ • entries: Dict{String, SessionEntry} │ │
│ │ • leaf_id: Union{String, Nothing} │ │
│ │ │ │
│ │ Methods: │ │
│ │ • get_branch() → SessionTreeEntry[] │ │
│ │ • build_context() → SessionContext │ │
│ │ • append_message() │ │
│ │ • append_compaction() │ │
│ │ • append_branch_summary() │ │
│ │ • set_leaf_id() │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ SessionStorage (JSONL) │ │
│ │ • file_path: String │ │
│ │ • header: SessionMetadata │ │
│ │ • entries: Vector{SessionEntry} │ │
│ │ │ │
│ │ Entry Types: │ │
│ │ • message (user/assistant/toolResult) │ │
│ │ • compaction (summary) │ │
│ │ • branch_summary (divergence) │ │
│ │ • leaf (current pointer) │ │
│ │ • thinking_level_change │ │
│ │ • model_change │ │
│ │ • active_tools_change │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ MEMORY MANAGEMENT (compaction.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Token Estimation │ │
│ │ • estimateContextTokens(messages) │ │
│ │ • estimateTokens(message) (chars / 4 heuristic) │ │
│ │ │ │
│ │ Compaction Decision │ │
│ │ • shouldCompact(contextTokens, contextWindow, settings) │ │
│ │ • Trigger: contextTokens > contextWindow - reserveTokens │ │
│ │ │ │
│ │ Cut Point Finding │ │
│ │ • findCutPoint(entries, keepRecentTokens) │ │
│ │ • Walk backward, skip toolResults │ │
│ │ │ │
│ │ Summary Generation │ │
│ │ • prepareCompaction(branchEntries, settings) │ │
│ │ • generateSummary(messages, previousSummary?) │ │
│ │ • compact(preparation, model, models) │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ EVENT SYSTEM (events.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Event Types │ │
│ │ • agent_start / agent_end │ │
│ │ • turn_start / turn_end │ │
│ │ • message_start / message_update / message_end │ │
│ │ • tool_execution_start / update / end │ │
│ │ │ │
│ │ Agent.subscribe(listener) → unsubscribe() │ │
│ │ processEvents(event) → await listeners │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ HOOK SYSTEM (hooks.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Hook Points │ │
│ │ • convertToLlm (AgentMessage[] → Message[]) │ │
│ │ • transformContext (optional pruning/injection) │ │
│ │ • beforeToolCall (can block: { block: true }) │ │
│ │ • afterToolCall (can override: { content, details, isError, ... }) │ │
│ │ • prepareNextTurn (context/model/thinkingLevel update) │ │
│ │ • shouldStopAfterTurn (graceful termination) │ │
│ │ • getSteeringMessages (mid-turn injection) │ │
│ │ • getFollowUpMessages (post-agent execution) │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
Overview
The Pi Agent Core is a sophisticated stateful agent system with the following characteristics:
- Stateful execution: Maintains conversation history and context across multiple turns
- Tool execution: Supports LLM tool calling with parallel/sequential execution modes
- Event streaming: Real-time event system for UI updates
- Session persistence: JSONL-based persistent storage with tree-structured branching
- Memory compaction: Automatic context window management through LLM summarization
- Flexible extension: Hook-based customization at every system boundary
Key Design Principles
- Separation of concerns: Core agent logic is separated from storage and provider implementations
- Streaming first: All operations are designed around async streams for responsiveness
- Type safety: Strong TypeScript types for compile-time guarantees
- Extensibility: Hooks at every major boundary allow customization
- Persistence: Session history survives restarts through JSONL files
High-Level Architecture Overview
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ USER INPUT │
│ "what product do you have in stock?" │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 1. AGENT.PROMPT() ENTRY │
│ File: packages/agent/src/agent.ts:339 │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ • Validate no active run (throws if busy) │ │
│ │ • normalizePromptInput() converts string to AgentMessage[] │ │
│ │ • runPromptMessages() launches execution with lifecycle events │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 2. AGENT LIFECYCLE INITIALIZATION │
│ File: packages/agent/src/agent.ts:398-412 │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Events Emitted: │ │
│ │ • agent_start │ │
│ │ • turn_start │ │
│ │ • message_start / message_end (for each prompt message) │ │
│ │ │ │
│ │ Context Snapshot Created: │ │
│ │ • systemPrompt │ │
│ │ • messages (copy) │ │
│ │ • tools (copy) │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 3. AGENT LOOP STARTS │
│ File: packages/agent/src/agent-loop.ts:95 │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ runPromptMessages() → runAgentLoop() │ │
│ │ • Prompts added to context.messages │ │
│ │ • Lifecycle events emitted for prompts │ │
│ │ • Calls runLoop() (main processing loop) │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 4. LLM CALL BOUNDARY - MESSAGE TRANSFORMATION │
│ File: packages/agent/src/agent-loop.ts:281-372 (streamAssistantResponse) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Step 4.1: Context Transform (optional) │ │
│ │ transformContext(messages) → transformed messages │ │
│ │ (Used for context pruning/injection) │ │
│ │ │ │
│ │ Step 4.2: LLM Conversion │ │
│ │ convertToLlm(messages) → Message[] │ │
│ │ - Filters non-LLM messages (bashExecution, branchSummary, etc.) │ │
│ │ - Converts: user → user, assistant → assistant, toolResult → toolResult │ │
│ │ │ │
│ │ Step 4.3: Build LLM Context │ │
│ │ { │ │
│ │ systemPrompt: context.systemPrompt, │ │
│ │ messages: llmMessages, │ │
│ │ tools: context.tools │ │
│ │ } │ │
│ │ │ │
│ │ Step 4.4: Resolve API Key │ │
│ │ getApiKey(model.provider) → apiKey │ │
│ │ │ │
│ │ Step 4.5: Stream Function Call │ │
│ │ streamFunction(model, llmContext, options) │ │
│ │ - Default: Models.streamSimple() from @earendil-works/pi-ai │ │
│ │ - Makes actual LLM API call │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 5. LLM RESPONSE STREAMING │
│ File: packages/agent/src/agent-loop.ts:317-371 │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Stream yields events: │ │
│ │ • start → creates partial AssistantMessage │ │
│ │ • text_start → streaming text begins │ │
│ │ • text_delta → text chunks arrive │ │
│ │ • toolcall_start → tool call block begins │ │
│ │ • toolcall_delta → tool call arguments arrive │ │
│ │ • toolcall_end → tool call block complete │ │
│ │ • text_end → text block complete │ │
│ │ • done → final message complete │ │
│ │ │ │
│ │ State Updates: │ │
│ │ • Partial message pushed to context.messages │ │
│ │ • message_start event emitted │ │
│ │ • message_update events emitted as text/tools stream in │ │
│ │ • Final message committed to context.messages │ │
│ │ • message_end event emitted │ │
│ │ │ │
│ │ Stop Reasons: │ │
│ │ • stop - normal completion │ │
│ │ • toolUse - model requested tool calls │ │
│ │ • length - token limit reached │ │
│ │ • error - failure │ │
│ │ • aborted - operation aborted │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 6. TOOL CALL PARSING AND EXECUTION │
│ File: packages/agent/src/agent-loop.ts:408-554 │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Step 6.1: Extract Tool Calls │ │
│ │ toolCalls = message.content.filter(c => c.type === "toolCall") │ │
│ │ │ │
│ │ Step 6.2: Determine Execution Mode │ │
│ │ - Check config.toolExecution: "parallel" or "sequential" │ │
│ │ - Check individual tool executionMode setting │ │
│ │ - Decides how to execute tool batch │ │
│ │ │ │
│ │ Step 6.3: For Each Tool Call │ │
│ │ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ 1. Tool Lookup │ │ │
│ │ │ tool = context.tools.find(t => t.name === toolCall.name) │ │ │
│ │ │ │ │ │
│ │ │ 2. Argument Preparation │ │ │
│ │ │ prepared = tool.prepareArguments?(toolCall.arguments) │ │ │
│ │ │ │ │ │
│ │ │ 3. Argument Validation │ │ │
│ │ │ validateToolArguments(tool, preparedToolCall) │ │ │
│ │ │ │ │ │
│ │ │ 4. Before Tool Hook │ │ │
│ │ │ beforeToolCall({ assistantMessage, toolCall, args, context }) │ │ │
│ │ │ - Can block execution by returning { block: true, reason } │ │ │
│ │ │ │ │ │
│ │ │ 5. Execution │ │ │
│ │ │ execute(toolCallId, params, signal, onUpdate, context) │ │ │
│ │ │ │ │ │
│ │ │ - Parallel Mode: Tools execute concurrently │ │ │
│ │ │ - Sequential Mode: Tools execute one-by-one │ │ │
│ │ └──────────────────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 7. TOOL EXECUTION EXAMPLE - READ TOOL │
│ File: packages/agent/src/harness/tools/read.ts │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ User asks: "what product do you have in stock?" │ │
│ │ │ │
│ │ Agent decides to read catalog file "products.json" │ │
│ │ │ │
│ │ Input Arguments: │ │
│ │ { │ │
│ │ "path": "products.json", │ │
│ │ "offset": 1, │ │
│ │ "limit": 100 │ │
│ │ } │ │
│ │ │ │
│ │ Execution Steps: │ │
│ │ 1. resolveReadToolPath(env, path, signal) → absolutePath │ │
│ │ 2. env.readBinaryFile(absolutePath, signal) → bytes │ │
│ │ 3. Detect mimeType (check if image) │ │
│ │ 4. For text files: │ │
│ │ - Decode UTF-8 → textContent │ │
│ │ - Split by lines → allLines │ │
│ │ - Apply offset/limit slicing │ │
│ │ - Truncate if exceeds DEFAULT_MAX_BYTES or DEFAULT_MAX_LINES │ │
│ │ - Add truncation notice to output │ │
│ │ 5. Return result: │ │
│ │ { │ │
│ │ content: [{ type: "text", text: output }], │ │
│ │ details: { truncation: ... } │ │
│ │ } │ │
│ │ │ │
│ │ Output Example: │ │
│ │ "Showing lines 1-50 of 150. [Showing 50 lines of 150. Use offset=51 to │ │
│ │ continue.]" │ │
│ │ │ │
│ │ Tool Result: │ │
│ │ { │ │
│ │ "role": "toolResult", │ │
│ │ "toolCallId": "tool_abc123", │ │
│ │ "toolName": "read", │ │
│ │ "content": [{ "type": "text", "text": "..." }], │ │
│ │ "isError": false, │ │
│ │ "timestamp": 1721721600000 │ │
│ │ } │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 8. TOOL RESULT HANDLING │
│ File: packages/agent/src/agent-loop.ts:556-792 │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Step 8.1: After Tool Hook │ │
│ │ afterToolCall({ assistantMessage, toolCall, args, result, isError, ctx }) │ │
│ │ - Can override result content, details, usage, terminate hint │ │
│ │ │ │
│ │ Step 8.2: Create Tool Result Message │ │
│ │ { │ │
│ │ role: "toolResult", │ │
│ │ toolCallId: toolCall.id, │ │
│ │ toolName: toolCall.name, │ │
│ │ content: result.content ?? [], │ │
│ │ details: result.details, │ │
│ │ usage: result.usage, │ │
│ │ isError: false, │ │
│ │ timestamp: Date.now() │ │
│ │ } │ │
│ │ │ │
│ │ Step 8.3: Emit Events │ │
│ │ • tool_execution_start │ │
│ │ • tool_execution_end │ │
│ │ • message_start (toolResult message) │ │
│ │ • message_end (toolResult message) │ │
│ │ │ │
│ │ Step 8.4: Update Context │ │
│ │ • Push tool result message to currentContext.messages │ │
│ │ • Push to newMessages array │ │
│ │ │ │
│ │ Step 8.5: Batch Termination Check │ │
│ │ shouldTerminateToolBatch(finalizedCalls) │ │
│ │ - Returns true if ALL tools set terminate: true │ │
│ │ - If true, agent may stop after this batch │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 9. NEXT TURN PREPARATION │
│ File: packages/agent/src/agent-loop.ts:224-257 │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Step 9.1: Turn End Event │ │
│ │ turn_end emitted with message and toolResults │ │
│ │ │ │
│ │ Step 9.2: prepareNextTurn Hook │ │
│ │ prepareNextTurn({ message, toolResults, context, newMessages }) │ │
│ │ - Can return updated context, model, or thinking level │ │
│ │ - Used for dynamic context management │ │
│ │ │ │
│ │ Step 9.3: Queue Polling │ │
│ │ getSteeringMessages() → inject messages for immediate processing │ │
│ │ getFollowUpMessages() → check for queued follow-up messages │ │
│ │ │ │
│ │ Step 9.4: Loop Decision │ │
│ │ • If steering messages exist → process them, continue loop │ │
│ │ • If follow-up messages exist → process them, continue loop │ │
│ │ • If tool calls remain in message → continue inner loop │ │
│ │ • If no messages → emit agent_end, exit loop │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 10. AGENT RESPONSE GENERATION │
│ File: packages/agent/src/agent-loop.ts:58-163 │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ The agent loop continues until: │ │
│ │ • No tool calls remain in assistant messages │ │
│ │ • No steering/follow-up messages queued │ │
│ │ • shouldStopAfterTurn() returns true (if configured) │ │
│ │ │ │
│ │ Final Response Generation: │ │
│ │ 1. LLM streams text content blocks │ │
│ │ 2. Message committed to context │ │
│ │ 3. turn_end emitted │ │
│ │ 4. agent_end emitted with all new messages │ │
│ │ 5. Agent returns to idle state │ │
│ │ │ │
│ │ Final Agent Response: │ │
│ │ "We have 15 products in stock:" │ │
│ │ • Product A - $29.99 │ │
│ │ • Product B - $49.99 │ │
│ │ • Product C - $19.99 │ │
│ │ (and 2 more products) │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
---
---
## Architecture Layers
┌─────────────────────────────────────────────────────────────────────┐ │ User Interface │ │ (VS Code Extension, CLI, etc.) │ └─────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────┐ │ Agent Harness (agent-harness.ts) │ │ - Session integration │ │ - Hook system │ │ - Skill/prompt template management │ │ - System prompt building │ └─────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────┐ │ Agent Class (agent.ts) │ │ - State management (messages, tools, model) │ │ - Event emission │ │ - Steering/follow-up queues │ │ - Active run management │ └─────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────┐ │ Agent Loop (agent-loop.ts) │ │ - Main execution loop │ │ - LLM call orchestration │ │ - Tool execution │ │ - Context transformation │ └─────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────┐ │ Provider Abstraction │ │ (@earendil-works/pi-ai - external) │ │ - LLM API calls │ │ - Streaming interface │ │ - Retry policies │ └─────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────┐ │ Session Storage (session/) │ │ - JSONL file persistence │ │ - Tree-structured history │ │ - Compaction and summarization │ └─────────────────────────────────────────────────────────────────────┘
---
## Core Components
### 1. Agent Class (`agent.ts`)
**Purpose**: Stateful wrapper around the low-level agent loop
**Architecture Flow**:
┌─────────────────────────────────────────────────────────────────────────────────────┐ │ AGENT CLASS ARCHITECTURE │ ├─────────────────────────────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────────────────────────────────────────────────────────────────────────┐ │ │ │ State (_state: MutableAgentState) │ │ │ │ • systemPrompt │ │ │ │ • model │ │ │ │ • thinkingLevel │ │ │ │ • tools (accessor) │ │ │ │ • messages (accessor) │ │ │ │ • isStreaming │ │ │ │ • streamingMessage │ │ │ │ • pendingToolCalls │ │ │ │ • errorMessage │ │ │ └──────────────────────────────────────────────────────────────────────────────┘ │ │ │ │ ┌──────────────────────────────────────────────────────────────────────────────┐ │ │ │ Queues (PendingMessageQueue) │ │ │ │ • steeringQueue │ │ │ │ └─ steer() → queue message for immediate injection │ │ │ │ • followUpQueue │ │ │ │ └─ followUp() → queue message for post-agent execution │ │ │ └──────────────────────────────────────────────────────────────────────────────┘ │ │ │ │ ┌──────────────────────────────────────────────────────────────────────────────┐ │ │ │ Hooks (optional customization points) │ │ │ │ • convertToLlm │ │ │ │ • transformContext │ │ │ │ • beforeToolCall │ │ │ │ • afterToolCall │ │ │ │ • prepareNextTurn │ │ │ │ • prepareNextTurnWithContext │ │ │ └──────────────────────────────────────────────────────────────────────────────┘ │ │ │ │ ┌──────────────────────────────────────────────────────────────────────────────┐ │ │ │ Active Run Management │ │ │ │ • activeRun: { promise, resolve, abortController } │ │ │ │ • abort() → sets signal.aborted │ │ │ │ • waitForIdle() → returns promise that settles after all listeners │ │ │ └──────────────────────────────────────────────────────────────────────────────┘ │ │ │ │ ┌──────────────────────────────────────────────────────────────────────────────┐ │ │ │ Event Subscription │ │ │ │ • listeners: Set │ │ │ │ • subscribe(listener) → unsubscribe() │ │ │ │ • processEvents(event) → await listeners for event │ │ │ └──────────────────────────────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────────────────────┘
**Key responsibilities**:
- Maintain in-memory transcript (`_state.messages`)
- Manage tools, system prompt, model configuration
- Emit lifecycle events to subscribers
- Handle steering/follow-up message queues
- Prevent concurrent runs (single active run at a time)
**State structure**:
```typescript
interface AgentState {
systemPrompt: string
model: Model<any>
thinkingLevel: ThinkingLevel
tools: AgentTool<any>[]
messages: AgentMessage[]
isStreaming: boolean
streamingMessage?: AgentMessage
pendingToolCalls: Set<string>
errorMessage?: string
}
Key methods:
prompt(message): Start a new prompt from text, message, or arraycontinue(): Continue from current contextsteer(message): Queue steering message (interruption during tool execution)followUp(message): Queue follow-up message (executes after agent stops)reset(): Clear all statesubscribe(listener): Register event listenerabort(): Cancel current operationwaitForIdle(): Wait for completion and event listeners
2. Agent Loop (agent-loop.ts)
Purpose: Core execution logic without state management
Outer Loop Flow:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ OUTER LOOP: TURN MANAGEMENT │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ while (true): │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Inner Loop (tool calls + steering messages) │ │
│ │ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ while (hasMoreToolCalls || pendingMessages): │ │ │
│ │ │ │ │ │
│ │ │ 1. Process pending messages │ │ │
│ │ │ (steering/follow-up injection) │ │ │
│ │ │ │ │ │
│ │ │ 2. Stream assistant response from LLM │ │ │
│ │ │ (streamAssistantResponse) │ │ │
│ │ │ │ │ │
│ │ │ 3. Extract tool calls │ │ │
│ │ │ (filter toolCall blocks from content) │ │ │
│ │ │ │ │ │
│ │ │ 4. Execute tool batch │ │ │
│ │ │ (executeToolCalls: parallel or sequential) │ │ │
│ │ │ │ │ │
│ │ │ 5. Update context │ │ │
│ │ │ (add tool results to messages) │ │ │
│ │ │ │ │ │
│ │ │ 6. Emit turn_end │ │ │
│ │ │ │ │ │
│ │ │ 7. Check prepareNextTurn hook │ │ │
│ │ │ │ │ │
│ │ │ 8. Check shouldStopAfterTurn hook │ │ │
│ │ │ │ │ │
│ │ │ 9. Get steering messages │ │ │
│ │ └──────────────────────────────────────────────────────────────────────┘ │ │
│ │ │ │
│ │ Check follow-up messages │ │
│ │ └─ if follow-ups exist → continue outer loop │ │
│ │ else → emit agent_end, exit │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
Inner Loop: LLM Call Boundary:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ LLM CALL BOUNDARY: streamAssistantResponse() │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 1. Context Transform (optional) │
│ transformContext(messages) → transformed messages │
│ (pruning, context injection) │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 2. Convert to LLM format │
│ convertToLlm(messages) → Message[] │
│ (filter non-LLM messages, convert custom types) │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 3. Build LLM Context │
│ { │
│ systemPrompt: context.systemPrompt, │
│ messages: llmMessages, │
│ tools: context.tools │
│ } │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 4. Resolve API Key │
│ getApiKey(model.provider) → apiKey │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 5. Stream Function Call │
│ streamFunction(model, llmContext, options) │
│ (calls LLM provider API, returns AssistantMessageEventStream) │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 6. Stream Events │
│ for await (event of response) { │
│ case "start": → partialMessage = event.partial │
│ case "text_delta": → update partialMessage │
│ case "toolcall_delta": → update partialMessage │
│ case "done": → commit final message │
│ case "error": → commit error message │
│ } │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 7. Commit Message │
│ If partial existed: context.messages[.end] = final │
│ Else: context.messages.push(final) │
└─────────────────────────────────────────────────────────────────────────────────────┘
Tool Execution Flow:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ TOOL EXECUTION FLOW │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Assistant Message (with toolCall content blocks) │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ extract tool calls from message.content │
│ filter(c => c.type === "toolCall") │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ determine execution mode │
│ • config.toolExecution: "parallel" or "sequential" │
│ • per-tool executionMode override │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ PARALLEL MODE │ │
│ │ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ 1. Preflight all tools sequentially │ │ │
│ │ │ (prepareToolCall for each) │ │ │
│ │ │ │ │ │
│ │ │ 2. Queue async executions │ │ │
│ │ │ (async () => execute + finalize) │ │ │
│ │ │ │ │ │
│ │ │ 3. Execute allowed tools concurrently │ │ │
│ │ │ (Promise.all for queued functions) │ │ │
│ │ │ │ │ │
│ │ │ 4. Emit toolResult messages │ │ │
│ │ │ (in assistant source order) │ │ │
│ │ └──────────────────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ SEQUENTIAL MODE │ │
│ │ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ For each tool call in order: │ │ │
│ │ │ 1. prepareToolCall │ │ │
│ │ │ 2. executePreparedToolCall │ │ │
│ │ │ 3. finalizeExecutedToolCall │ │ │
│ │ │ 4. Emit tool_result messages │ │ │
│ │ └──────────────────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ check shouldTerminateToolBatch() │
│ (returns true if ALL tools set terminate: true) │
└─────────────────────────────────────────────────────────────────────────────────────┘
Main entry points:
runAgentLoop(prompts, context, config, emit, signal, streamFn): Start new loop with promptsrunAgentLoopContinue(context, config, emit, signal, streamFn): Continue from existing context
Execution phases:
Phase 1: Outer Loop (Turn Management)
while (true):
1. Check steering messages (inject if any)
2. Stream assistant response from LLM
3. Extract tool calls from response
4. Execute tool batch
5. Emit turn_end event
6. Check prepareNextTurn hook
7. Check shouldStopAfterTurn hook
8. Check follow-up messages
9. If follow-up exists, continue outer loop
10. If no follow-up, exit
Phase 2: Inner Loop (Tool Call Processing)
while (hasMoreToolCalls || pendingMessages):
1. Process pending messages
2. Stream assistant response
3. Extract tool calls
4. Execute tool batch (parallel or sequential)
5. Emit turn_end
Phase 3: LLM Call Boundary
1. transformContext(messages) // Optional pruning/injection
2. convertToLlm(messages) // Filter/custom message conversion
3. Build LLM context
4. Resolve API key
5. Call streamFunction(model, context, options)
6. Stream events from LLM
7. Commit final message to context
Tool execution modes:
- Parallel (default): Preflight sequentially, execute allowed tools concurrently
- Sequential: Execute tools one-by-one
3. Types System (types.ts)
Key types:
AgentMessage
type AgentMessage = Message | CustomAgentMessages[keyof CustomAgentMessages]
Extensible union of LLM messages and custom app-specific messages.
AgentTool
interface AgentTool<TParameters, TDetails> extends Tool<TParameters> {
label: string // UI display name
prepareArguments?: (args) => Static<T> // Argument transformation
execute: (toolCallId, params, signal, onUpdate) => Promise<AgentToolResult>
executionMode?: "parallel" | "sequential" // Per-tool override
}
AgentContext
interface AgentContext {
systemPrompt: string
messages: AgentMessage[]
tools?: AgentTool<any>[]
}
AgentLoopConfig
Configuration object passed to low-level loop functions, including:
- Model specification
- convertToLlm transformation
- transformContext (optional)
- beforeToolCall hook
- afterToolCall hook
- prepareNextTurn hook
- shouldStopAfterTurn hook
- getSteeringMessages hook
- getFollowUpMessages hook
Message System
Message Types
LLM Messages (standard)
user: User inputassistant: LLM response (streaming, contains tool calls)toolResult: Tool execution result
Custom Messages (app-specific)
bashExecution: Shell command execution result (hidden from LLM by default)custom: Custom app messages (visible to LLM if projected)branchSummary: Branch divergence summarycompactionSummary: History compaction result
Message Type Hierarchy:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ MESSAGE TYPE HIERARCHY │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ AgentMessage (Union) │
├─────────────────────────────────────────────────────────────────────────────────────┤
│ │ │
│ ├─ Message (LLM messages) │
│ │ │ │
│ │ ├─ UserMessage │
│ │ │ role: "user" │
│ │ │ content: TextContent[] │
│ │ │ timestamp: number │
│ │ │ │
│ │ ├─ AssistantMessage │
│ │ │ role: "assistant" │
│ │ │ content: (TextContent | ToolCall | Thinking)[] │
│ │ │ api, provider, model │
│ │ │ usage: Usage │
│ │ │ stopReason: string │
│ │ │ errorMessage?: string │
│ │ │ timestamp: number │
│ │ │ │
│ │ └─ ToolResultMessage │
│ │ role: "toolResult" │
│ │ toolCallId, toolName │
│ │ content: TextContent[] │
│ │ details, usage │
│ │ isError: boolean │
│ │ timestamp: number │
│ │ │
│ └─ CustomAgentMessages (app extensions) │
│ │ │
│ ├─ BashExecutionMessage │
│ │ role: "bashExecution" │
│ │ command, output, exitCode │
│ │ excludeFromContext?: boolean │
│ │ │
│ ├─ CustomMessage │
│ │ role: "custom" │
│ │ customType, content, display, details │
│ │ │
│ ├─ BranchSummaryMessage │
│ │ role: "branchSummary" │
│ │ summary, fromId │
│ │ │
│ └─ CompactionSummaryMessage │
│ role: "compactionSummary" │
│ summary, tokensBefore │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ MESSAGE FLOW: AGENT → LLM │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ AgentMessage[] (in-memory transcript) │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ transformContext() (optional) │
│ (pruning, context injection) │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ convertToLlm() │
│ ┌────────────────────────────────────────────────────────────────────────────┐ │
│ │ for each message: │ │
│ │ • user/assistant/toolResult → pass through │ │
│ │ • bashExecution → convert to user message (with command/output) │ │
│ │ • custom → convert to user message (text content) │ │
│ │ • branchSummary → convert to user message │ │
│ │ • compactionSummary → convert to user message │ │
│ └────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Message[] (LLM-compatible) │
│ (only user, assistant, toolResult) │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ LLM provider API call │
└─────────────────────────────────────────────────────────────────────────────────────┘
Message Flow
User Input
↓
normalizePromptInput() → AgentMessage[]
↓
AgentContext.messages (in-memory)
↓
transformContext() (optional, for pruning/injection)
↓
convertToLlm() (filters/custom → LLM format)
↓
LLM provider (Message[])
↓
AssistantMessage (streamed)
↓
AgentContext.messages (committed)
Message Commitment
Messages are committed to context.messages at specific points:
- Assistant partial message: Added when
startevent arrives - Assistant final message: Replaces partial on
done/error - Tool result message: Added after
tool_execution_end
Agent Loop
Main Algorithm
async function runLoop(initialContext, newMessages, config, signal, emit, streamFn) {
let currentContext = initialContext
let pendingMessages = (await config.getSteeringMessages?.()) || []
while (true) {
let hasMoreToolCalls = true
while (hasMoreToolCalls || pendingMessages.length > 0) {
// Process pending messages (steering/follow-up)
if (pendingMessages.length > 0) {
for (const msg of pendingMessages) {
await emit({ type: "message_start", message: msg })
await emit({ type: "message_end", message: msg })
currentContext.messages.push(msg)
newMessages.push(msg)
}
pendingMessages = []
}
// Stream assistant response
const message = await streamAssistantResponse(
currentContext, config, signal, emit, streamFn
)
newMessages.push(message)
// Check for errors
if (message.stopReason === "error" || message.stopReason === "aborted") {
await emit({ type: "turn_end", message, toolResults: [] })
await emit({ type: "agent_end", messages: newMessages })
return
}
// Extract and execute tool calls
const toolCalls = message.content.filter(c => c.type === "toolCall")
const toolResults = []
hasMoreToolCalls = false
if (toolCalls.length > 0) {
const executedBatch = await executeToolCalls(
currentContext, message, config, signal, emit
)
toolResults.push(...executedBatch.messages)
hasMoreToolCalls = !executedBatch.terminate
for (const result of toolResults) {
currentContext.messages.push(result)
newMessages.push(result)
}
}
await emit({ type: "turn_end", message, toolResults })
// Check prepareNextTurn hook
const nextTurnSnapshot = await config.prepareNextTurn?.({
message, toolResults, context: currentContext, newMessages
})
if (nextTurnSnapshot) {
currentContext = nextTurnSnapshot.context ?? currentContext
config = { ...config, model: nextTurnSnapshot.model ?? config.model }
}
// Check shouldStopAfterTurn hook
if (await config.shouldStopAfterTurn?.({ ... })) {
await emit({ type: "agent_end", messages: newMessages })
return
}
// Get steering messages for next iteration
pendingMessages = (await config.getSteeringMessages?.()) || []
}
// Check follow-up messages
const followUpMessages = (await config.getFollowUpMessages?.()) || []
if (followUpMessages.length > 0) {
pendingMessages = followUpMessages
continue
}
// No more messages, exit
break
}
await emit({ type: "agent_end", messages: newMessages })
}
Assistant Response Streaming
async function streamAssistantResponse(context, config, signal, emit, streamFn) {
// 1. Transform context (optional)
let messages = context.messages
if (config.transformContext) {
messages = await config.transformContext(messages, signal)
}
// 2. Convert to LLM format
const llmMessages = await config.convertToLlm(messages)
// 3. Build LLM context
const llmContext = {
systemPrompt: context.systemPrompt,
messages: llmMessages,
tools: context.tools,
}
// 4. Resolve API key
const resolvedApiKey = await config.getApiKey?.(config.model.provider) || config.apiKey
// 5. Call stream function
const response = await streamFn(config.model, llmContext, {
...config, apiKey: resolvedApiKey, signal
})
// 6. Stream events
let partialMessage: AssistantMessage | null = null
let addedPartial = false
for await (const event of response) {
switch (event.type) {
case "start":
partialMessage = event.partial
context.messages.push(partialMessage)
addedPartial = true
await emit({ type: "message_start", message: { ...partialMessage } })
break
case "text_start" | "text_delta" | "text_end" |
"thinking_start" | "thinking_delta" | "thinking_end" |
"toolcall_start" | "toolcall_delta" | "toolcall_end":
if (partialMessage) {
partialMessage = event.partial
context.messages[context.messages.length - 1] = partialMessage
await emit({
type: "message_update",
assistantMessageEvent: event,
message: { ...partialMessage }
})
}
break
case "done" | "error": {
const finalMessage = await response.result()
if (addedPartial) {
context.messages[context.messages.length - 1] = finalMessage
} else {
context.messages.push(finalMessage)
await emit({ type: "message_start", message: { ...finalMessage } })
}
await emit({ type: "message_end", message: finalMessage })
return finalMessage
}
}
}
}
Tool Execution
Tool Call Flow
Assistant Message (with toolCall content blocks)
↓
1. Extract tool calls from message.content
2. Determine execution mode (parallel/sequential)
3. For each tool:
- Look up tool by name
- Prepare arguments (prepareArguments hook)
- Validate arguments (JSON Schema)
- beforeToolCall hook (can block)
- Execute tool (parallel or sequential)
- onUpdate stream (optional)
- afterToolCall hook (can override result)
- Create toolResult message
- Emit events
↓
4. Check shouldTerminateToolBatch
5. Return toolResult messages
Parallel vs Sequential Execution
Parallel Execution
async function executeToolCallsParallel(...) {
const finalizedCalls = []
// Preflight all tools sequentially
for (const toolCall of toolCalls) {
const preparation = await prepareToolCall(...)
if (preparation.kind === "immediate") {
finalizedCalls.push(preparation)
} else {
// Queue async execution
finalizedCalls.push(async () => {
const executed = await executePreparedToolCall(preparation, signal, emit)
const finalized = await finalizeExecutedToolCall(...)
return finalized
})
}
}
// Execute allowed tools concurrently
const orderedFinalizedCalls = await Promise.all(
finalizedCalls.map(entry => typeof entry === "function" ? entry() : Promise.resolve(entry))
)
// Emit toolResult messages in assistant source order
const messages = []
for (const finalized of orderedFinalizedCalls) {
const toolResultMessage = createToolResultMessage(finalized)
await emitToolResultMessage(toolResultMessage, emit)
messages.push(toolResultMessage)
}
return { messages, terminate: shouldTerminateToolBatch(orderedFinalizedCalls) }
}
Sequential Execution
async function executeToolCallsSequential(...) {
const finalizedCalls = []
const messages = []
for (const toolCall of toolCalls) {
const preparation = await prepareToolCall(...)
let finalized
if (preparation.kind === "immediate") {
finalized = preparation
} else {
const executed = await executePreparedToolCall(preparation, signal, emit)
finalized = await finalizeExecutedToolCall(...)
}
await emitToolExecutionEnd(finalized, emit)
const toolResultMessage = createToolResultMessage(finalized)
await emitToolResultMessage(toolResultMessage, emit)
finalizedCalls.push(finalized)
messages.push(toolResultMessage)
}
return { messages, terminate: shouldTerminateToolBatch(finalizedCalls) }
}
Tool Execution Stages
-
Preparation (
prepareToolCall)- Look up tool by name
- Call
prepareArgumentsif defined - Validate with
validateToolArguments - Call
beforeToolCallhook (can block with{ block: true }) - Return
PreparedToolCallorImmediateToolCallOutcome
-
Execution (
executePreparedToolCall)- Call
tool.execute(toolCallId, args, signal, onUpdate) - Stream partial results via
onUpdate - Handle errors, return
ExecutedToolCallOutcome
- Call
-
Finalization (
finalizeExecutedToolCall)- Call
afterToolCallhook (can override result) - Return
FinalizedToolCallOutcome
- Call
-
Message Creation (
createToolResultMessage)- Create
ToolResultMessage - Set
toolCallId,toolName,content,details,usage,isError,timestamp
- Create
Session Management
JSONL Storage Format
{"type":"session","version":3,"id":"abc123","timestamp":"2024-01-15T10:30:00.000Z",
"cwd":"/home/user/project","parentSession":"...","metadata":{}}
{"type":"message","id":"e001","parentId":null,"timestamp":"...","message":{...}}
{"type":"message","id":"e002","parentId":"e001","timestamp":"...","message":{...}}
{"type":"compaction","id":"e003","parentId":"e002","timestamp":"...",
"summary":"## Goal: ...\n...","firstKeptEntryId":"e001",
"tokensBefore":185000}
{"type":"leaf","id":"e004","parentId":"e003","timestamp":"...",
"targetId":"e002"}
Session File Structure:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ SESSION FILE (.jsonl) │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Line 1: Session Header (metadata) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ { │ │
│ │ "type": "session", │ │
│ │ "version": 3, │ │
│ │ "id": "abc123", │ │
│ │ "timestamp": "2024-01-15T10:30:00.000Z", │ │
│ │ "cwd": "/home/user/project", │ │
│ │ "parentSessionPath": "/path/to/parent.jsonl", │ │
│ │ "metadata": {} │ │
│ │ } │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Lines 2+: Entries (one per line) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ { │ │
│ │ "type": "message", │ │
│ │ "id": "e001", │ │
│ │ "parentId": null, │ │
│ │ "timestamp": "...", │ │
│ │ "message": { /* message data */ } │ │
│ │ } │ │
│ │ │ │
│ │ { │ │
│ │ "type": "compaction", │ │
│ │ "id": "e003", │ │
│ │ "parentId": "e002", │ │
│ │ "timestamp": "...", │ │
│ │ "summary": "...", │ │
│ │ "firstKeptEntryId": "e001", │ │
│ │ "tokensBefore": 185000 │ │
│ │ } │ │
│ │ │ │
│ │ { │ │
│ │ "type": "leaf", │ │
│ │ "id": "e004", │ │
│ │ "parentId": "e003", │ │
│ │ "timestamp": "...", │ │
│ │ "targetId": "e002" │ │
│ │ } │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
Entry Types
| Type | LLM Context? | Purpose |
|---|---|---|
message |
Yes | User, assistant, toolResult |
compaction |
Yes (as summary message) | Replaces compacted history |
branch_summary |
Yes (as summary message) | Summary of diverged branch |
leaf |
No | Points to current tree leaf |
thinking_level_change |
No | Tracking thinking level changes |
model_change |
No | Tracking model changes |
active_tools_change |
No | Tracking tool enable/disable |
custom |
No (unless projected) | App-defined data |
custom_message |
Yes | App-defined messages |
label |
No | Human-readable labels |
session_info |
No | Session name history |
Session Tree Structure:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ SESSION TREE (Branch Navigation) │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌───[e01]───┐
│ user: "a" │
└─────┬──────┘
▼
┌──────────┐
│ assist 1 │
└─────┬────┘
▼
┌──────────┐
│ toolCall │
└─────┬────┘
▼
┌──────────┐
│ toolRes 1│
└─────┬────┘
▼
┌──────────┐
│ user: "b"│ ← user navigates here
└─────┬────┘
│
┌─────┴─────┐
│ │
┌──────────┐ ┌──────────┐
│ user: "c" │ │ user: "d" │ ← branch point
└────┬─────┘ └────┬─────┘
│ │
┌────▼─────┐ ┌────▼─────┐
│ assist 2 │ │ assist 3 │ ← current leaf (d)
└──────────┘ └──────────┘
Context sent to LLM (when leaf is at "d"):
[compaction, user:b, user:d, assist:3]
When user navigates to "b":
1. Leaf moves from "d" back to "b"
2. Branch summary generated for diverged work ("c" → "assist 2")
3. Context: [compaction, branch_summary, turns 1-8]
4. New branch grows from "b"
Session Tree (Branching)
Sessions form a tree, not a linear log:
┌───[e01]───┐
│ user: "a" │
└─────┬──────┘
▼
┌──────────┐
│ assist 1 │
└─────┬────┘
▼
┌──────────┐
│ toolCall │
└─────┬────┘
▼
┌──────────┐
│ toolRes 1│
└─────┬────┘
▼
┌──────────┐
│ user: "b"│ ← user goes back here
└─────┬────┘
│
┌─────┴─────┐
│ │
┌──────────┐ ┌──────────┐
│ user: "c" │ │ user: "d" │ ← branch point
└────┬─────┘ └────┬─────┘
│ │
┌────▼─────┐ ┌────▼─────┐
│ assist 2 │ │ assist 3 │ ← current leaf (d)
└──────────┘ └──────────┘
Branch navigation:
- User navigates to a different point in history
- Leaf moves to the selected entry
- Branch summary generated for diverged work
- New work branches from the selected point
Session API
class Session {
async getBranch(fromId?: string): Promise<SessionTreeEntry[]>
async buildContext(options?: SessionContextBuildOptions): Promise<SessionContext>
async getLeafId(): Promise<string | null>
async setLeafId(id: string): Promise<void>
async appendMessage(message: AgentMessage): Promise<void>
async appendCompaction(summary: string, firstKeptEntryId: string, tokensBefore: number): Promise<void>
async appendBranchSummary(summary: string, fromId: string): Promise<void>
async appendLeaf(targetId: string): Promise<void>
}
Memory & Context Management
Token Estimation
Character-based heuristic (4 chars ≈ 1 token):
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ TOKEN ESTIMATION ALGORITHM │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ estimateTokens(message) │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ role === "user" │
│ content.length / 4 │
│ (images ≈ 4800 chars each) │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ role === "assistant" │
│ sum of all content blocks: │
│ • text blocks → text.length │
│ • thinking blocks → thinking.length │
│ • toolCall blocks → name.length + JSON.stringify(args).length │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ role === "toolResult" / "custom" │
│ content.length / 4 │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ role === "bashExecution" │
│ (command.length + output.length) / 4 │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ role === "compactionSummary" / "branchSummary" │
│ summary.length / 4 │
└─────────────────────────────────────────────────────────────────────────────────────┘
Compaction Strategy
Trigger condition:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ COMPACTION TRIGGER DECISION │
└─────────────────────────────────────────────────────────────────────────────────────┘
contextTokens > contextWindow - reserveTokens
Example (Claude with 200K context window):
Triggers when: contextTokens > 200000 - 16384 = 183616
Default Settings:
reserveTokens: 16384 (~16K for summary prompt + output)
keepRecentTokens: 20000 (~20K tokens of recent history to keep)
Cut Point Finding Algorithm:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ CUT POINT SELECTION │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ findCutPoint(entries, startIndex, endIndex, keepRecentTokens) │
└─────────────────────────────────────────────────────────────────────────────────────┘
Accumulate = 0
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Walk BACKWARD from endIndex │
│ ┌────────────────────────────────────────────────────────────────────────────┐ │
│ │ for i from endIndex-1 down to startIndex: │ │
│ │ │ │
│ │ // Skip invalid cut points │ │
│ │ if entry is toolResult message: │ │
│ │ continue // tool results stay with their call │ │
│ │ │ │
│ │ tokens = estimateTokens(entry) │ │
│ │ accumulated += tokens │ │
│ │ │ │
│ │ if accumulated >= keepRecentTokens: │ │
│ │ return i + 1 // Snap to nearest valid cut point │ │
│ └────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Valid Cut Points (safe to split): │
│ • user message │
│ • assistant message │
│ • custom message │
│ • branch_summary │
│ │
│ NOT Valid (tool results stay with their call): │
│ • toolResult message (skipped) │
└─────────────────────────────────────────────────────────────────────────────────────┘
Example:
┌────┐ ┌────┐ ┌────┐ ┌────┐ ┌────┐ ┌────┐ ┌────┐
│ u1 │ │ a1 │ │ tr1│ │ u2 │ │ a2 │ │ tr2│ │ u3 │
└────┘ └────┘ └────┘ └────┘ └────┘ └────┘ └────┘
▲ ▲ ▲
│ │ │
└── kept └── cut └── discarded
(~20K tokens) point history
Compaction Preparation:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ COMPACTION PREPARATION │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ prepareCompaction(branchEntries, settings) │
└─────────────────────────────────────────────────────────────────────────────────────┘
1. Find previous compaction (if any) → previousSummary
│
▼
2. Estimate tokens of current context
│
▼
3. findCutPoint() → firstKeptEntryId
│
▼
4. Split into 3 groups:
│
├─ messagesToSummarize: entries BEFORE cut point
│ (these become the summary)
│
├─ retainedTail: entries AFTER cut point
│ (these stay verbatim)
│
└─ turnPrefixMessages: if cut splits a turn
(the beginning of an interrupted turn)
│
▼
5. Extract file operations from messagesToSummarize:
└─ readFiles, modifiedFiles
Summary Generation:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ SUMMARY GENERATION │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ generateSummary(messages, previousSummary?) │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌──────────────────────────────────────────────────────────────────────────────┐ │
│ Has previousSummary? │ │
│ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ YES → UPDATE_SUMMARIZATION_PROMPT (iterative) │ │ │
│ │ ┌────────────────────────────────────────────────────────────────┐ │ │ │
│ │ │ <previous_summary> │ │ │ │
│ │ │ ${previousSummary} │ │ │ │
│ │ │ </previous_summary> │ │ │ │
│ │ │ │ │ │ │
│ │ │ <new_history> │ │ │ │
│ │ │ ${serializeConversation(messages)} │ │ │ │
│ │ │ </new_history> │ │ │ │
│ │ │ │ │ │ │
│ │ │ Update the previous summary with new progress: │ │ │ │
│ │ │ - Add completed tasks │ │ │ │
│ │ │ - Update progress │ │ │ │
│ │ │ - Add new goals │ │ │ │
│ │ │ - Keep existing information │ │ │ │
│ │ └────────────────────────────────────────────────────────────────┘ │ │ │
│ └──────────────────────────────────────────────────────────────────────┘ │ │
│ │ │
│ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ NO → FRESH_SUMMARIZATION_PROMPT (fresh) │ │ │
│ │ ┌────────────────────────────────────────────────────────────────┐ │ │ │
│ │ │ <conversation> │ │ │ │
│ │ │ ${serializeConversation(messages)} │ │ │ │
│ │ │ </conversation> │ │ │ │
│ │ │ │ │ │ │
│ │ │ Generate a summary: │ │ │ │
│ │ │ ## Goal │ │ │ │
│ │ │ ## Constraints & Preferences │ │ │ │
│ │ │ ## Progress │ │ │ │
│ │ │ ### Done │ │ │ │
│ │ │ ### In Progress │ │ │ │
│ │ │ ### Blocked │ │ │ │
│ │ │ ## Key Decisions │ │ │ │
│ │ │ ## Next Steps │ │ │ │
│ │ │ ## Critical Context │ │ │ │
│ │ │ Files read: [...] │ │ │ │
│ │ │ Files modified: [...] │ │ │ │
│ │ └────────────────────────────────────────────────────────────────┘ │ │ │
│ └──────────────────────────────────────────────────────────────────────┘ │ │
└──────────────────────────────────────────────────────────────────────────────┘ │
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ LLM generates structured summary: │
│ │
│ ## Goal │
│ - [What is the user trying to accomplish?] │
│ │
│ ## Constraints & Preferences │
│ - [Any constraints, preferences, or requirements] │
│ │
│ ## Progress │
│ ### Done │
│ - [x] [Completed tasks] │
│ ### In Progress │
│ - [ ] [Current work] │
│ ### Blocked │
│ - [Issues preventing progress] │
│ │
│ ## Key Decisions │
│ - **[Decision]**: [Brief rationale] │
│ │
│ ## Next Steps │
│ 1. [Ordered list of what should happen next] │
│ │
│ ## Critical Context │
│ - [Any data, examples, or references needed to continue] │
│ │
│ Files read: [src/index.ts, package.json] │
│ Files modified: [src/index.ts] │
└─────────────────────────────────────────────────────────────────────────────────────┘
Iterative Compaction:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ ITERATIVE COMPACTION │
└─────────────────────────────────────────────────────────────────────────────────────┘
Compaction 1 (at ~185K tokens):
┌──────────────────────────────────────────────────────────────────────────────┐ │
│ Summary: "## Goal: Build a login page..." │ │
│ firstKeptEntryId: "entry-003" │ │
└──────────────────────────────────────────────────────────────────────────────┘ │
Compaction 2 (at ~185K tokens again):
┌──────────────────────────────────────────────────────────────────────────────┐ │
│ previousSummary: "## Goal: Build a login page..." │ │
│ → UPDATE_SUMMARIZATION_PROMPT │ │
│ → PRESERVES existing information │ │
│ → ADDS new progress (move "In Progress" → "Done") │ │
│ → NEW summary: "## Goal: Build a login page... Add OAuth..." │ │
│ firstKeptEntryId: "entry-003" (same boundary) │ │
└──────────────────────────────────────────────────────────────────────────────┘ │
Result:
• History before cut point replaced by summary
• Cut point stays at same location
• Summary grows incrementally with new progress
• Token budget maintained (~25K tokens after compaction)
Event System
Event Types
Agent Lifecycle
agent_start: Agent begins processingagent_end: Final event for the run
Turn Lifecycle
turn_start: New turn beginsturn_end: Turn completes with assistant message and tool results
Message Lifecycle
message_start: Any message beginsmessage_update: Assistant only. IncludesassistantMessageEventwith deltamessage_end: Message completes
Tool Execution Lifecycle
tool_execution_start: Tool beginstool_execution_update: Tool streams progresstool_execution_end: Tool completes
Complete Event Flow:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ COMPLETE EVENT SEQUENCE │
│ (Agent with Tool Calls - Product Query) │
└─────────────────────────────────────────────────────────────────────────────────────┘
agent_start
↓
turn_start
↓
message_start (user message: "what product do you have in stock?")
message_end (user message)
↓
message_start (assistant message - streaming from LLM)
message_update (text delta: "We have")
message_update (toolcall delta: {"name":"read","arguments":{...}})
message_update (toolcall end)
message_end (assistant message with tool calls)
↓
tool_execution_start (tool call: read products.json)
tool_execution_end (tool call: read complete)
message_start (toolResult message)
message_end (toolResult message)
↓
turn_end
↓
[Inner loop continues: send tool result to LLM]
↓
turn_start
↓
message_start (assistant message - streaming from LLM)
message_update (text delta: "We have 15 products in stock:")
message_update (text delta: "• Product A - $29.99")
message_update (text delta: "• Product B - $49.99")
message_end (final assistant message)
↓
turn_end
↓
agent_end
Event Sequence Without Tools:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ EVENT FLOW (No Tool Calls) │
└─────────────────────────────────────────────────────────────────────────────────────┘
prompt("Hello!")
↓
agent_start
↓
turn_start
↓
message_start (user)
message_end (user)
↓
message_start (assistant - streaming)
message_update (text_delta)
message_update (text_delta)
message_end (assistant)
↓
turn_end
↓
agent_end
Event Sequence with Steering Messages:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ EVENT FLOW (With Steering Messages) │
└─────────────────────────────────────────────────────────────────────────────────────┘
Turn 1: Agent executing tool calls...
↓
turn_end
↓
[Agent would stop, but...]
↓
steer({ role: "user", content: "Stop! Do this instead." })
↓
turn_start (steering)
↓
message_start (steering message)
message_end (steering message)
↓
message_start (assistant response to steering)
message_update (text_delta)
message_end (assistant)
↓
turn_end
↓
agent_end
Event Flow with Tools
agent_start
↓
turn_start
↓
message_start { user message }
message_end { user message }
↓
message_start { assistant message - streaming }
message_update { text_delta: "We have" }
message_update { toolcall_delta: {"name":"read","arguments":{...}} }
message_update { toolcall_end }
message_end { assistant message with tool calls }
↓
tool_execution_start { tool call: read products.json }
tool_execution_end { tool call: complete }
message_start { toolResult message }
message_end { toolResult message }
↓
turn_end { message, toolResults: [toolResult] }
↓
[Inner loop continues: send tool result to LLM]
↓
turn_start
↓
message_start { assistant message }
message_update { text_delta: "We have 15 products" }
message_end { final assistant message }
↓
turn_end
↓
agent_end
Subscription Model
const unsubscribe = agent.subscribe(async (event, signal) => {
switch (event.type) {
case "message_update":
if (event.assistantMessageEvent.type === "text_delta") {
// Stream text to UI
process.stdout.write(event.assistantMessageEvent.delta)
}
break
case "agent_end":
// Cleanup, save state, etc.
await flushSessionState(signal)
}
})
Subscription semantics:
- Listeners are awaited in subscription order
agent_endlisteners are included in run settlement- Agent becomes idle only after all awaited listeners finish
- All listeners receive the active abort signal
Hook System
Hook Points
| Hook | Location | Purpose |
|---|---|---|
convertToLlm |
agent.ts:99 | Transform messages before LLM call |
transformContext |
agent.ts:100 | Modify context (pruning, injection) |
beforeToolCall |
agent.ts:105 | Block or modify tool execution |
afterToolCall |
agent.ts:106 | Override tool results |
prepareNextTurn |
agent.ts:107 | Dynamic context/model updates |
shouldStopAfterTurn |
agent-loop.ts | Graceful termination |
getSteeringMessages |
agent.ts:114 | Inject messages mid-turn |
getFollowUpMessages |
agent.ts:115 | Queue follow-up messages |
Hook Flow Diagram:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ HOOK EXECUTION FLOW │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Prompt Execution │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ agent.prompt("Hello!") │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Agent Class │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ runPromptMessages() │ │
│ │ • normalizePromptInput() │ │
│ │ • createContextSnapshot() │ │
│ │ • createLoopConfig() │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Agent Loop │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ runLoop() │ │
│ │ • streamAssistantResponse() │ │
│ │ • executeToolCalls() │ │
│ │ • turn_end │ │
│ │ • prepareNextTurn hook │ │
│ │ • shouldStopAfterTurn hook │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ LLM Call Boundary │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ streamAssistantResponse() │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ 1. transformContext hook (optional) │ │
│ │ (pruning, context injection) │ │
│ │ │ │
│ │ 2. convertToLlm hook │ │
│ │ (filter custom messages, convert to LLM format) │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Tool Execution │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ executeToolCalls() │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ For each tool call: │ │
│ │ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ prepareToolCall() │ │ │
│ │ │ • validateToolArguments() │ │ │
│ │ │ • beforeToolCall hook │ │ │
│ │ │ (can block: return { block: true }) │ │ │
│ │ │ │ │ │
│ │ │ execute() │ │ │
│ │ │ • onUpdate() (stream partial results) │ │ │
│ │ │ │ │ │
│ │ │ finalizeExecutedToolCall() │ │ │
│ │ │ • afterToolCall hook │ │ │
│ │ │ (can override: return { content: [...], details: {...} }) │ │ │
│ │ └──────────────────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ prepareNextTurn │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ prepareNextTurn hook │ │
│ │ • Can return updated context │ │
│ │ • Can return updated model │ │
│ │ • Can return updated thinkingLevel │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ shouldStopAfterTurn │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ shouldStopAfterTurn hook │ │
│ │ • Returns true to stop agent gracefully │ │
│ │ • Used for context management before compaction │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
BeforeToolCall Hook:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ BEFORE_TOOL_CALL HOOK │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ beforeToolCall(context, signal) │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌──────────────────────────────────────────────────────────────────────────────┐ │
│ BeforeToolCallContext: │ │
│ • assistantMessage: AssistantMessage │ │
│ • toolCall: AgentToolCall │ │
│ • args: unknown (validated) │ │
│ • context: AgentContext │ │
└──────────────────────────────────────────────────────────────────────────────┘ │
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Return Value │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ { block: true, reason?: string } │ │
│ │ → Tool execution blocked, error result emitted │ │
│ │ │ │
│ │ undefined │ │
│ │ → Tool execution proceeds │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
Example (blocking bash tool):
beforeToolCall: async ({ toolCall, context }) => {
if (toolCall.name === "bash") {
return { block: true, reason: "bash is disabled" };
}
}
AfterToolCall Hook:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ AFTER_TOOL_CALL HOOK │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ afterToolCall(context, signal) │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌──────────────────────────────────────────────────────────────────────────────┐ │
│ AfterToolCallContext: │ │
│ • assistantMessage: AssistantMessage │ │
│ • toolCall: AgentToolCall │ │
│ • args: unknown (validated) │ │
│ • result: AgentToolResult<any> │ │
│ • isError: boolean │ │
│ • context: AgentContext │ │
└──────────────────────────────────────────────────────────────────────────────┘ │
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Return Value (Partial Override) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ { │ │
│ │ content?: (TextContent | ImageContent)[] │ │
│ │ details?: unknown │ │
│ │ isError?: boolean │ │
│ │ usage?: Usage │ │
│ │ terminate?: boolean // Hint to stop after batch │ │
│ │ } │ │
│ │ │ │
│ │ Omitted fields keep original values │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
Example (terminate on success):
afterToolCall: async ({ toolCall, result, isError }) => {
if (toolCall.name === "notify_done" && !isError) {
return { terminate: true };
}
}
PrepareNextTurn Hook:
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ PREPARE_NEXT_TURN HOOK │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ prepareNextTurn(context) │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌──────────────────────────────────────────────────────────────────────────────┐ │
│ PrepareNextTurnContext (extends ShouldStopAfterTurnContext): │ │
│ • message: AssistantMessage │ │
│ • toolResults: ToolResultMessage[] │ │
│ • context: AgentContext │ │
│ • newMessages: AgentMessage[] │ │
└──────────────────────────────────────────────────────────────────────────────┘ │
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Return Value │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ { │ │
│ │ context?: AgentContext │ │
│ │ model?: Model<any> │ │
│ │ thinkingLevel?: ThinkingLevel │ │
│ │ } │ │
│ │ │ │
│ │ undefined │ │
│ │ → Keep current context/config │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
Example (switch to smaller model for next turn):
prepareNextTurn: async ({ message, context }) => {
if (context.messages.length > 100) {
return { model: getSmallerModel() };
}
}
Implementation Guide for Julia
Architecture Overview
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ JULIA AGENT IMPLEMENTATION STRUCTURE │
└─────────────────────────────────────────────────────────────────────────────────────┘
pi-agent/
├── src/
│ ├── agent.jl # Core Agent class (stateful wrapper)
│ │ ├── Agent structure
│ │ ├── State management (_state)
│ │ ├── Queue management (steering, follow-up)
│ │ ├── Event subscription
│ │ └── Methods: prompt(), continue(), reset(), subscribe()
│ │
│ ├── agent_loop.jl # Low-level execution loop
│ │ ├── run_agent_loop()
│ │ ├── run_agent_loop_continue()
│ │ ├── run_loop()
│ │ ├── stream_assistant_response()
│ │ └── execute_tool_calls() (parallel/sequential)
│ │
│ ├── messages.jl # Message types and conversion
│ │ ├── Message types (User, Assistant, ToolResult)
│ │ ├── Custom message types
│ │ ├── convert_to_llm()
│ │ └─ create_summary_messages()
│ │
│ ├── tools.jl # Tool execution
│ │ ├── Tool structure
│ │ ├── prepare_tool_call()
│ │ ├── execute_prepared_tool_call()
│ │ ├── execute_tool_calls_parallel()
│ │ └── execute_tool_calls_sequential()
│ │
│ ├── session.jl # Session persistence
│ │ ├── Session structure
│ │ ├── JSONL storage
│ │ ├── append_entry()
│ │ ├── get_branch()
│ │ └── build_context()
│ │
│ ├── compaction.jl # Memory management
│ │ ├── estimate_tokens()
│ │ ├── should_compact()
│ │ ├── find_cut_point()
│ │ ├── prepare_compaction()
│ │ ├── generate_summary()
│ │ └── compact()
│ │
│ ├── events.jl # Event system
│ │ ├── Event types (AgentStart, TurnEnd, etc.)
│ │ └── emit_event()
│ │
│ ├── hooks.jl # Hook system
│ │ ├── Hook context types
│ │ ├── Hook return types
│ │ └── default hooks
│ │
│ ├── types.jl # Type definitions
│ │ ├── ThinkingLevel enum
│ │ ├── Tool structure
│ │ ├── Message types
│ │ └── Context structures
│ │
│ ├── stream.jl # Stream utilities
│ │ ├── Stream struct
│ │ ├── next_event()
│ │ └── stream_simple()
│ │
│ └── utils/
│ ├── truncate.jl # Text truncation
│ └── shell_output.jl # Shell command execution
│
├── test/
│ ├── agent_test.jl
│ ├── tools_test.jl
│ └── session_test.jl
│
└── project.toml
Step 1: Type Definitions (types.jl)
# thinking_level.jl
@enum ThinkingLevel begin
THINKING_OFF
THINKING_MINIMAL
THINKING_LOW
THINKING_MEDIUM
THINKING_HIGH
THINKING_XHIGH
THINKING_MAX
end
# tool.jl
mutable struct Tool{TParameters, TDetails}
name::String
label::String
description::String
parameters::TParameters
execute::Function
prepare_arguments::Union{Function, Nothing}
execution_mode::Symbol # :parallel or :sequential
end
# message.jl
abstract type Message end
struct UserMessage <: Message
content::Vector{Union{TextContent, ImageContent}}
timestamp::Int64
end
struct AssistantMessage <: Message
content::Vector{Union{TextContent, ToolCall, Thinking}}
api::String
provider::String
model::String
usage::Usage
stop_reason::String
error_message::Union{String, Nothing}
timestamp::Int64
end
struct ToolResultMessage <: Message
tool_call_id::String
tool_name::String
content::Vector{Union{TextContent, ImageContent}}
details::Any
usage::Union{Usage, Nothing}
is_error::Bool
timestamp::Int64
end
struct CustomMessage <: Message
custom_type::String
content::Union{String, Vector{Union{TextContent, ImageContent}}}
display::Bool
details::Any
timestamp::Int64
end
struct AgentMessage
message::Union{UserMessage, AssistantMessage, ToolResultMessage, CustomMessage}
end
# context.jl
struct AgentContext
system_prompt::String
messages::Vector{AgentMessage}
tools::Union{Vector{Tool}, Nothing}
end
Step 2: Message System (messages.jl)
# messages.jl
const COMPACTION_SUMMARY_PREFIX = """
The conversation history before this point was compacted into the following summary:
<summary>
"""
const COMPACTION_SUMMARY_SUFFIX = """
</summary>
"""
const BRANCH_SUMMARY_PREFIX = """
The following is a summary of a branch that this conversation came back from:
<summary>
"""
const BRANCH_SUMMARY_SUFFIX = """
</summary>
"""
function convert_to_llm(messages::Vector{AgentMessage})
llm_messages = Vector{Any}()
for msg in messages
if msg.message isa UserMessage
push!(llm_messages, msg.message)
elseif msg.message isa AssistantMessage
push!(llm_messages, msg.message)
elseif msg.message isa ToolResultMessage
push!(llm_messages, msg.message)
elseif msg.message isa CustomMessage
# Convert custom to user message
content = if msg.message.content isa String
[TextContent(msg.message.content)]
else
msg.message.content
end
push!(llm_messages, UserMessage(content, msg.message.timestamp))
end
end
return llm_messages
end
function create_branch_summary_message(summary::String, from_id::String, timestamp::String)
BranchSummaryMessage(summary, from_id, parse(Int64, timestamp))
end
function create_compaction_summary_message(summary::String, tokens_before::Int, timestamp::String)
CompactionSummaryMessage(summary, tokens_before, parse(Int64, timestamp))
end
Step 3: Tool Execution (tools.jl)
# tools.jl
struct ToolExecutionResult{TDetails}
content::Vector{Union{TextContent, ImageContent}}
details::TDetails
usage::Union{Usage, Nothing}
added_tool_names::Vector{String}
terminate::Bool
end
struct PreparedToolCall
tool_call::ToolCall
tool::Tool
args::Any
end
struct ImmediateToolOutcome
result::ToolExecutionResult
is_error::Bool
end
struct ExecutedToolOutcome
result::ToolExecutionResult
is_error::Bool
end
struct FinalizedToolOutcome
tool_call::ToolCall
result::ToolExecutionResult
is_error::Bool
end
function prepare_tool_call(
current_context::AgentContext,
assistant_message::AssistantMessage,
tool_call::ToolCall,
config::AgentLoopConfig,
signal::Union{AbortSignal, Nothing}
)
tool = find_tool(current_context.tools, tool_call.name)
if tool === nothing
return ImmediateToolOutcome(
ToolExecutionResult([TextContent("Tool $(tool_call.name) not found")], Dict(), nothing, [], false),
true
)
end
try
prepared_tool_call = prepare_tool_call_arguments(tool, tool_call)
validated_args = validate_tool_arguments(tool, prepared_tool_call)
if config.before_tool_call !== nothing
before_result = config.before_tool_call(
BeforeToolCallContext(assistant_message, tool_call, validated_args, current_context),
signal
)
if before_result.block
return ImmediateToolOutcome(
ToolExecutionResult([TextContent(before_result.reason)], Dict(), nothing, [], false),
true
)
end
end
return PreparedToolCall(tool_call, tool, validated_args)
catch error
return ImmediateToolOutcome(
ToolExecutionResult([TextContent(error.message)], Dict(), nothing, [], false),
true
)
end
end
function execute_prepared_tool_call(
prepared::PreparedToolCall,
signal::Union{AbortSignal, Nothing},
emit::Function
)
update_events = Vector{Future}()
accepting_updates = true
try
result = prepared.tool.execute(
prepared.tool_call.id,
prepared.args,
signal,
partial_result -> begin
if !accepting_updates
return
end
push!(update_events, Threads.@spawn begin
emit({
type: "tool_execution_update",
tool_call_id: prepared.tool_call.id,
tool_name: prepared.tool_call.name,
args: prepared.tool_call.arguments,
partial_result: partial_result
})
end)
end
)
accepting_updates = false
wait(update_events)
return ExecutedToolOutcome(result, false)
catch error
accepting_updates = false
wait(update_events)
return ExecutedToolOutcome(
ToolExecutionResult([TextContent(error.message)], Dict(), nothing, [], false),
true
)
finally
accepting_updates = false
end
end
function execute_tool_calls_parallel(...)
finalized_calls = Vector{Any}()
# Preflight all tools sequentially
for tool_call in tool_calls
preparation = prepare_tool_call(...)
if preparation isa ImmediateToolOutcome
push!(finalized_calls, preparation)
else
push!(finalized_calls, Threads.@async begin
executed = execute_prepared_tool_call(preparation, signal, emit)
finalize_executed_tool_call(...)
end)
end
end
# Execute allowed tools concurrently
ordered_finalized = wait(finalized_calls)
# Emit toolResult messages
messages = Vector{ToolResultMessage}()
for finalized in ordered_finalized
tool_result_message = create_tool_result_message(finalized)
emit({ type: "message_start", message: tool_result_message })
emit({ type: "message_end", message: tool_result_message })
push!(messages, tool_result_message)
end
return { messages: messages, terminate: should_terminate_tool_batch(ordered_finalized) }
end
Step 4: Agent Loop (agent_loop.jl)
# agent_loop.jl
function run_agent_loop(
prompts::Vector{AgentMessage},
context::AgentContext,
config::AgentLoopConfig,
emit::Function,
signal::Union{AbortSignal, Nothing},
stream_fn::Function
)
new_messages = copy(prompts)
current_context = AgentContext(
context.system_prompt,
vcat(context.messages, prompts),
context.tools
)
emit({ type: "agent_start" })
emit({ type: "turn_start" })
for prompt in prompts
emit({ type: "message_start", message: prompt })
emit({ type: "message_end", message: prompt })
end
run_loop(current_context, new_messages, config, signal, emit, stream_fn)
return new_messages
end
function run_loop(
initial_context::AgentContext,
new_messages::Vector{AgentMessage},
initial_config::AgentLoopConfig,
signal::Union{AbortSignal, Nothing},
emit::Function,
stream_fn::Function
)
current_context = initial_context
config = initial_config
first_turn = true
pending_messages = config.get_steering_messages !== nothing ? config.get_steering_messages() : []
while true
has_more_tool_calls = true
while has_more_tool_calls || !isempty(pending_messages)
if !first_turn
emit({ type: "turn_start" })
else
first_turn = false
end
# Process pending messages
if !isempty(pending_messages)
for msg in pending_messages
emit({ type: "message_start", message: msg })
emit({ type: "message_end", message: msg })
push!(current_context.messages, msg)
push!(new_messages, msg)
end
pending_messages = []
end
# Stream assistant response
message = stream_assistant_response(current_context, config, signal, emit, stream_fn)
push!(new_messages, message)
# Check for errors
if message.stop_reason == "error" || message.stop_reason == "aborted"
emit({ type: "turn_end", message: message, tool_results: [] })
emit({ type: "agent_end", messages: new_messages })
return
end
# Extract tool calls
tool_calls = filter(c -> c.type == "toolCall", message.content)
tool_results = []
has_more_tool_calls = false
if !isempty(tool_calls)
executed_batch = execute_tool_calls(current_context, message, config, signal, emit)
append!(tool_results, executed_batch.messages)
has_more_tool_calls = !executed_batch.terminate
for result in tool_results
push!(current_context.messages, result)
push!(new_messages, result)
end
end
emit({ type: "turn_end", message: message, tool_results: tool_results })
# Check prepareNextTurn hook
if config.prepare_next_turn !== nothing
next_turn_snapshot = config.prepare_next_turn({
message: message,
tool_results: tool_results,
context: current_context,
new_messages: new_messages
})
if next_turn_snapshot !== nothing
current_context = next_turn_snapshot.context !== nothing ? next_turn_snapshot.context : current_context
config = merge(config, model = next_turn_snapshot.model !== nothing ? next_turn_snapshot.model : config.model)
end
end
# Check shouldStopAfterTurn hook
if config.should_stop_after_turn !== nothing && config.should_stop_after_turn({
message: message,
tool_results: tool_results,
context: current_context,
new_messages: new_messages
})
emit({ type: "agent_end", messages: new_messages })
return
end
# Get steering messages
pending_messages = config.get_steering_messages !== nothing ? config.get_steering_messages() : []
end
# Check follow-up messages
follow_up_messages = config.get_follow_up_messages !== nothing ? config.get_follow_up_messages() : []
if !isempty(follow_up_messages)
pending_messages = follow_up_messages
continue
end
break
end
emit({ type: "agent_end", messages: new_messages })
end
function stream_assistant_response(context, config, signal, emit, stream_fn)
# 1. Transform context (optional)
messages = context.messages
if config.transform_context !== nothing
messages = config.transform_context(messages, signal)
end
# 2. Convert to LLM format
llm_messages = config.convert_to_llm(messages)
# 3. Build LLM context
llm_context = Context(
system_prompt = context.system_prompt,
messages = llm_messages,
tools = context.tools
)
# 4. Resolve API key
resolved_api_key = config.get_api_key !== nothing ? config.get_api_key(config.model.provider) : config.api_key
# 5. Call stream function
response = stream_fn(config.model, llm_context, merge(config, api_key = resolved_api_key, signal = signal))
# 6. Stream events
partial_message = nothing
added_partial = false
for event in response
if event.type == "start"
partial_message = event.partial
push!(context.messages, partial_message)
added_partial = true
emit({ type: "message_start", message: deepcopy(partial_message) })
elseif event.type in ["text_start", "text_delta", "text_end", "thinking_start", "thinking_delta", "thinking_end", "toolcall_start", "toolcall_delta", "toolcall_end"]
if partial_message !== nothing
partial_message = event.partial
context.messages[end] = partial_message
emit({
type: "message_update",
assistant_message_event: event,
message: deepcopy(partial_message)
})
end
elseif event.type == "done" || event.type == "error"
final_message = response.result()
if added_partial
context.messages[end] = final_message
else
push!(context.messages, final_message)
emit({ type: "message_start", message: deepcopy(final_message) })
end
emit({ type: "message_end", message: final_message })
return final_message
end
end
end
Step 5: Agent Class (agent.jl)
# agent.jl
mutable struct ActiveRun
promise::Promise
resolve::Function
abort_controller::AbortController
end
mutable struct Agent
_state::MutableAgentState
listeners::Set{Function}
steering_queue::PendingMessageQueue
follow_up_queue::PendingMessageQueue
convert_to_llm::Function
transform_context::Union{Function, Nothing}
stream_function::Function
get_api_key::Union{Function, Nothing}
on_payload::Union{Function, Nothing}
on_response::Union{Function, Nothing}
before_tool_call::Union{Function, Nothing}
after_tool_call::Union{Function, Nothing}
prepare_next_turn::Union{Function, Nothing}
prepare_next_turn_with_context::Union{Function, Nothing}
active_run::Union{ActiveRun, Nothing}
session_id::Union{String, Nothing}
thinking_budgets::Union{ThinkingBudgets, Nothing}
transport::Symbol
max_retry_delay_ms::Union{Int, Nothing}
tool_execution::Symbol
end
struct MutableAgentState
system_prompt::String
model::Model
thinking_level::ThinkingLevel
tools::Vector{Tool}
messages::Vector{AgentMessage}
is_streaming::Bool
streaming_message::Union{AssistantMessage, Nothing}
pending_tool_calls::Set{String}
error_message::Union{String, Nothing}
end
function Agent(; kwargs...)
state = MutableAgentState(
kwargs[:initial_state].system_prompt,
kwargs[:initial_state].model,
kwargs[:initial_state].thinking_level,
copy(kwargs[:initial_state].tools),
copy(kwargs[:initial_state].messages),
false,
nothing,
Set{String}(),
nothing
)
Agent(
state,
Set{Function}(),
PendingMessageQueue(kwargs[:steering_mode] === nothing ? "one-at-a-time" : kwargs[:steering_mode]),
PendingMessageQueue(kwargs[:follow_up_mode] === nothing ? "one-at-a-time" : kwargs[:follow_up_mode]),
kwargs[:convert_to_llm] !== nothing ? kwargs[:convert_to_llm] : default_convert_to_llm,
kwargs[:transform_context],
kwargs[:stream_fn],
kwargs[:get_api_key],
kwargs[:on_payload],
kwargs[:on_response],
kwargs[:before_tool_call],
kwargs[:after_tool_call],
kwargs[:prepare_next_turn],
kwargs[:prepare_next_turn_with_context],
nothing,
kwargs[:session_id],
kwargs[:thinking_budgets],
kwargs[:transport] === nothing ? :auto : kwargs[:transport],
kwargs[:max_retry_delay_ms],
kwargs[:tool_execution] === nothing ? :parallel : kwargs[:tool_execution]
)
end
function subscribe(agent::Agent, listener::Function)
push!(agent.listeners, listener)
return () -> delete!(agent.listeners, listener)
end
function prompt(agent::Agent, input::String)
if agent.active_run !== nothing
throw(ErrorException("Agent is already processing."))
end
messages = normalize_prompt_input(input)
run_prompt_messages(agent, messages)
end
function prompt(agent::Agent, messages::Vector{AgentMessage})
if agent.active_run !== nothing
throw(ErrorException("Agent is already processing."))
end
run_prompt_messages(agent, messages)
end
function continue(agent::Agent)
if agent.active_run !== nothing
throw(ErrorException("Agent is already processing."))
end
last_message = agent._state.messages[end]
if last_message.message isa AssistantMessage
queued_steering = drain(agent.steering_queue)
if !isempty(queued_steering)
run_prompt_messages(agent, queued_steering, skip_initial_steering_poll = true)
return
end
queued_follow_ups = drain(agent.follow_up_queue)
if !isempty(queued_follow_ups)
run_prompt_messages(agent, queued_follow_ups)
return
end
throw(ErrorException("Cannot continue from message role: assistant"))
end
run_continuation(agent)
end
function run_prompt_messages(agent::Agent, messages::Vector{AgentMessage}, options = Dict())
run_with_lifecycle(agent) do signal
run_agent_loop(
messages,
create_context_snapshot(agent),
create_loop_config(agent, options),
event -> process_events(agent, event),
signal,
agent.stream_function
)
end
end
function create_context_snapshot(agent::Agent)
AgentContext(
agent._state.system_prompt,
copy(agent._state.messages),
copy(agent._state.tools)
)
end
function create_loop_config(agent::Agent, options = Dict())
skip_initial_steering_poll = get(options, :skip_initial_steering_poll, false)
AgentLoopConfig(
model = agent._state.model,
reasoning = agent._state.thinking_level == THINKING_OFF ? nothing : agent._state.thinking_level,
session_id = agent.session_id,
on_payload = agent.on_payload,
on_response = agent.on_response,
transport = agent.transport,
thinking_budgets = agent.thinking_budgets,
max_retry_delay_ms = agent.max_retry_delay_ms,
tool_execution = agent.tool_execution,
before_tool_call = agent.before_tool_call,
after_tool_call = agent.after_tool_call,
prepare_next_turn = create_prepare_next_turn(agent),
convert_to_llm = agent.convert_to_llm,
transform_context = agent.transform_context,
get_api_key = agent.get_api_key,
get_steering_messages = () -> begin
if skip_initial_steering_poll
skip_initial_steering_poll = false
return []
end
return drain(agent.steering_queue)
end,
get_follow_up_messages = () -> drain(agent.follow_up_queue)
)
end
function create_prepare_next_turn(agent::Agent)
function prepare(context, signal)
if agent.prepare_next_turn_with_context !== nothing
return agent.prepare_next_turn_with_context(context, signal)
end
if agent.prepare_next_turn !== nothing
return agent.prepare_next_turn(signal)
end
return nothing
end
return prepare
end
function run_with_lifecycle(agent::Agent, executor::Function)
abort_controller = AbortController()
promise = Promise{Void}()
resolve = () -> nothing
function set_resolve(value)
resolve = value
if !isready(promise)
put!(promise, nothing)
end
end
agent.active_run = ActiveRun(promise, set_resolve, abort_controller)
agent._state.is_streaming = true
agent._state.streaming_message = nothing
agent._state.error_message = nothing
try
executor(abort_controller.signal)
catch error
handle_run_failure(agent, error, abort_controller.signal.aborted)
finally
finish_run(agent)
end
end
function handle_run_failure(agent::Agent, error, aborted)
failure_message = AssistantMessage(
[TextContent("")],
agent._state.model.api,
agent._state.model.provider,
agent._state.model.id,
EMPTY_USAGE,
aborted ? "aborted" : "error",
error isa ErrorException ? error.message : string(error),
Dates.datetime2timestamp(Dates.now())
)
process_events(agent, { type: "message_start", message: failure_message })
process_events(agent, { type: "message_end", message: failure_message })
process_events(agent, { type: "turn_end", message: failure_message, tool_results: [] })
process_events(agent, { type: "agent_end", messages: [failure_message] })
end
function finish_run(agent::Agent)
agent._state.is_streaming = false
agent._state.streaming_message = nothing
agent._state.pending_tool_calls = Set{String}()
if agent.active_run !== nothing
agent.active_run.resolve()
agent.active_run = nothing
end
end
function process_events(agent::Agent, event)
if event.type == "message_start"
agent._state.streaming_message = event.message
elseif event.type == "message_update"
agent._state.streaming_message = event.message
elseif event.type == "message_end"
agent._state.streaming_message = nothing
push!(agent._state.messages, event.message)
elseif event.type == "tool_execution_start"
push!(agent._state.pending_tool_calls, event.tool_call_id)
elseif event.type == "tool_execution_end"
delete!(agent._state.pending_tool_calls, event.tool_call_id)
elseif event.type == "turn_end"
if event.message.message isa AssistantMessage && event.message.message.error_message !== nothing
agent._state.error_message = event.message.message.error_message
end
elseif event.type == "agent_end"
agent._state.streaming_message = nothing
end
signal = agent.active_run !== nothing ? agent.active_run.abort_controller.signal : nothing
for listener in agent.listeners
Threads.@spawn listener(event, signal)
end
end
function normalize_prompt_input(input::String)
content = [TextContent(input)]
return [AgentMessage(UserMessage(content, Dates.datetime2timestamp(Dates.now())))]
end
function normalize_prompt_input(messages::Vector{AgentMessage})
return messages
end
Step 6: Session Persistence (session.jl)
# session.jl
struct SessionEntry
type::String
id::String
parent_id::Union{String, Nothing}
timestamp::Int64
# Dynamic fields based on type
end
struct MessageEntry <: SessionEntry
message::AgentMessage
end
struct CompactionEntry <: SessionEntry
summary::String
first_kept_entry_id::Union{String, Nothing}
tokens_before::Int
end
struct BranchSummaryEntry <: SessionEntry
summary::String
from_id::String
end
struct LeafEntry <: SessionEntry
target_id::String
end
struct SessionStorage{TMetadata}
file_path::String
metadata::TMetadata
entries::Dict{String, SessionEntry}
leaf_id::Union{String, Nothing}
end
function create_session(file_system::FileSystem, cwd::String, id::String, timestamp::DateTime)
session_dir = join_path(file_system.sessions_root, encode_cwd(cwd))
create_dir(session_dir, recursive = true)
file_path = join_path(session_dir, "$(replace(timestamp, r"[:\.]" => "-"))_$id.jsonl")
metadata = SessionMetadata(
path = file_path,
cwd = cwd,
session_id = id,
parent_session_path = nothing,
created_at = timestamp
)
storage = SessionStorage(file_path, metadata, Dict{String, SessionEntry}(), nothing)
# Write header
header = Dict(
"type" => "session",
"version" => 3,
"id" => id,
"timestamp" => string(timestamp),
"cwd" => cwd,
"metadata" => Dict{String, Any}()
)
open(file_path, "w") do f
write(f, JSON.json(header))
write(f, "\n")
end
return storage
end
function append_message(storage::SessionStorage, message::AgentMessage)
entry_id = uuid7()
timestamp = Dates.datetime2timestamp(Dates.now())
entry = Dict(
"type" => "message",
"id" => entry_id,
"parentId" => storage.leaf_id,
"timestamp" => string(DateTime(timestamp)),
"message" => message_to_dict(message)
)
open(storage.file_path, "a") do f
write(f, JSON.json(entry))
write(f, "\n")
end
storage.entries[entry_id] = entry
storage.leaf_id = entry_id
end
function append_compaction(storage::SessionStorage, summary::String, first_kept_entry_id::String, tokens_before::Int)
entry_id = uuid7()
timestamp = Dates.datetime2timestamp(Dates.now())
entry = Dict(
"type" => "compaction",
"id" => entry_id,
"parentId" => storage.leaf_id,
"timestamp" => string(DateTime(timestamp)),
"summary" => summary,
"firstKeptEntryId" => first_kept_entry_id,
"tokensBefore" => tokens_before
)
open(storage.file_path, "a") do f
write(f, JSON.json(entry))
write(f, "\n")
end
storage.entries[entry_id] = entry
storage.leaf_id = entry_id
end
function get_branch(storage::SessionStorage, from_id::Union{String, Nothing} = nothing)
leaf_id = from_id !== nothing ? from_id : storage.leaf_id
if leaf_id === nothing
return []
end
entries = Vector{SessionEntry}()
current_id = leaf_id
while current_id !== nothing
entry = get(storage.entries, current_id, nothing)
if entry === nothing
break
end
push!(entries, entry)
if entry.type == "compaction" && entry.first_kept_entry_id !== nothing
current_id = entry.first_kept_entry_id
else
current_id = entry.parent_id
end
end
return reverse(entries)
end
Step 7: Compaction (compaction.jl)
# compaction.jl
const DEFAULT_COMPACTION_SETTINGS = Dict(
:enabled => true,
:reserve_tokens => 16384,
:keep_recent_tokens => 20000
)
function estimate_tokens(message::AgentMessage)
if message.message isa UserMessage
return length(join(message.message.content)) / 4
elseif message.message isa AssistantMessage
total = 0
for block in message.message.content
if block.type == "text"
total += length(block.text)
elseif block.type == "thinking"
total += length(block.thinking)
elseif block.type == "toolCall"
total += length(block.name) + length(JSON.json(block.arguments))
end
end
return total
elseif message.message isa ToolResultMessage
return length(join(message.message.content)) / 4
elseif message.message isa CustomMessage
if message.message.content isa String
return length(message.message.content) / 4
else
return length(join(message.message.content)) / 4
end
else
return 0
end
end
function estimate_context_tokens(messages::Vector{AgentMessage})
total = 0
for message in messages
total += estimate_tokens(message)
end
return total
end
function should_compact(context_tokens::Int, context_window::Int, settings::Dict)
return context_tokens > context_window - get(settings, :reserve_tokens, 16384)
end
function find_cut_point(entries::Vector{SessionEntry}, keep_recent_tokens::Int)
accumulated = 0
for i = length(entries):-1:1
entry = entries[i]
# Skip invalid cut points
if entry.type == "message" && entry.message.message isa ToolResultMessage
continue
end
tokens = estimate_context_tokens([entry.message])
accumulated += tokens
if accumulated >= keep_recent_tokens
return i + 1
end
end
return 1
end
function prepare_compaction(branch_entries::Vector{SessionEntry}, settings::Dict)
# Find previous compaction
previous_compaction = nothing
for entry in branch_entries
if entry.type == "compaction"
previous_compaction = entry
end
end
# Estimate tokens
context_tokens = estimate_context_tokens([e.message for e in branch_entries])
# Find cut point
cut_point = find_cut_point(branch_entries, get(settings, :keep_recent_tokens, 20000))
# Split into groups
messages_to_summarize = branch_entries[1:cut_point]
retained_tail = branch_entries[cut_point:end]
# Extract file operations
file_ops = extract_file_operations(messages_to_summarize)
return Dict(
:previous_compaction => previous_compaction,
:context_tokens => context_tokens,
:cut_point => cut_point,
:messages_to_summarize => messages_to_summarize,
:retained_tail => retained_tail,
:file_ops => file_ops
)
end
function generate_summary(messages::Vector{AgentMessage}, previous_summary::Union{String, Nothing} = nothing)
conversation = serialize_conversation(messages)
if previous_summary !== nothing
# Update prompt
prompt = """
<previous_summary>
$previous_summary
</previous_summary>
<new_history>
$conversation
</new_history>
Update the previous summary with new progress.
"""
else
# Fresh prompt
prompt = """
<conversation>
$conversation
</conversation>
Generate a summary:
## Goal
## Constraints & Preferences
## Progress
### Done
### In Progress
### Blocked
## Key Decisions
## Next Steps
## Critical Context
## Files read: [...]
## Files modified: [...]
"""
end
# Call LLM
response = models.complete_simple(
model,
Context(
system_prompt = "You are a helpful assistant that summarizes conversations.",
messages = [UserMessage([TextContent(prompt)], Dates.datetime2timestamp(Dates.now()))],
tools = nothing
)
)
return response.choices[1].message.content
end
function compact(preparation::Dict, model::Model, models::Models)
messages_to_summarize = preparation[:messages_to_summarize]
previous_compaction = preparation[:previous_compaction]
if previous_compaction !== nothing
previous_summary = previous_compaction.summary
summary = generate_summary(messages_to_summarize, previous_summary)
else
summary = generate_summary(messages_to_summarize, nothing)
end
# Extract file operations
file_ops = preparation[:file_ops]
summary *= "\n\nFiles read: $(file_ops.read_files)"
summary *= "\nFiles modified: $(file_ops.modified_files)"
return Dict(
:summary => summary,
:first_kept_entry_id => messages_to_summarize[end].id,
:tokens_before => preparation[:context_tokens],
:retained_tail => preparation[:retained_tail]
)
end
Step 8: Event System (events.jl)
# events.jl
struct AgentEvent
type::String
# Dynamic fields based on type
end
struct AgentStart <: AgentEvent
type::String = "agent_start"
end
struct AgentEnd <: AgentEvent
type::String = "agent_end"
messages::Vector{AgentMessage}
end
struct TurnStart <: AgentEvent
type::String = "turn_start"
end
struct TurnEnd <: AgentEvent
type::String = "turn_end"
message::AgentMessage
tool_results::Vector{ToolResultMessage}
end
struct MessageStart <: AgentEvent
type::String = "message_start"
message::AgentMessage
end
struct MessageUpdate <: AgentEvent
type::String = "message_update"
message::AgentMessage
assistant_message_event::Any
end
struct MessageEnd <: AgentEvent
type::String = "message_end"
message::AgentMessage
end
struct ToolExecutionStart <: AgentEvent
type::String = "tool_execution_start"
tool_call_id::String
tool_name::String
args::Any
end
struct ToolExecutionUpdate <: AgentEvent
type::String = "tool_execution_update"
tool_call_id::String
tool_name::String
args::Any
partial_result::Any
end
struct ToolExecutionEnd <: AgentEvent
type::String = "tool_execution_end"
tool_call_id::String
tool_name::String
result::Any
is_error::Bool
end
function create_tool_result_message(finalized)
ToolResultMessage(
finalized.tool_call.id,
finalized.tool_call.name,
finalized.result.content,
finalized.result.details,
finalized.result.usage,
finalized.is_error,
Dates.datetime2timestamp(Dates.now())
)
end
Step 9: Hook System (hooks.jl)
# hooks.jl
struct BeforeToolCallContext
assistant_message::AssistantMessage
tool_call::ToolCall
args::Any
context::AgentContext
end
struct BeforeToolCallResult
block::Bool
reason::Union{String, Nothing}
end
struct AfterToolCallContext
assistant_message::AssistantMessage
tool_call::ToolCall
args::Any
result::ToolExecutionResult
is_error::Bool
context::AgentContext
end
struct AfterToolCallResult
content::Union{Vector{Union{TextContent, ImageContent}}, Nothing}
details::Union{Any, Nothing}
is_error::Union{Bool, Nothing}
usage::Union{Usage, Nothing}
terminate::Union{Bool, Nothing}
end
struct PrepareNextTurnContext
message::AssistantMessage
tool_results::Vector{ToolResultMessage}
context::AgentContext
new_messages::Vector{AgentMessage}
end
struct AgentLoopTurnUpdate
context::Union{AgentContext, Nothing}
model::Union{Model, Nothing}
thinking_level::Union{ThinkingLevel, Nothing}
end
function default_before_tool_call(context::BeforeToolCallContext, signal::Union{AbortSignal, Nothing})
return nothing
end
function default_after_tool_call(context::AfterToolCallContext, signal::Union{AbortSignal, Nothing})
return AfterToolCallResult(nothing, nothing, nothing, nothing, nothing)
end
function default_prepare_next_turn(context::PrepareNextTurnContext)
return nothing
end
Step 10: Stream Utilities (stream.jl)
# stream.jl
struct StreamEvent
type::String
# Dynamic fields based on type
end
struct StartEvent <: StreamEvent
type::String = "start"
partial::AssistantMessage
end
struct TextStartEvent <: StreamEvent
type::String = "text_start"
end
struct TextDeltaEvent <: StreamEvent
type::String = "text_delta"
delta::String
partial::AssistantMessage
end
struct TextEndEvent <: StreamEvent
type::String = "text_end"
end
struct ThinkingStartEvent <: StreamEvent
type::String = "thinking_start"
end
struct ThinkingDeltaEvent <: StreamEvent
type::String = "thinking_delta"
delta::String
partial::AssistantMessage
end
struct ThinkingEndEvent <: StreamEvent
type::String = "thinking_end"
end
struct ToolcallStartEvent <: StreamEvent
type::String = "toolcall_start"
end
struct ToolcallDeltaEvent <: StreamEvent
type::String = "toolcall_delta"
delta::String
partial::AssistantMessage
end
struct ToolcallEndEvent <: StreamEvent
type::String = "toolcall_end"
end
struct DoneEvent <: StreamEvent
type::String = "done"
end
struct ErrorEvent <: StreamEvent
type::String = "error"
error::String
end
struct Stream
events::Channel{StreamEvent}
end
function Stream()
return Stream(Channel{StreamEvent}(32))
end
function stream_simple(model::Model, context::Context, options::Dict)
stream = Stream()
Threads.@spawn begin
# Call LLM provider
response = make_llm_call(model, context, options)
# Stream events
for chunk in response
if chunk.delta !== nothing
push!(stream.events, TextDeltaEvent(chunk.delta, chunk.message))
end
if chunk.tool_calls !== nothing
for tool_call in chunk.tool_calls
push!(stream.events, ToolcallDeltaEvent(JSON.json(tool_call), chunk.message))
end
end
end
push!(stream.events, DoneEvent())
close(stream.events)
end
return stream
end
function next_event(stream::Stream)
return take!(stream.events)
end
function isdone(stream::Stream)
return isclosed(stream.events)
end
Usage Example
# main.jl
using PiAgent
# Create models
models = create_models()
models.set_provider(anthropic_provider())
model = models.get_model("anthropic", "claude-sonnet-4-6")
# Create agent
agent = Agent(
initial_state = AgentState(
system_prompt = "You are a helpful assistant.",
model = model,
thinking_level = THINKING_OFF,
tools = [],
messages = []
),
stream_fn = models.stream_simple,
before_tool_call = default_before_tool_call,
after_tool_call = default_after_tool_call
)
# Subscribe to events
unsubscribe = agent.subscribe() do event, signal
if event.type == "message_update" && event.assistant_message_event.type == "text_delta"
print(event.assistant_message_event.delta)
end
end
# Run agent
prompt(agent, "Hello!")
# Clean up
unsubscribe()
Data Flow Diagrams
Complete Prompt Flow
User: "What product do you have in stock?"
1. normalize_prompt_input()
↓
[{ role: "user", content: [{ type: "text", text: "..." }], timestamp: ... }]
2. run_prompt_messages()
↓
create_context_snapshot()
create_loop_config()
3. run_agent_loop()
↓
emit(agent_start)
emit(turn_start)
emit(message_start, message_end) for user prompt
4. run_loop()
↓
stream_assistant_response()
↓
transform_context() (optional)
↓
convert_to_llm()
↓
Build LLM context
↓
stream_function()
↓
LLM API call
5. Stream events from LLM
↓
start → text_delta* → done
6. Commit message to context
↓
emit(message_start, message_update*, message_end)
7. Extract tool calls
↓
execute_tool_calls()
↓
prepare_tool_call() for each
↓
execute_prepared_tool_call()
↓
finalize_executed_tool_call()
↓
create_tool_result_message()
↓
emit(tool_execution_start, tool_execution_end)
emit(message_start, message_end) for tool result
8. emit(turn_end)
↓
prepareNextTurn hook
↓
shouldStopAfterTurn hook
9. Check steering/follow-up queues
↓
Continue loop or exit
10. emit(agent_end)
Tool Execution Flow
Assistant Message (with toolCall blocks)
↓
extract tool calls
↓
determine execution mode
↓
for each tool_call:
↓
prepare_tool_call()
├─ find_tool(tool_call.name)
├─ prepare_tool_call_arguments()
├─ validate_tool_arguments()
├─ before_tool_call hook
│ └─ block? → error result
└─ return PreparedToolCall
↓
execute_prepared_tool_call()
├─ tool.execute()
├─ stream partial results via onUpdate()
└─ return ExecutedToolCallOutcome
↓
finalize_executed_tool_call()
├─ after_tool_call hook
└─ return FinalizedToolCallOutcome
↓
create_tool_result_message()
↓
emit(tool_execution_start/end)
emit(message_start/end)
↓
add to context.messages
Session Persistence Flow
AgentEvent
↓
handle_agent_event()
↓
pendingSessionWrites.push()
↓
flush_pending_session_writes()
↓
for each pending write:
├─ message → session.append_message()
├─ compaction → session.append_compaction()
├─ branch_summary → session.append_branch_summary()
├─ leaf → session.set_leaf_id()
└─ custom → session.append_custom_entry()
↓
JSONL file update
Key Algorithms
1. Context Token Estimation
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ CONTEXT TOKEN ESTIMATION ALGORITHM │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ estimateContextTokens(messages) → Int │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ total = 0 │
│ for each message in messages: │
│ total += estimateTokens(message) │
│ return total │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ estimateTokens(message) → Int │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ role === "user" │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ content.length / 4 (4 chars ≈ 1 token) │ │
│ │ (images ≈ 4800 chars each) │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ role === "assistant" │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ sum of all content blocks: │ │
│ │ • text blocks → text.length │ │
│ │ • thinking blocks → thinking.length │ │
│ │ • toolCall blocks → name.length + JSON.stringify(args).length │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ role === "toolResult" / "custom" │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ content.length / 4 │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ role === "bashExecution" │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ (command.length + output.length) / 4 │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ role === "compactionSummary" / "branchSummary" │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ summary.length / 4 │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
function estimateContextTokens(messages) {
let total = 0
for (const message of messages) {
total += estimateTokens(message)
}
return total
}
function estimateTokens(message) {
switch (message.role) {
case "user":
return message.content.reduce((sum, c) => sum + c.text.length, 0) / 4
case "assistant":
return message.content.reduce((sum, c) => {
if (c.type === "text") return sum + c.text.length
if (c.type === "thinking") return sum + c.thinking.length
if (c.type === "toolCall") return sum + c.name.length + JSON.stringify(c.arguments).length
return sum
}, 0)
case "toolResult":
case "custom":
return message.content.reduce((sum, c) => sum + c.text.length, 0) / 4
case "bashExecution":
return (message.command.length + message.output.length) / 4
case "compactionSummary":
case "branchSummary":
return message.summary.length / 4
}
}
2. Turn Start Index Detection
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ TURN START INDEX DETECTION ALGORITHM │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ findTurnStartIndex(entries, startIndex, endIndex) → Int │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Walk backward from endIndex │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ for i from endIndex-1 down to startIndex: │ │
│ │ │ │
│ │ entry = entries[i] │ │
│ │ │ │
│ │ if entry.type == "message": │ │
│ │ role = entry.message.role │ │
│ │ if role == "user" || role == "assistant": │ │
│ │ return i // Found turn start │ │
│ │ │ │
│ │ continue │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ ▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Return endIndex (no valid turn start found) │
└─────────────────────────────────────────────────────────────────────────────────────┘
3. File Operations Extraction
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ FILE OPERATIONS EXTRACTION ALGORITHM │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ createFileOps() → FileOps │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ { │ │
│ │ read: new Set(), │ │
│ │ edited: new Set() │ │
│ │ } │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ extractFileOpsFromMessage(message, fileOps) │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ if message.role === "toolResult": │
│ details = message.details │
│ if details: │
│ if details.readFiles: │
│ for f of details.readFiles: │
│ fileOps.read.add(f) │
│ if details.modifiedFiles: │
│ for f of details.modifiedFiles: │
│ fileOps.edited.add(f) │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ computeFileLists(fileOps, messages, entries, prevCompactionIndex) │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ { │
│ readFiles: [...fileOps.read], │
│ modifiedFiles: [...fileOps.edited] │
│ } │
└─────────────────────────────────────────────────────────────────────────────────────┘
4. Branch Context Building
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ BRANCH CONTEXT BUILDING ALGORITHM │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ buildContextEntries(pathEntries, options) → SessionTreeEntry[] │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ entries = defaultContextEntryTransform(pathEntries) │
│ for transform of options.entryTransforms: │
│ entries = transform(entries) │
│ return entries │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ defaultContextEntryTransform(pathEntries) → SessionTreeEntry[] │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ Find latest compaction entry │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ for entry of pathEntries: │ │
│ │ if entry.type == "compaction": │ │
│ │ compaction = entry │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ if !compaction: │
│ return [...pathEntries] │
│ │
│ entries = [compaction] │
│ compactionIdx = pathEntries.findIndex(e => e.id == compaction.id) │
│ │
│ if compaction.retainedTail: │
│ for i from compactionIdx+1 to end: │
│ entries.push(pathEntries[i]) │
│ return entries │
│ │
│ if compaction.firstKeptEntryId: │
│ foundFirstKept = false │
│ for i from 0 to compactionIdx-1: │
│ if pathEntries[i].id == compaction.firstKeptEntryId: │
│ foundFirstKept = true │
│ if foundFirstKept: │
│ entries.push(pathEntries[i]) │
│ │
│ for i from compactionIdx+1 to end: │
│ entries.push(pathEntries[i]) │
│ │
│ return entries │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ sessionEntryToContextMessages(entry, index, entries, options) → AgentMessage[] │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ if entry.type === "message": │
│ return [entry.message] │
│ │
│ if entry.type === "compaction": │
│ return [ │
│ createCompactionSummaryMessage(...), │
│ ...(entry.retainedTail ?? []) │
│ ] │
│ │
│ if entry.type === "branchSummary" && entry.summary: │
│ return [createBranchSummaryMessage(...)] │
│ │
│ if entry.type === "custom": │
│ return [...(options.entryProjectors?.[entry.customType]?.(...) ?? [])] │
│ │
│ return [] │
└─────────────────────────────────────────────────────────────────────────────────────┘
5. JSONL File Format
// Header
{
type: "session",
version: 3,
id: sessionId,
timestamp: createdAt.toISOString(),
cwd: cwd,
parentSessionPath: parentSessionPath,
metadata: metadata
}
// Entry types
{
type: "message",
id: uuid7(),
parentId: previousLeafId,
timestamp: now.toISOString(),
message: {
role: "user" | "assistant" | "toolResult" | "custom",
// ... message fields
}
}
{
type: "compaction",
id: uuid7(),
parentId: previousLeafId,
timestamp: now.toISOString(),
summary: "...",
firstKeptEntryId: entryId,
tokensBefore: tokenCount,
details: {
readFiles: [...],
modifiedFiles: [...]
}
}
{
type: "branch_summary",
id: uuid7(),
parentId: previousLeafId,
timestamp: now.toISOString(),
summary: "...",
fromId: branchStartId
}
{
type: "leaf",
id: uuid7(),
parentId: previousLeafId,
timestamp: now.toISOString(),
targetId: newLeafId
}
6. Tool Call Argument Preparation
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ TOOL CALL ARGUMENT PREPARATION ALGORITHM │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ prepareToolCallArguments(tool, toolCall) → ToolCall │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ if !tool.prepareArguments: │
│ return toolCall │
│ │
│ preparedArguments = tool.prepareArguments(toolCall.arguments) │
│ if preparedArguments === toolCall.arguments: │
│ return toolCall │
│ │
│ return { │
│ ...toolCall, │
│ arguments: preparedArguments │
│ } │
└─────────────────────────────────────────────────────────────────────────────────────┘
7. Tool Batch Termination Check
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ TOOL BATCH TERMINATION CHECK ALGORITHM │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ shouldTerminateToolBatch(finalizedCalls) → Boolean │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ if finalizedCalls.length == 0: │
│ return false │
│ │
│ return finalizedCalls.every(finalized => │
│ finalized.result.terminate === true │
│ ) │
└─────────────────────────────────────────────────────────────────────────────────────┘
8. Message Normalization
function normalizePromptInput(input, images) {
if (Array.isArray(input)) {
return input
}
if (typeof input !== "string") {
return [input]
}
const content = [{ type: "text", text: input }]
if (images && images.length > 0) {
content.push(...images)
}
return [{ role: "user", content, timestamp: Date.now() }]
}
Summary
The Pi Agent Core architecture is a sophisticated stateful agent system with:
- Stateful execution: Maintains conversation history across multiple turns
- Tool execution: Supports LLM tool calling with parallel/sequential modes
- Event streaming: Real-time event system for UI updates
- Session persistence: JSONL-based persistent storage with tree-structured branching
- Memory compaction: Automatic context window management through LLM summarization
- Flexible extension: Hook-based customization at every system boundary
The implementation follows these key principles:
- Separation of concerns: Core agent logic separated from storage and provider implementations
- Streaming first: All operations designed around async streams for responsiveness
- Type safety: Strong types for compile-time guarantees
- Extensibility: Hooks at every major boundary allow customization
- Persistence: Session history survives restarts through JSONL files
The Julia implementation should mirror this architecture, using Julia's type system for strong typing, async/await for streaming, and JSON for persistence.
Complete Architecture Summary
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ COMPLETE SYSTEM ARCHITECTURE SUMMARY │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ USER INPUT │
│ "what product do you have in stock?" │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ AGENT CLASS (agent.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ • prompt() → normalizePromptInput() → runPromptMessages() │ │
│ │ • continue() → runContinuation() │ │
│ │ • reset() → clear state │ │
│ │ • subscribe(listener) → processEvents() │ │
│ │ • abort() → signal.abort() │ │
│ │ │ │
│ │ State: systemPrompt, model, thinkingLevel, tools, messages │ │
│ │ Queues: steeringQueue, followUpQueue │ │
│ │ Hooks: convertToLlm, transformContext, beforeToolCall, │ │
│ │ afterToolCall, prepareNextTurn │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ AGENT LOOP (agent_loop.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ runLoop() │ │
│ │ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ while (true): │ │ │
│ │ │ • Process pending messages (steering/follow-up) │ │ │
│ │ │ • streamAssistantResponse() → LLM API │ │ │
│ │ │ • executeToolCalls() (parallel/sequential) │ │ │
│ │ │ • emit(turn_end) │ │ │
│ │ │ • prepareNextTurn hook │ │ │
│ │ │ • shouldStopAfterTurn hook │ │ │
│ │ │ → Check steering/follow-up queues │ │ │
│ │ └──────────────────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ LLM PROVIDER API CALL │
│ Models.streamSimple(model, llmContext, options) │
│ • transformContext() (optional) │
│ • convertToLlm() │
│ • streamFunction() → AssistantMessageEventStream │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ SESSION PERSISTENCE (session.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Session: storage, metadata, entries, leaf_id │ │
│ │ • append_message() │ │
│ │ • append_compaction() │ │
│ │ • append_branch_summary() │ │
│ │ • set_leaf_id() │ │
│ │ │ │
│ │ JSONL format: │ │
│ │ { type: "message", ... } │ │
│ │ { type: "compaction", summary: "...", ... } │ │
│ │ { type: "branch_summary", summary: "...", ... } │ │
│ │ { type: "leaf", targetId: "..." } │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ MEMORY MANAGEMENT (compaction.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ estimateContextTokens(messages) → Int │ │
│ │ │ │
│ │ shouldCompact(contextTokens, contextWindow, settings) → Boolean │ │
│ │ Trigger: contextTokens > contextWindow - reserveTokens │ │
│ │ │ │
│ │ findCutPoint(entries, keepRecentTokens) → Int │ │
│ │ Walk backward, skip toolResults │ │
│ │ │ │
│ │ prepareCompaction(branchEntries, settings) → Preparation │ │
│ │ generateSummary(messages, previousSummary?) → String │ │
│ │ compact(preparation, model, models) → Summary │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ EVENT SYSTEM (events.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Events: │ │
│ │ • agent_start / agent_end │ │
│ │ • turn_start / turn_end │ │
│ │ • message_start / message_update / message_end │ │
│ │ • tool_execution_start / update / end │ │
│ │ │ │
│ │ subscribe(listener) → processEvents(event) │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ HOOK SYSTEM (hooks.jl) │
│ ┌──────────────────────────────────────────────────────────────────────────────┐ │
│ │ Hooks: │ │
│ │ • convertToLlm (AgentMessage[] → Message[]) │ │
│ │ • transformContext (pruning/injection) │ │
│ │ • beforeToolCall (block: { block: true }) │ │
│ │ • afterToolCall (override: { content, details, isError, ... }) │ │
│ │ • prepareNextTurn (context/model/thinkingLevel update) │ │
│ │ • shouldStopAfterTurn (graceful termination) │ │
│ │ • getSteeringMessages (mid-turn injection) │ │
│ │ • getFollowUpMessages (post-agent execution) │ │
│ └──────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
Quick Reference Guide
Architecture Flow
User Input → Agent Class → Agent Loop → LLM API → Session → Compaction
│ │ │ │ │
└── Events ──┘ └── Hooks ┘ └── Tree
Key Components
| Component | File | Purpose |
|---|---|---|
| Agent | agent.jl |
Stateful wrapper, event emission, queues |
| Agent Loop | agent_loop.jl |
Core execution, LLM calls, tool execution |
| Messages | messages.jl |
Message types, conversion, summaries |
| Tools | tools.jl |
Tool execution, validation, hooks |
| Session | session.jl |
JSONL persistence, tree structure |
| Compaction | compaction.jl |
Token estimation, summarization |
| Events | events.jl |
Event types, subscription |
| Hooks | hooks.jl |
Hook contexts, return types |
| Types | types.jl |
All type definitions |
| Stream | stream.jl |
Stream utilities |
Event Flow (With Tools)
agent_start
↓
turn_start
↓
message_start (user)
↓
message_end (user)
↓
message_start (assistant - streaming)
↓
message_update (text_delta, toolcall_delta)
↓
message_end (assistant with tool calls)
↓
tool_execution_start (read tool)
↓
tool_execution_end (read complete)
↓
message_start (toolResult)
↓
message_end (toolResult)
↓
turn_end
↓
[Inner loop continues]
↓
turn_start
↓
message_start (assistant - streaming)
↓
message_end (final response)
↓
turn_end
↓
agent_end
Tool Execution Modes
Parallel (default):
1. Preflight all tools sequentially
2. Queue async executions
3. Execute concurrent (Promise.all)
4. Emit results in assistant source order
Sequential:
1. Execute one-by-one
2. Wait for each to complete
3. Emit in execution order
Memory Management
contextTokens > contextWindow - reserveTokens → Trigger Compaction
findCutPoint: Walk backward, skip toolResults
generateSummary: LLM creates structured summary
compact: Replace history with summary, keep tail verbatim
Result: ~185K tokens → ~25K tokens (saved ~160K tokens)
Hook Points
| Hook | Purpose |
|---|---|
| convertToLlm | Transform messages for LLM |
| transformContext | Pruning/injection (optional) |
| beforeToolCall | Block execution (can return { block: true }) |
| afterToolCall | Override result (can return { content, details, ... }) |
| prepareNextTurn | Update context/model/thinkingLevel |
| shouldStopAfterTurn | Graceful termination |
| getSteeringMessages | Inject messages mid-turn |
| getFollowUpMessages | Queue messages for post-agent |