YiemAgent/learn_pi_agent/architecture.md

Agent Architecture: Pi Agent Core - A Comprehensive Guide for Julia Implementation

Table of Contents

1. Overview
2. Architecture Layers
3. Core Components
4. Message System
5. Agent Loop
6. Tool Execution
7. Session Management
8. Memory & Context Management
9. Event System
10. Hook System
11. Implementation Guide for Julia
12. Data Flow Diagrams
13. Key Algorithms

---

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

1. Separation of concerns: Core agent logic is separated from storage and provider implementations
2. Streaming first: All operations are designed around async streams for responsiveness
3. Type safety: Strong TypeScript types for compile-time guarantees
4. Extensibility: Hooks at every major boundary allow customization
5. Persistence: Session history survives restarts through JSONL files

---

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

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:

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 array
• continue(): Continue from current context
• steer(message): Queue steering message (interruption during tool execution)
• followUp(message): Queue follow-up message (executes after agent stops)
• reset(): Clear all state
• subscribe(listener): Register event listener
• abort(): Cancel current operation
• waitForIdle(): Wait for completion and event listeners

2. Agent Loop (agent-loop.ts)

Purpose: Core execution logic without state management

Main entry points:

• runAgentLoop(prompts, context, config, emit, signal, streamFn): Start new loop with prompts
• runAgentLoopContinue(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 input
• assistant: 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 summary
• compactionSummary: History compaction result

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:

1. Assistant partial message: Added when start event arrives
2. Assistant final message: Replaces partial on done/error
3. 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

1. Preparation (prepareToolCall)

   • Look up tool by name
   • Call prepareArguments if defined
   • Validate with validateToolArguments
   • Call beforeToolCall hook (can block with { block: true })
   • Return PreparedToolCall or ImmediateToolCallOutcome

2. Execution (executePreparedToolCall)

   • Call tool.execute(toolCallId, args, signal, onUpdate)
   • Stream partial results via onUpdate
   • Handle errors, return ExecutedToolCallOutcome

3. Finalization (finalizeExecutedToolCall)

   • Call afterToolCall hook (can override result)
   • Return FinalizedToolCallOutcome

4. Message Creation (createToolResultMessage)

   • Create ToolResultMessage
   • Set toolCallId, toolName, content, details, usage, isError, timestamp

---

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"}

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

1. User navigates to a different point in history
2. Leaf moves to the selected entry
3. Branch summary generated for diverged work
4. 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

function estimateTokens(message: AgentMessage): number {
  switch (message.role) {
    case "user":
      return message.content.length / 4  // 4 chars ≈ 1 token
    
    case "assistant":
      return message.content.reduce((sum, block) => {
        if (block.type === "text") return sum + block.text.length
        if (block.type === "thinking") return sum + block.thinking.length
        if (block.type === "toolCall") return sum + block.name.length + JSON.stringify(block.arguments).length
        return sum
      }, 0)
    
    case "toolResult" | "custom":
      return message.content.length / 4
    
    case "bashExecution":
      return (message.command.length + message.output.length) / 4
    
    case "compactionSummary" | "branchSummary":
      return message.summary.length / 4
  }
}

Compaction Strategy

Trigger condition:

function shouldCompact(contextTokens, contextWindow, settings) {
  return contextTokens > contextWindow - settings.reserveTokens
}

// Defaults
const DEFAULT_COMPACTION_SETTINGS = {
  enabled: true,
  reserveTokens: 16384,   // ~16K for summary prompt + output
  keepRecentTokens: 20000  // ~20K tokens of recent history
}

Cut point finding:

function findCutPoint(entries, startIndex, endIndex, keepRecentTokens) {
  let accumulated = 0
  
  // Walk backward from endIndex
  for (let i = endIndex - 1; i >= startIndex; i--) {
    const entry = entries[i]
    
    // Skip invalid cut points (toolResult stays with its call)
    if (entry.type === "message" && entry.message.role === "toolResult") {
      continue
    }
    
    const tokens = estimateTokens(entry)
    accumulated += tokens
    
    if (accumulated >= keepRecentTokens) {
      return i + 1  // Snap to nearest valid cut point
    }
  }
  
  return startIndex
}

Compaction preparation:

function prepareCompaction(branchEntries, settings) {
  // Find previous compaction
  let previousCompaction = null
  for (const entry of branchEntries) {
    if (entry.type === "compaction") {
      previousCompaction = entry
    }
  }
  
  // Estimate tokens
  const contextTokens = estimateContextTokens(branchEntries)
  
  // Find cut point
  const firstKeptEntryId = findCutPoint(branchEntries, 0, branchEntries.length, settings.keepRecentTokens)
  
  // Split into groups
  const messagesToSummarize = branchEntries.slice(0, firstKeptEntryId)
  const retainedTail = branchEntries.slice(firstKeptEntryId)
  
  // Extract file operations
  const fileOps = extractFileOperations(messagesToSummarize, branchEntries, prevIndex)
  
  return {
    previousCompaction,
    contextTokens,
    firstKeptEntryId,
    messagesToSummarize,
    retainedTail,
    turnPrefixMessages: extractTurnPrefix(messagesToSummarize),
    fileOps
  }
}

Summary generation:

async function generateSummary(messages, previousSummary) {
  if (previousSummary) {
    // UPDATE_SUMMARIZATION_PROMPT (iterative update)
    const prompt = `
      <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
    `
    
    return await models.completeSimple(model, { systemPrompt, messages: [{ role: "user", content: prompt }] })
  } else {
    // FRESH_SUMMARIZATION_PROMPT
    const prompt = `
      <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: [...]
    `
    
    return await models.completeSimple(model, { systemPrompt, messages: [{ role: "user", content: prompt }] })
  }
}

---

Event System

Event Types

Agent Lifecycle

• agent_start: Agent begins processing
• agent_end: Final event for the run

Turn Lifecycle

• turn_start: New turn begins
• turn_end: Turn completes with assistant message and tool results

Message Lifecycle

• message_start: Any message begins
• message_update: Assistant only. Includes assistantMessageEvent with delta
• message_end: Message completes

Tool Execution Lifecycle

• tool_execution_start: Tool begins
• tool_execution_update: Tool streams progress
• tool_execution_end: Tool completes

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_end listeners 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 Context Types

BeforeToolCallContext

interface BeforeToolCallContext {
  assistantMessage: AssistantMessage
  toolCall: AgentToolCall
  args: unknown  // Validated arguments
  context: AgentContext
}

AfterToolCallContext

interface AfterToolCallContext {
  assistantMessage: AssistantMessage
  toolCall: AgentToolCall
  args: unknown
  result: AgentToolResult<any>
  isError: boolean
  context: AgentContext
}

PrepareNextTurnContext

interface PrepareNextTurnContext extends ShouldStopAfterTurnContext {}

interface ShouldStopAfterTurnContext {
  message: AssistantMessage
  toolResults: ToolResultMessage[]
  context: AgentContext
  newMessages: AgentMessage[]
}

Hook Return Types

BeforeToolCallResult

interface BeforeToolCallResult {
  block?: boolean  // Prevent execution
  reason?: string  // Error message if blocked
}

AfterToolCallResult

interface AfterToolCallResult {
  content?: (TextContent | ImageContent)[]
  details?: unknown
  isError?: boolean
  usage?: Usage
  terminate?: boolean  // Hint to stop after batch
}

AgentLoopTurnUpdate

interface AgentLoopTurnUpdate {
  context?: AgentContext
  model?: Model<any>
  thinkingLevel?: ThinkingLevel
}

---

Implementation Guide for Julia

Architecture Overview

Julia Agent Implementation
├── agent.jl                  # Core Agent class
├── agent_loop.jl            # Low-level execution loop
├── messages.jl              # Message types and conversion
├── tools.jl                 # Tool execution
├── session.jl               # Session persistence
├── compaction.jl            # Memory management
├── events.jl                # Event system
├── hooks.jl                 # Hook system
├── types.jl                 # Type definitions
└── stream.jl                # Stream utilities

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

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

function findTurnStartIndex(entries, startIndex, endIndex) {
  // Find first message where role is user or assistant
  for (let i = endIndex - 1; i >= startIndex; i--) {
    const entry = entries[i]
    if (entry.type === "message") {
      const role = entry.message.role
      if (role === "user" || role === "assistant") {
        return i
      }
    }
  }
  return endIndex
}

3. File Operations Extraction

function extractFileOpsFromMessage(message, fileOps) {
  if (message.role === "toolResult") {
    const details = message.details
    if (details) {
      if (details.readFiles) {
        for (const f of details.readFiles) fileOps.read.add(f)
      }
      if (details.modifiedFiles) {
        for (const f of details.modifiedFiles) fileOps.edited.add(f)
      }
    }
  }
}

function createFileOps() {
  return {
    read: new Set(),
    edited: new Set()
  }
}

function computeFileLists(fileOps, messages, entries, prevCompactionIndex) {
  return {
    readFiles: [...fileOps.read],
    modifiedFiles: [...fileOps.edited]
  }
}

4. Branch Context Building

function buildContextEntries(pathEntries, options) {
  let entries = defaultContextEntryTransform(pathEntries)
  for (const transform of options.entryTransforms ?? []) {
    entries = transform(entries)
  }
  return entries
}

function defaultContextEntryTransform(pathEntries) {
  let compaction = null
  for (const entry of pathEntries) {
    if (entry.type === "compaction") {
      compaction = entry
    }
  }
  
  if (!compaction) {
    return [...pathEntries]
  }
  
  const entries = [compaction]
  const compactionIdx = pathEntries.findIndex(e => e.id === compaction.id)
  
  if (compaction.retainedTail) {
    for (let i = compactionIdx + 1; i < pathEntries.length; i++) {
      entries.push(pathEntries[i])
    }
    return entries
  }
  
  if (compaction.firstKeptEntryId) {
    let foundFirstKept = false
    for (let i = 0; i < compactionIdx; i++) {
      const entry = pathEntries[i]
      if (entry.id === compaction.firstKeptEntryId) foundFirstKept = true
      if (foundFirstKept) entries.push(entry)
    }
  }
  
  for (let i = compactionIdx + 1; i < pathEntries.length; i++) {
    entries.push(pathEntries[i])
  }
  
  return entries
}

function sessionEntryToContextMessages(entry, index, entries, options) {
  if (entry.type === "message") {
    return [entry.message]
  }
  if (entry.type === "compaction") {
    return [
      createCompactionSummaryMessage(entry.summary, entry.tokensBefore, entry.timestamp),
      ...(entry.retainedTail ?? [])
    ]
  }
  if (entry.type === "branchSummary" && entry.summary) {
    return [createBranchSummaryMessage(entry.summary, entry.fromId, entry.timestamp)]
  }
  if (entry.type === "custom") {
    return [...(options.entryProjectors?.[entry.customType]?.(entry, index, entries) ?? [])]
  }
  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

function prepareToolCallArguments(tool, toolCall) {
  if (!tool.prepareArguments) {
    return toolCall
  }
  
  const preparedArguments = tool.prepareArguments(toolCall.arguments)
  if (preparedArguments === toolCall.arguments) {
    return toolCall
  }
  
  return {
    ...toolCall,
    arguments: preparedArguments
  }
}

7. Tool Batch Termination Check

function shouldTerminateToolBatch(finalizedCalls) {
  return finalizedCalls.length > 0 && 
         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:

1. Stateful execution: Maintains conversation history across multiple turns
2. Tool execution: Supports LLM tool calling with parallel/sequential modes
3. Event streaming: Real-time event system for UI updates
4. Session persistence: JSONL-based persistent storage with tree-structured branching
5. Memory compaction: Automatic context window management through LLM summarization
6. 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.