第8章 The Core Agent

CodexThread: The Conversation Container

The CodexThread struct represents a single conversation thread – the continuous dialogue between user and agent. It's not just a container for messages; it's the orchestrator that manages thread state, processes turns, and maintains conversation context.

pub struct CodexThread {
    pub(crate) codex: Codex,
    rollout_path: Option<PathBuf>,
    out_of_band_elicitation_count: Mutex<u64>,
    _watch_registration: WatchRegistration,
}

At first glance, CodexThread appears deceptively simple. The real complexity lies in the Codex struct it wraps, which contains the actual session management logic. The thread maintains:

• A bidirectional channel pair (tx_sub for submissions, rx_event for events) implementing the queue-pair pattern
• Rollout persistence via an optional path where conversation history is written as JSONL
• Elicitation counting to track out-of-band approval requests
• File watching capabilities through the _watch_registration field

🤿 深入底层：The queue-pair architecture (async_channel::bounded / async_channel::unbounded) enables backpressure handling. When the submission channel fills up, producers naturally slow down, preventing unbounded memory growth. This pattern appears throughout Codex for inter-component communication.

The ThreadConfigSnapshot captures the configuration state at a point in time:

#[derive(Clone, Debug)]
pub struct ThreadConfigSnapshot {
    pub model: String,
    pub model_provider_id: String,
    pub service_tier: Option<ServiceTier>,
    pub approval_policy: AskForApproval,
    pub approvals_reviewer: ApprovalsReviewer,
    pub sandbox_policy: SandboxPolicy,
    pub cwd: PathBuf,
    pub ephemeral: bool,
    pub reasoning_effort: Option<ReasoningEffort>,
    pub personality: Option<Personality>,
    pub session_source: SessionSource,
}

This snapshot is critical because config can change mid-session (via SetConfig submissions), but each turn needs consistent configuration throughout its execution.

The threading model in Codex uses async tasks extensively. Each session spawns a background task that polls the submission channel:

// The session loop runs in the background
let session_loop = tokio::spawn(async move {
    session_loop(
        rx_sub,
        tx_event,
        session,
        agent_status_tx,
        termination_tx,
    )
    .await
});

This architecture allows the main thread to remain responsive while turn processing happens asynchronously. The agent_status watch channel provides real-time status updates to anyone listening.

Turn Processing from Start to Finish

When you type a message in the TUI, it embarks on a journey through multiple layers before becoming a response. Let's trace this path:

The turn begins with steer_input, which validates and queues user input:

pub async fn steer_input(
    &self,
    input: Vec<UserInput>,
    expected_turn_id: Option<&str>,
) -> Result<String, SteerInputError> {
    if input.is_empty() {
        return Err(SteerInputError::EmptyInput);
    }
    
    let mut active = self.active_turn.lock().await;
    let Some(active_turn) = active.as_mut() else {
        return Err(SteerInputError::NoActiveTurn(input));
    };
    
    // Validate turn_id if provided, then queue input
    // ...
}

The SteerInputError enum represents the failure modes:

#[derive(Debug, PartialEq)]
pub enum SteerInputError {
    NoActiveTurn(Vec<UserInput>),
    ExpectedTurnMismatch { expected: String, actual: String },
    ActiveTurnNotSteerable { turn_kind: NonSteerableTurnKind },
    EmptyInput,
}

Once input is validated, context building begins. This includes:

• Loading skills based on cwd and mentions
• Building tool specifications
• Collecting MCP server capabilities
• Gathering connector status

The TurnContext struct captures everything needed for this turn:

pub struct TurnContext {
    pub sub_id: String,              // Unique turn identifier
    pub cwd: PathBuf,                // Current working directory
    pub model: String,               // Model being used
    pub model_info: Arc<ModelInfo>,  // Model capabilities
    pub sandbox_policy: SandboxPolicy,
    pub file_system_sandbox_policy: FileSystemSandboxPolicy,
    pub network_sandbox_policy: NetworkSandboxPolicy,
    pub config: Arc<Config>,
    pub features: ManagedFeatures,
    // ... additional fields
}

⚠️ 避坑指南：Turn contexts are cloned extensively throughout the codebase. If you're adding fields to TurnContext, consider whether they need to be Arc-wrapped to avoid expensive copies. The config and model_info fields are already Arc-wrapped for this reason.

The request context sent to the API consists of:

1. Base instructions (system prompt)
2. Skill injections (context-aware capabilities)
3. Conversation history (compacted if too long)
4. Current user input

Event Handling and Streaming

The Responses API streams results via Server-Sent Events (SSE). Each event is parsed and dispatched:

pub enum ResponseEvent {
    OutputItemAdded(ResponseItem),
    OutputItemDone(ResponseItem),
    TokenCount(TokenUsage),
    Error(StreamErrorEvent),
    RateLimitSnapshot(RateLimitSnapshot),
    // ... more variants
}

Event processing happens in a loop:

while let Some(event) = response_stream.next().await {
    match event {
        ResponseEvent::OutputItemAdded(item) => {
            // Streaming content arrival
        }
        ResponseEvent::OutputItemDone(item) => {
            // Item complete - may trigger tool execution
            let output_result = handle_output_item_done(
                &mut ctx, item, previously_active_item
            ).await?;
            
            if let Some(tool_future) = output_result.tool_future {
                in_flight.push_back(tool_future);
            }
        }
        ResponseEvent::Error(err) => {
            // Handle stream errors
        }
        // ... other events
    }
}

The async stream processing uses futures::Stream for composability:

use futures::stream::StreamExt;

let response_stream = client.stream_responses(request).await?;
pin_mut!(response_stream);

while let Some(event) = response_stream.next().await {
    // Process event
}

This approach allows for:

• Cancellation via CancellationToken
• Timeout handling with tokio::time::timeout
• Backpressure through async channel bounds

The handle_output_item_done function is particularly important – it's where completed items are recorded and tool execution futures are created:

pub(crate) async fn handle_output_item_done(
    sess: &Session,
    turn_context: &TurnContext,
    item: ResponseItem,
    previously_active_item: Option<ResponseItem>,
) -> Result<OutputItemResult, CodexErr> {
    // Record the item in conversation history and rollout
    record_completed_response_item(sess, turn_context, &item).await;
    
    // Extract tool calls if present
    let tool_future = if let ResponseItem::FunctionCall { .. } = item {
        Some(execute_tool_call(sess, turn_context, item))
    } else {
        None
    };
    
    Ok(OutputItemResult {
        tool_future,
        last_agent_message: extract_message(&item),
        needs_follow_up: requires_follow_up(&item),
    })
}

☕ 茶水间时间：The "OutputItemAdded" vs "OutputItemDone" distinction mirrors the incremental vs complete nature of streaming. "Added" events fire as content arrives (great for UI updates), while "Done" events fire when an item is complete (needed for tool execution and history recording). This dual-event pattern is common in streaming systems.

Tool Orchestration

The ToolOrchestrator in codex-rs/core/src/tools/orchestrator.rs is the central coordinator for all tool execution. It handles:

1. Approval requests – User or automated reviewer approval
2. Sandbox selection – Choosing the right sandbox level
3. Retry logic – Escalating sandbox on denial
4. Network approval – Managed network policy checks

The orchestrator's run method implements a clear sequence:

pub async fn run<Rq, Out, T>(
    &mut self,
    tool: &mut T,
    req: &Rq,
    tool_ctx: &ToolCtx,
    turn_ctx: &TurnContext,
    approval_policy: AskForApproval,
) -> Result<OrchestratorRunResult<Out>, ToolError>
where
    T: ToolRuntime<Rq, Out>,
{
    // 1) Approval phase
    let requirement = tool.exec_approval_requirement(req)?;
    match requirement {
        ExecApprovalRequirement::Skip => { /* proceed */ }
        ExecApprovalRequirement::Forbidden { reason } => {
            return Err(ToolError::Rejected(reason));
        }
        ExecApprovalRequirement::NeedsApproval { reason } => {
            let decision = tool.start_approval_async(req, approval_ctx).await;
            match decision {
                ReviewDecision::Denied => return Err(ToolError::Rejected(reason)),
                ReviewDecision::Approved => { /* proceed */ }
                // ... other variants
            }
        }
    }
    
    // 2) First attempt with initial sandbox
    let initial_sandbox = self.sandbox.select_initial(
        &turn_ctx.file_system_sandbox_policy,
        turn_ctx.network_sandbox_policy,
        tool.sandbox_preference(),
        // ...
    );
    
    let (first_result, _) = Self::run_attempt(
        tool, req, tool_ctx, &initial_attempt,
    ).await;
    
    // 3) Handle denial with escalation
    match first_result {
        Ok(out) => Ok(/* success */),
        Err(ToolError::Codex(CodexErr::Sandbox(SandboxErr::Denied { .. }))) => {
            if !tool.escalate_on_failure() {
                return Err(/* cannot retry */);
            }
            // Request approval for retry, then run without sandbox
            let escalated_attempt = SandboxAttempt {
                sandbox: SandboxType::None,
                // ...
            };
            Self::run_attempt(tool, req, tool_ctx, &escalated_attempt).await
        }
    }
}

The ToolRuntime trait defines the interface all tools must implement:

pub trait ToolRuntime<Rq, Out> {
    fn exec_approval_requirement(&self, req: &Rq) -> Option<ExecApprovalRequirement>;
    
    async fn start_approval_async(
        &mut self,
        req: &Rq,
        ctx: ApprovalCtx<'_>,
    ) -> ReviewDecision;
    
    fn sandbox_mode_for_first_attempt(&self, req: &Rq) -> SandboxOverride;
    
    fn sandbox_preference(&self) -> SandboxPreference;
    
    async fn run(
        &mut self,
        req: &Rq,
        attempt: &SandboxAttempt<'_>,
        ctx: &ToolCtx,
    ) -> Result<Out, ToolError>;
    
    // ... additional methods
}

⚠️ 避坑指南：The approval caching mechanism (via already_approved flag) prevents re-approval on sandbox retry. If you modify the retry logic, ensure you don't accidentally bypass this safety check – users should only approve once per tool invocation.

Parallel execution of independent tools is handled by the ToolCallRuntime:

pub struct ToolCallRuntime {
    router: ToolRouter,
    orchestrator: ToolOrchestrator,
    approval_store: ApprovalStore,
    // ...
}

// Multiple tools can execute concurrently
let futures: Vec<_> = tool_calls
    .into_iter()
    .map(|call| self.execute_single_tool(call, ctx))
    .collect();

let results = futures::future::join_all(futures).await;

The orchestrator tracks network approvals separately, using a deferred approval pattern for tools that need network access:

match network_approval.mode() {
    NetworkApprovalMode::Immediate => {
        finish_immediate_network_approval(&session, network_approval).await;
    }
    NetworkApprovalMode::Deferred => {
        // Stash for completion after tool succeeds
        deferred_network_approval = network_approval.into_deferred();
    }
}

Skill Injection and Context Building

Skills transform the system prompt from static instructions into context-aware capabilities. The injection process happens in several stages:

pub(crate) async fn build_skill_injections(
    mentioned_skills: &[SkillMetadata],
    session_telemetry: Option<&SessionTelemetry>,
    analytics_client: &AnalyticsEventsClient,
    tracking: AnalyticsTracking,
) -> SkillInjections {
    // 1) Resolve skill dependencies
    let resolved = resolve_skill_dependencies_for_turn(
        mentioned_skills,
        &skills_manager,
        &config,
    ).await;
    
    // 2) Load skill content from disk
    let loaded = load_skill_contents(&resolved.skills).await?;
    
    // 3) Build skill injection items
    let items = loaded
        .iter()
        .map(|skill| build_skill_item(skill, &cwd))
        .collect();
    
    // 4) Generate warnings for deprecated/missing skills
    let warnings = generate_skill_warnings(&resolved);
    
    SkillInjections { items, warnings }
}

Skills are discovered through multiple mechanisms:

• Explicit mentions – User types /skill-name or mentions skill in conversation
• CWD-based – Skills in .claude/skills/ relative to working directory
• System skills – Built-in skills embedded in the binary

The final prompt assembly combines all context:

pub(crate) fn build_prompt(
    input: Vec<ResponseItem>,
    router: &ToolRouter,
    turn_context: &TurnContext,
    base_instructions: BaseInstructions,
) -> Prompt {
    // Filter deferred tools (tools that load on first use)
    let deferred_tools: HashSet<_> = turn_context
        .dynamic_tools
        .iter()
        .filter(|tool| tool.defer_loading)
        .map(|tool| tool.name.as_str())
        .collect();
    
    let tools = if deferred_tools.is_empty() {
        router.model_visible_specs()
    } else {
        router
            .model_visible_specs()
            .into_iter()
            .filter(|spec| !deferred_tools.contains(spec.name()))
            .collect()
    };
    
    Prompt {
        input,                                    // Conversation + skills
        tools,                                    // Available tools
        parallel_tool_calls: turn_context.model_info.supports_parallel_tool_calls,
        base_instructions,                        // System prompt
        personality: turn_context.personality,     // Optional personality
        output_schema: turn_context.final_output_json_schema.clone(),
    }
}

The context building pipeline for a typical turn:

1. Parse user input → Extract explicit skill mentions
2. Scan cwd → Find project-level skills
3. Load system skills → Built-in capabilities
4. Resolve dependencies → Handle skill dependencies
5. Load skill content → Read markdown files
6. Build injection items → Create prompt segments
7. Assemble final prompt → Combine with conversation history

🤿 深入底层：Skill loading uses include_dir! to embed system skills at compile time. This means the binary contains all default skills as static data, eliminating file I/O for system skills and ensuring they're always available. Project skills are loaded from disk and watched for changes via SkillsWatcher.

---

Chapter Exercises

1. Add logging to trace event processing by modifying handle_output_item_done to emit trace-level logs for each event type processed.

2. Modify tool invocation to add a pre-hook by creating a trait method pre_execute on ToolRuntime and calling it in ToolOrchestrator::run_attempt.

3. Create a custom skill by adding a markdown file to .claude/skills/ and verify it's injected correctly by checking the skill warnings event stream.

---

The core agent processes turns through an elegant pipeline: input → validation → context building → API call → event streaming → tool execution → response. Each component has clear responsibilities, and the event-driven architecture enables real-time UI updates and cancellation. But users need to see what's happening – which brings us to the TUI implementation.