What is the execution logic of the agent in Codex? This is the result extracted by me using AIL (AI-Language).

[pyted]

Codex Agent Execution Logic

Analyzed from: codex-main (OpenAI Codex CLI - Rust implementation)

Key source files:

codex-rs/core/src/tasks/regular.rs - main turn loop

codex-rs/core/src/codex.rs - core run_turn / run_sampling_request

codex-rs/core/src/tools/orchestrator.rs - tool approval + sandbox + retry

codex-rs/core/src/memories/phase1.rs - memory extraction (RAG phase 1)

codex-rs/core/src/memories/phase2.rs - memory consolidation (RAG phase 2)

─────────────────────────────────────────────────────────────

Data Types

─────────────────────────────────────────────────────────────

type TaskInput:
user_query: str
attachments: list[str] = []
context_items: list[str] = [] # injected background context

type TurnConfig:
collaboration_mode: str # "auto" | "plan" | "suggest"
reasoning_effort: str # "low" | "medium" | "high"
max_retries: int = 3
sandbox_policy: str # "none" | "container" | "vm"

type ApprovalDecision:
kind: str # "skip" | "approved" | "forbidden" | "needs_approval"
reason: str = ""

type ToolCall:
tool_name: str
arguments: dict
call_id: str

type ToolResult:
call_id: str
output: str
is_error: bool = false
needs_followup: bool = false

type MemoryRecord:
raw_memory: str
rollout_summary: str
rollout_slug: str
usage_count: int = 0

type QualityScore:
relevance: float
completeness: float

─────────────────────────────────────────────────────────────

Tools & Skills

─────────────────────────────────────────────────────────────

use tool shell_exec(cmd: str) -> str
use tool file_read(path: str) -> str
use tool file_write(path: str, content: str) -> str
use tool vector_search(query: str, top_k: int) -> list[MemoryRecord]
use tool rerank(query: str, docs: list[MemoryRecord]) -> list[MemoryRecord]
use tool redact_secrets(text: str) -> str
use tool db_claim_jobs(phase: int) -> list[str] # returns rollout ids
use tool db_load_rollout(rollout_id: str) -> list[str] # returns rollout items
use tool db_persist_memory(rollout_id: str, record: MemoryRecord) -> bool
use tool db_compute_watermark() -> str

use skill spawn_sub_agent(config: dict) -> str # returns agent_id
use skill run_consolidation_agent(memory_root: str) -> str

─────────────────────────────────────────────────────────────

Phase 1 — Memory Extraction (background, on session startup)

Ref: memories/phase1.rs

Runs in parallel across eligible rollouts, extracts structured memory

─────────────────────────────────────────────────────────────

def extract_memories_phase1() -> int: # returns count of processed rollouts
rollout_ids = db_claim_jobs(phase=1)
if len(rollout_ids) == 0:
return 0

processed = 0

# Parallel extraction with concurrency cap (CONCURRENCY_LIMIT in phase1.rs)
results = []
for rollout_id in rollout_ids:                       # buffered concurrency in Rust impl
    result = extract_single_rollout(rollout_id)
    results.append(result)

for r in results:
    if r != "failed":
        processed += 1

return processed

def extract_single_rollout(rollout_id: str) -> str:
items = db_load_rollout(rollout_id)
if len(items) == 0:
return "failed"

prompt p_extract = """
    You are a memory extraction agent.
    Analyze the following conversation rollout and extract:
    1. raw_memory: key facts, decisions, and patterns worth remembering
    2. rollout_summary: a concise summary of what happened
    3. rollout_slug: a short identifier phrase

    Rollout items:
    {items}

    Output valid JSON matching: {{ raw_memory, rollout_summary, rollout_slug }}
"""

try:
    raw_output = @ask(p_extract)
    record = @extract(raw_output, type=MemoryRecord)

    # Redact secrets before persistence (phase1.rs line ~290)
    record.raw_memory      = redact_secrets(record.raw_memory)
    record.rollout_summary = redact_secrets(record.rollout_summary)

    db_persist_memory(rollout_id, record)
    return "ok"
fallback:
    return "failed"

─────────────────────────────────────────────────────────────

Phase 2 — Memory Consolidation (background, serialized global job)

Ref: memories/phase2.rs

Claims a global job, syncs artifacts, spawns a consolidation sub-agent

─────────────────────────────────────────────────────────────

def consolidate_memories_phase2(memory_root: str) -> bool:
# Only one phase2 job runs at a time (global claim in DB)
job_claimed = db_claim_jobs(phase=2)
if len(job_claimed) == 0:
return false

# Load top-N stage-1 outputs ranked by usage_count + last_usage
candidates = vector_search("all memories", top_k=50)
top_records = rerank("recency and relevance", candidates)[:20]

# Sync local artifact files
raw_memories_content = ""
for rec in top_records:
    raw_memories_content = rec.raw_memory + "\n\n---\n\n" + raw_memories_content

file_write(memory_root + "/raw_memories.md", raw_memories_content)

for rec in top_records:
    file_write(
        memory_root + "/rollout_summaries/" + rec.rollout_slug + ".md",
        rec.rollout_summary
    )

# Spawn a restricted internal consolidation agent (phase2.rs line ~131)
# Config: no network, workspace-write-only sandbox, no recursive memory gen
run_consolidation_agent(memory_root)

watermark = db_compute_watermark()
memory.save(watermark, key="phase2:watermark")

return true

─────────────────────────────────────────────────────────────

Tool Orchestrator — Approval + Sandbox + Retry

Ref: tools/orchestrator.rs

Called for every tool the model wants to invoke

─────────────────────────────────────────────────────────────

def orchestrate_tool(call: ToolCall, config: TurnConfig) -> ToolResult:

# ── Step 1: Approval ────────────────────────────────────────
approval = get_approval(call, config)

if approval.kind == "forbidden":
    return ToolResult(
        call_id  = call.call_id,
        output   = "Tool call rejected: " + approval.reason,
        is_error = true
    )

if approval.kind == "needs_approval":
    decision = @ask_user("Allow tool '{call.tool_name}' with args {call.arguments}?")
    if @judge("user rejected the tool call: {decision}"):
        return ToolResult(
            call_id  = call.call_id,
            output   = "User rejected tool: " + decision,
            is_error = true
        )

# ── Step 2: First Sandbox Attempt ───────────────────────────
sandbox = pick_sandbox(config.sandbox_policy)
first_result = run_in_sandbox(call, sandbox)

if not first_result.is_error:
    return first_result

# ── Step 3: Escalation on Sandbox Denial ────────────────────
# orchestrator.rs lines ~228-359
if @judge("first attempt failed due to sandbox denial: {first_result.output}"):
    if approval.kind != "approved":
        # Re-ask user before escalating sandbox
        escalate_decision = @ask_user(
            "Tool '{call.tool_name}' was denied by sandbox. Allow with elevated permissions?"
        )
        if @judge("user rejected escalation: {escalate_decision}"):
            return first_result

    # Retry with no sandbox (escalated)
    escalated_result = run_in_sandbox(call, sandbox="none")
    return escalated_result

return first_result

def get_approval(call: ToolCall, config: TurnConfig) -> ApprovalDecision:
if config.sandbox_policy == "skip_approval":
return ApprovalDecision(kind="skip")

# Forbidden tools check
if @judge("tool '{call.tool_name}' is in the forbidden list for config {config}"):
    return ApprovalDecision(kind="forbidden", reason="policy")

return ApprovalDecision(kind="needs_approval")

def run_in_sandbox(call: ToolCall, sandbox: str) -> ToolResult:
try:
output = shell_exec(call.tool_name + " " + str(call.arguments))
return ToolResult(call_id=call.call_id, output=output)
fallback:
return ToolResult(call_id=call.call_id, output="execution failed", is_error=true)

─────────────────────────────────────────────────────────────

Single Turn Execution

Ref: codex.rs run_turn() + try_run_sampling_request()

One model call + all resulting tool calls

─────────────────────────────────────────────────────────────

def run_turn(
session_messages: list[str],
pending_input: str,
config: TurnConfig,
memories: list[MemoryRecord]
) -> (str, bool): # (agent_reply, needs_followup)

with context(system="You are Codex, an AI coding agent.") as ctx:

    # Inject retrieved memories as background context (RAG)
    if len(memories) > 0:
        ctx.remember("Relevant memories from past sessions:\n" + str(memories))

    # Inject skills/plugins available this turn (codex.rs ~line 6065)
    ctx.remember("Available tools: shell_exec, file_read, file_write, ...")

    # Inject conversation history
    for msg in session_messages:
        ctx.remember(msg)

    # ── Reasoning Phase ─────────────────────────────────────
    # In "plan" mode, model outputs a proposed plan before acting
    if config.collaboration_mode == "plan":
        prompt p_plan = """
            The user wants: {pending_input}
            First, think through what steps are needed.
            Output a numbered plan before taking any actions.
        """
        plan_text = @ask(p_plan)
        @confirm("Proceed with this plan?\n\n" + plan_text)

    # ── Sampling Request + Stream Processing ─────────────────
    # codex.rs try_run_sampling_request() ~line 7520
    # The model streams tokens; tool calls are extracted as they arrive

    prompt p_main = """
        User request: {pending_input}
        Think carefully and use the available tools to accomplish the task.
        Cite sources when drawing on retrieved context.
    """

    retry max=3:
        agent_reply = @ask(p_main, model="gpt-4o")
        @validate(agent_reply, "reply is not empty and addresses the user request")

    # Extract all tool calls the model wants to make
    tool_calls = @extract(agent_reply, "list of tool calls", type=list[ToolCall])

    # ── Parallel Tool Execution ──────────────────────────────
    # codex.rs FuturesOrdered ~line 7552 — tools run concurrently
    tool_results = []
    if len(tool_calls) > 0:
        results_raw = []
        for call in tool_calls:    # Rust uses FuturesOrdered for true parallelism
            r = orchestrate_tool(call, config)
            results_raw.append(r)
        tool_results = results_raw

    # ── Determine Follow-up Need ─────────────────────────────
    needs_followup = @judge(
        "any tool result indicates more work is needed or user input is required: {tool_results}"
    )

    # If tools produced output, synthesize final reply
    if len(tool_results) > 0:
        prompt p_synthesize = """
            Original request: {pending_input}
            Tool results:
            {tool_results}

            Synthesize a final, complete response to the user.
            Be specific about what was done and what the outcome is.
        """
        retry max=3:
            agent_reply = @ask(p_synthesize)
            @validate(agent_reply, "answer is grounded in tool results, no fabrication")

    return agent_reply, needs_followup

─────────────────────────────────────────────────────────────

Main Agent Entry Point

Ref: tasks/regular.rs RegularTask::run()

Outer loop: keeps processing pending inputs until none remain

─────────────────────────────────────────────────────────────

def run_codex_agent(task: TaskInput, config: TurnConfig) -> str:

# ── RAG: Retrieve relevant memories before starting ──────────
# Phase 1 and 2 run in background at session startup;
# here we perform runtime retrieval for this specific query.
candidate_memories = vector_search(task.user_query, top_k=10)
memories = rerank(task.user_query, candidate_memories)[:5]

session_messages: list[str] = []
pending_input = task.user_query
last_reply    = ""

# ── Outer Loop: process all pending inputs ───────────────────
# regular.rs lines 67-81
loop max=20 until @judge("no more pending input and task is complete: {pending_input}"):

    try:
        # Run a single model + tool-execution turn
        reply, needs_followup = run_turn(
            session_messages = session_messages,
            pending_input    = pending_input,
            config           = config,
            memories         = memories
        )

        last_reply = reply
        session_messages.append("assistant: " + reply)

        # Quality gate (codex.rs stop hooks ~line 6324)
        quality = @eval(reply, type={"relevance": float, "completeness": float})
        if quality.relevance < 0.7 or quality.completeness < 0.7:
            improvement = @ask("Improve this reply, current scores: {quality}\n\nReply: {reply}")
            last_reply = improvement
            session_messages.append("assistant (revised): " + improvement)

        if not needs_followup:
            break

        # If model is waiting for more user input, ask
        pending_input = @ask_user("Agent needs more information. Please respond:")
        session_messages.append("user: " + pending_input)

    fallback:
        # Turn-level error recovery (codex.rs ~line 6792)
        error_reply = @ask("The previous turn encountered an error. Summarize what was attempted and suggest next steps.")
        last_reply  = error_reply
        break

# ── Final Summary Output ─────────────────────────────────────
prompt p_summary = """
    Task: {task.user_query}
    Final result: {last_reply}

    Provide a concise completion summary:
    - What was accomplished
    - Any files changed or commands run
    - Any caveats or follow-up actions needed
"""
summary = @ask(p_summary)

# Persist to memory for future sessions (runtime memory.save)
memory.save(
    "Task: " + task.user_query + "\nResult: " + summary,
    key  = "session:" + task.user_query[:30],
    tags = ["history", "completed"]
)

@show(summary)
return summary

─────────────────────────────────────────────────────────────

Session Bootstrap — wires everything together

Ref: codex.rs session init + memories startup

─────────────────────────────────────────────────────────────

def bootstrap_session(task: TaskInput, config: TurnConfig) -> str:

# Background memory jobs run concurrently with main task
# (phase1 and phase2 are fire-and-forget at session start)
# In Rust these are spawned as tokio tasks; modeled here as parallel:
parallel:
    _m1 = extract_memories_phase1()
    _m2 = consolidate_memories_phase2(memory_root="/home/user/.codex/memories")
    result = run_codex_agent(task, config)

return result