Codex Harness(2026)
大约 23 分钟
Codex Harness 分析相关 agent harness 和 runtime，后续需要深入了解 OpenCode 分析 multi agent 以及可拓展性。
一、启动架构
终端执行 codex 之后，Codex CLI 的总入口在 codex-rs/cli/src/main.rs，其中通过 run_interactive_tui 启动 codex_tui::run_main -> startup_orchestration::run_main_inner()
1.1 后端链接方式
run_main_inner() 在此处先决定 UI 的后端链接方式
Embedded：后端 app-server 直接运行在当前codex进程里LocalDaemon：连接本机已经运行的 app-server daemonRemote：连接远程 app-server
而后根据选择配置数据库，环境等。然后调用 run_ratatui_app(...) 启动 app server，最后通过一个 AppServerSession 包装启动，类似：
async def run_ratatui_app(...):
app_server = AppServerSession(...)
result = await App.run(
tui=tui,
app_server=app_server,
config=config,
...
)
1.2 App 启动方式
App 中选择如何启动，如 StartFresh 启动完全新对话, Resume 或 Fork。
App.run 中配置启动参数，如 model, workspace, permissions 等。然后通过 JSON-RPC（本地启动）向 app-server 发送 thread/start。
app-server 启动 thread-start，内容包括：
AgentControl：把 multi-agent 控制能力 塞进即将创建的 thread。thread_source和session_source：在 继续对话的时候有用到。ThreadSpawnRequest：包装所有启动需要的东西，获得ThreadManagerState。ThreadManagerState里面有Session.spawn()。
Session.spawn() 启动 session，把 tool，skills 等东西传入启动。
session, io = await Session.spawn(
SessionSpawnArgs(
config=config,
# instructions
user_instructions=user_instructions,
# auth / model
auth_manager=auth_manager,
models_manager=models_manager,
# environment
environment_manager=environment_manager,
environment_selections=environments,
# skills
skills_service=skills_service,
# plugins
plugins_manager=plugins_manager,
# MCP
mcp_manager=mcp_manager,
client_mcp_extensions=client_mcp_extensions,
# history
conversation_history=initial_history,
# agent
agent_control=agent_control,
# tools
dynamic_tools=dynamic_tools,
# session relationships
session_source=session_source,
thread_source=thread_source,
parent_thread_id=parent_thread_id,
forked_from_thread_id=forked_from_thread_id,
...
)
)
其中 session 是 agent 运行状态， io 是外界与 session 通信的通道。
这边的 Session 伪代码：
class Session:
@staticmethod
async def spawn(args):
return await Session.spawn_internal(args)
@staticmethod
async def spawn_internal(args):
# 下文中 Harness 中介绍 Session.new
session = await Session.new(...)
# 只是共享同一个 session 引用
session_for_loop = clone_reference(session)
create_task(
submission_loop(session_for_loop)
)
io = SessionIo(...)
# 返回刚才 Session.new 创建的那个 session
return session, io
二、Agent Harness
在 session.spawn 后，可以看到部分 Harness 涉及到的变量。Harness 过程中存在一些预处理步骤
- 构建
UserInstruction：这个用户 / host / project 希望 Codex 在这个工作环境里遵守什么规则？ - 加载
Exec Policy：允许、拒绝、需要 - 确定使用的 model
- 构建 base instruction：Base instructions for the session。包含这个 agent 应该干什么之类的。可能从 config，或者 history 或者 model_info 中获取。
- 构建 developer_instructions：Developer instructions that supplement the base instructions.
- 确定
Dynamic Tools：这些工具是 Thread 启动时确定的 tool surface 的一部分 ，而且会被写进 thread metadata，resume 时还能恢复。 SessionConfiguration：把上面的信息加入到SessionConfiguration配置中。
Runtime
Codex harness 的核心 runtime 结构
Session
│
├── SessionState
│ ├── SessionConfiguration
│ ├── ContextManager ← conversation history/context
│ └── additional context
│
└── SessionServices
├── ModelClient ← 调 LLM
├── SkillsService ← skills
├── AgentsMdManager ← AGENTS.md/instructions
├── McpRuntime ← MCP tools
├── McpManager
├── PluginsManager
├── AgentControl ← multi-agent
├── ExecPolicy
├── TurnEnvironments
├── Hooks
└── Extensions
Session 启动
Session 启动，然后进入主事件循环。
session = await Session.new(
session_configuration,
environment_selections,
config,
user_instructions,
auth_manager,
models_manager,
model_info,
exec_policy,
conversation_history,
skills_service,
plugins_manager,
mcp_manager,
agent_control,
environment_manager,
...
)
session_loop = asyncio.create_task(
submission_loop(
session=session,
config=config,
submissions=submission_rx,
)
)
io = SessionIo(
tx_sub=submission_tx,
rx_event=event_rx,
agent_status=agent_status_rx,
session_loop=session_loop,
)
return session, io
其中 IO：
class SessionIo:
tx_sub # 发消息给 Session
rx_event # 接收 Session 的事件
agent_status
async def Session_new(...):
# 1. 确定 thread/session identity # L759
# 2. 初始化持久化、environment、shell # L845 AgentsMdManager.refresh()
# 3. 加载 AGENTS.md 其中包含 user_instruction # L1224
# 4. 预加载 plugins / skills # L1225 SkillsService.snapshot_for_config()
# 5. 创建 SessionState -> ContextManager # L1257 ContextManager()
# 6. 创建 MCP / hooks / tool runtime # L1323
# 7. 创建 SessionServices # L1398
# 8. 创建 Session 对象 # L1488
# 9. 发 SessionConfigured # L1528
# 10. 安装 MCP runtime # L1589
# 11. 写入 initial history # L1610
return session
三、Agent Loop
3.1 submission_loop
通过上面构造的 session 构造 submission_loop。
session_loop_handle = create_task(
submission_loop(
session_for_loop,
configured_config,
rx_sub,
)
)
session submission_loop 在 codex-rs/core/src/session/handlers.rs
- 用户消息进入
submission_loop：决定是要 start new，还是 steer 当前 turn。（类似返回一个标签，不会真正执行） - 创建
TurnContext(如果没有的话) - 新创建的对话使用功能
RegularTask.run()，：
# RegularTask.run()
while True:
# 似乎用于应对用户连续提问，或者 pending input
result = await run_turn(
session,
turn_context,
input,
model_client_session,
)
if no_pending_input:
return result
input = []
3.2 run_turn()
Turn：Agent 针对一次用户请求进行完整处理的生命周期。
async def run_turn(
session,
turn_context,
input,
mcp_startup_requirements,
client_session,
cancellation_token,
):
# 1. 上一轮 hook 结果
await drain_async_hook_results(...)
# 2. 创建/复用 model client session
client_session = (
client_session
or session.model_client.new_session()
)
# 3. 必要时提前 compact
await run_pre_sampling_compact(
session,
turn_context,
client_session,
)
# 4. 提取本轮 user input
user_input = turn_user_input(input)
# 5. 根据输入解析 MCP / plugin requirements
required_plugins += collect_explicit_plugin_ids(user_input)
required_servers, mentioned_plugins = (
await required_mcp_servers_for_input(
session,
turn_context,
user_input,
)
)
# 6. 构造第一次 StepContext
first_step_context = (
await session.capture_step_context_with_required_mcp_servers(
turn_context,
required_servers,
required_plugins,
)
)
# 7. 把 runtime/context 的变化写入 history
world_state = (
await session.record_context_updates_and_set_reference_context_item(
first_step_context
)
)
# 8. 构造 skill/plugin 注入项
injection_items, enabled_connectors = (
await build_skills_and_plugins(
session,
first_step_context,
user_input,
mentioned_plugins,
)
)
# 9. hooks + 记录真正的用户输入
await run_hooks_and_record_inputs(
session,
turn_context,
input,
)
# 10. skill/plugin injection 写入 history
for item in injection_items:
await session.record_conversation_items(
turn_context,
[item],
)
next_step_context = first_step_context
# ============================================
# Agent loop
# ============================================
while True:
# 11. 获取 steer / pending input
pending_input = await get_pending_input(...)
await run_hooks_and_record_inputs(
session,
turn_context,
pending_input,
)
# 12. 确定这一轮 sampling 使用哪个 StepContext
if pending_input:
# steer 可能改变 MCP/plugin requirements
update_required_mcp_and_plugins(pending_input)
step_context = (
await session.capture_step_context_with_required_mcp_servers(
turn_context,
required_servers,
required_plugins,
)
)
elif next_step_context is not None:
step_context = next_step_context
next_step_context = None
else:
step_context = (
await session.capture_step_context_with_required_mcp_servers(...)
)
# 13. runtime/world-state 变化写入 history
world_state = (
await session.record_step_world_state_if_changed(
world_state,
step_context,
)
)
# 14. ContextManager → 模型 input
sampling_request_input = (
await session.clone_history()
).for_prompt(
step_context.settings.model_info.input_modalities
)
# 15. 进入 sampling controller
sampling_result, used_input = (
await run_sampling_request(
session=session,
step_context=step_context,
client_session=client_session,
input=sampling_request_input,
...
)
)
# 16. 判断是否需要下一次 sampling
if sampling_result.needs_follow_up:
continue
# 还会检查 pending input / compact / stop hooks 等
...
break
3.3 要点
- MCP/tool 在 steer 后更新，更加载到 stepContext 中
- plugin 分 2 种，plugin 自带的描述性内容会被添加到
developerinstruction 中(role = developer) ，这部分 steer 时候不会重新加载。但 plugin 这包含的 tool/MCP 会被重新整理加入到request.tools当中 。 - agent.md 好像是 session 更新时候加载的。
- skill 在 steer 过程中不会被重新加载，但在不是 steer （
run_turn()重新被执行）的时候会重新加载。一个run_turn()只加载一次 skill。 - history 是 ContextManager 的实例。
- prompt 具体拼接，最后发送给 API 的 request；
.for_prompt，run_sampling_request，build_prompt里面具体怎么把所有东西整合。
3.4 不同 Context
TurnContext 大部分用在环境或者配置上：
class TurnContext:
sub_id: str
trace_id: str | None
config: Config
initial_settings: ResolvedStepSettings
current_settings: ResolvedStepSettings
auth_manager: AuthManager | None
provider: ModelProvider
environments: TurnEnvironmentSnapshot
network: NetworkProxy | None
windows_sandbox_level: WindowsSandboxLevel
unified_exec_shell_mode: UnifiedExecShellMode
session_source: SessionSource
history_mode: ThreadHistoryMode
parent_thread_id: ThreadId | None
current_date: str | None
timezone: str | None
cwd: AbsolutePath
developer_instructions: str | None
multi_agent_version: MultiAgentVersion
dynamic_tools: list[DynamicToolSpec]
extension_data: ExtensionData
available_models: list[ModelPreset]
final_output_json_schema: dict | None
session_telemetry: SessionTelemetry
turn_metadata_state: TurnMetadataState
turn_timing_state: TurnTimingState
terminal_error: ErrorEvent | None
这几个东西用来储存请求中几乎所有用到的信息。
class StepContext:
# 基本不变 同一个 run_turn() 中，通常始终引用同一个 TurnContext。
turn: TurnContext
# 可能变化 每次重新 capture StepContext 时，从
# turn_context.current_settings 获取当前 settings snapshot。
# 例如 model、reasoning 等配置发生更新。
settings: ResolvedStepSettings
# 可能变化 根据当前 step 的 model/settings 重新解析出的 token budget。
token_budget: TokenBudgetConfig | None
# 可能变化 当前 sampling request 对应的 telemetry context。
# 不属于模型真正看到的 prompt context。
session_telemetry: SessionTelemetry
# 可能变化 当前 step 的 environment snapshot。
# environment readiness / attachment 状态变化时可能不同。
environments: TurnEnvironmentSnapshot
# 可能变化 当前已经 ready、并绑定到 environment 的 capability roots。
selected_capability_roots: list[ResolvedSelectedCapabilityRoot]
# 可能变化
# 当前 step 已 materialize / discover 到的 capability 文件。
executor_capability_discovery: ExecutorCapabilityDiscoverySnapshot | None
# 可能变化，重要
# 当前 step 对应的 MCP binding：
# MCP connections + config + tool catalog。
#
# steer 中 mention 新 MCP/plugin 后，
# 下一次 capture StepContext 时这里可能发生变化。
mcp: McpBinding
# 可能变化，最重要
# 当前 sampling request 最终使用的 ToolRouter。
#
# MCP/plugin/environment/tool policy 变化后会重新构建，
# 最终：
# tool_router.model_visible_specs()
# -> Prompt.tools
# -> request.tools
tool_router: ToolRouter
# 通常不变，但属于 step snapshot
# capture 时读取 agents_md_manager.get_loaded()。
# 当前 run_turn 普通 loop 中不会每次重新 refresh AGENTS.md，
# 所以一般多个 StepContext 看到的是同一个已加载版本。
loaded_agents_md: LoadedAgentsMd | None
ContextManager
class SessionState:
def __init__(...):
self.history = ContextManager()
self.additional_context = AdditionalContextStore()
self.previous_turn_settings = None
...
其中
class ContextManager:
def __init__(self):
# oldest -> newest
self.items: list[ResponseItemEnvelope] = []
def for_prompt(self, input_modalities):
"""
把内部 history 转成最终可以发给模型的 ResponseItem[]
"""
envelopes = self.for_prompt_annotated(input_modalities)
# 去掉 Codex 自己使用的 envelope / metadata
return [
envelope.item
for envelope in envelopes
]
def for_prompt_annotated(self, input_modalities):
# 在“发给模型”之前规范化 history
self.normalize_history(input_modalities)
return self.items
def normalize_history(self, input_modalities):
# 1. 每个 tool/function call 必须有对应 output
ensure_call_outputs_present(self.items)
# 2. 删除没有对应 call 的孤立 output
remove_orphan_outputs(self.items)
# 3. 如果当前模型不支持 image
# 删除/处理 history 中的 image 内容
strip_images_when_unsupported(
input_modalities,
self.items,
)
# 4. 如果模型不支持 audio
# 删除/处理 audio 内容
strip_audio_when_unsupported(
input_modalities,
self.items,
)
3.5 最终请求
最终请求可以写成：
request = ResponsesApiRequest(
model=step_context.settings.model_info.slug,
instructions=prompt.base_instructions.text,
input=prompt.get_formatted_input_for_request(
use_responses_lite=False
),
tools=serialize_tools(prompt.tools),
tool_choice="auto",
parallel_tool_calls=prompt.parallel_tool_calls,
reasoning=build_reasoning_config(...),
output_schema=prompt.output_schema,
store=False,
stream=True,
stream_options=...,
include=...,
metadata=...,
)
其中 input 为统一的 list[ResponseItem] 形式，发送给 API，然后 API 端的 LLM 调用对应的 chat_template 组装成字符串。
ResponseItem =
# 消息 / reasoning
AdditionalTools
Message
AgentMessage
Reasoning
# tool 调用相关
LocalShellCall
FunctionCall
FunctionCallOutput
ToolSearchCall
ToolSearchOutput
CustomToolCall
CustomToolCallOutput
WebSearchCall
ImageGenerationCall
# context / control
Compaction
ContextCompaction
ConfigurationUpdate
CompactionTrigger
Other
四、各组件分析
4.0 WordState
管理的是一批“可能随运行状态变化、需要做 diff 的模型上下文”。
# 伪代码
class WorldState:
def __init__(self):
self.sections = []
def add_section(self, section):
self.sections.append(section)
def snapshot(self):
return {
section.name: section.snapshot()
for section in self.sections
}
def diff(self, previous_snapshot):
updates = []
for section in self.sections:
old = previous_snapshot.get(section.name)
new = section.snapshot()
if old != new:
updates.append(section.render_update(old))
return updates
在 Codex 里，每个 section 可以代表一种状态，比如：
world_state = WorldState()
world_state.add_section(ModelState(current_model))
world_state.add_section(AgentsMdState(loaded_agents_md))
world_state.add_section(PermissionsState(sandbox_policy))
world_state.add_section(ManagedDeveloperInstructionsState(policy))
world_state.add_section(EnvironmentState(cwd, current_date))
world_state.add_section(ToolsState(available_tools))
第一次调用模型时，可以把完整状态放进去：
messages += world_state.render_full()
之后如果状态发生变化，就会把变化放到 messages 中：
new_world_state = build_world_state()
updates = new_world_state.diff(old_world_state.snapshot())
messages += updates
比如原来：
ResponseItem::Message(user)
AGENTS:
"Use pytest"
后来变成：
Use pytest
Run mypy
那么没必要重新发送所有状态，只需要添加一个新的 ReponseItem.Message 告诉模型：
"These AGENTS.md instructions replace all previously
provided AGENTS.md instructions.
Use pytest
Run mypy"
这种设计主要是为了保持之前的 cache 不变动吧？不过似乎也有少数的操作能够移除或者重构之前的 Message（如 reset）
4.1 Instructions
简要说明：除了普通 developer instructions，其他的 instructions 基本上都会在 turn 的过程中通过 wordstate 获取到更新信息。
Instructions 分为 3 种， base, developer 还有 user。
| 项目 | base_instructions / developer_instructions | user_instructions |
|---|---|---|
| 属于谁 | Session / harness 本身 | 用户 / workspace |
| 存在哪里 | SessionConfiguration | AgentsMdManager |
| 生命周期 | 基本属于整个 Thread 的稳定配置 | 会随着 workspace / environment 的 AGENTS.md 刷新 |
| 主要来源 | model metadata / config / app-server | host 用户配置 |
| 是否和 AGENTS.md 合并 | 不会 | 会 |
| 语义 | “Codex 应该是什么、怎么运行” | “这个用户/项目希望 Codex 怎么做” |
- 对于
baseInstruction
session.spawn 初始阶段，模型选出来以后，Codex 开始解析。不依赖 StepContext 和 sampling_request_input 传递； run_sampling_request 直接从 Session 读取，最终进入 request.instructions。
- 对于
Developer Instructions
最终都变成 ResponseItem.Message(role="developer") 加入 request.input 而非 stepContext
- Collaboration Model Instructions: 如 Plan Mode/ Default Mode 自己带的 developer instructions.
- Managed Developer instructions: 如项目组织者或企业统一制定的 instructions。
- 普通 Developer instructions: 如用户
config.toml文件中读取的developer_instructions值
- 对于
UserInstruction
全局 AGENTS.MD 在 spawn 前加载。
session.new 的时候加载项目 AGENTS.md。不过每个 step 都会通过 WordState 检查是否有更新 AGENTS.md。
UserInstructions 最终都变成类似
ResponseItem.Message(
role="user",
content=[
InputText("""
# AGENTS.md instructions ...
<INSTRUCTIONS>
...
""")
]
)
- sub-agent 处理
| instruction 类型 | sub-agent 怎么处理 |
|---|---|
| Base/model instructions | child session 自己根据 model/config 构造 |
普通 developer_instructions | 可能继承，也可能被 sub-agent 专用 instruction 替换 |
| Managed developer instructions | 保留或由 child 当前 requirements 重新生成 |
AGENTS.md | 属于 child world state，按 child 环境重新构造 |
| Multi-agent mode instruction | 不直接继承父版本 |
| Multi-agent role instruction | 删除父版本，child 添加自己的 role instruction |
| Multi-agent usage hint | 删除父版本，child 添加 sub-agent usage hint |
| Current-time reminder | 不继承父版本 |
| Persistent/reasoning-mode instruction | 保留或由 child 当前状态重新构造 |
| 父 agent 的普通 user conversation | fork 模式下可以继承部分 history |
4.1 Tools
Codex 不希望 Context 一开始就承载全部 capability 的完整描述。 这是它现在 Harness 设计里很核心的一条原则；Codex 自己的 prompting guide 也明确建议只暴露任务相关工具。不是所有已注册 Tool 的 description 都会在每次请求里完整放进 LLM context。
Direct：初始就放进 model-visible tools。Deferred：先不把完整 tool schema/description 给模型，只注册为可搜索；需要时通过tool_search找出来。DeferredModelOnly：可通过 tool search 发现，但不进入初始工具列表。CodeModeOnly：只给 Code Mode。
Codex 有专门的 tool_search 工具，第一次模型调用可能只给：
shell
read_file
write_file
apply_patch
git
tool_search
而 不会把 Salesforce 的 50 个 tool description、Slack 的 50 个、Drive 的 30 个全部塞进去 。模型可以自己调用 tool_search 将其他的 tool 信息引入到 context 中。
| 类型 | 来源 | 谁执行 | 例子 |
|---|---|---|---|
| Core / Internal Tools | Codex Core 内置 | Codex harness | exec_command、apply_patch、view_image、spawn_agent |
| MCP Tools | MCP Server | MCP Server | GitHub/数据库/第三方 MCP 工具 |
| Extension Tools | Codex extension/plugin | Extension executor | 插件提供的工具 |
| Dynamic Tools | Thread 启动时由 host 动态注入 | 外部 host/client | 自定义 query_database |
| Hosted Model Tools | 模型/API backend 提供 | OpenAI backend | 典型如 hosted web_search |
相关对象：
ToolRegistry：储存和管理 Tool 相关 runtime。ToolRouter：连接 LLM 和 ToolRegistry。发送 LLM request 之前会用ToolRouter来调动ToolRegistry整理出发送给 API 的 Tool list 信息。LLM function call 时候，先发给ToolRouter来调动ToolRegistry运行相关工具。
class ToolRegistry:
def __init__(self):
# tool name -> runtime
self.tools = {}
def register(self, runtime, exposure):
self.tools[runtime.name] = {
"runtime": runtime,
"exposure": exposure,
}
def get(self, tool_name):
return self.tools.get(tool_name)
async def dispatch(self, invocation):
tool = self.get(invocation.tool_name)
if tool is None:
return ToolError("unsupported tool")
runtime = tool["runtime"]
# 检查调用类型是否正确
if not runtime.matches(invocation.payload):
return ToolError("invalid tool call")
# 执行前 hook
invocation = run_pre_tool_hooks(
runtime,
invocation,
)
# 真正执行工具
result = await runtime.handle(invocation)
# 执行后 hook
result = run_post_tool_hooks(
runtime,
invocation,
result,
)
return result
和
class ToolRouter:
def __init__(self, registry, model_visible_specs):
self.registry = registry
self.model_visible_specs = model_visible_specs
def get_model_visible_tools(self):
# 给 LLM 的 tool schemas
return self.model_visible_specs
def build_tool_call(self, response_item):
# 把 LLM 返回的 function/custom tool call
# 转成 Codex 内部统一的 ToolCall
return ToolCall(
tool_name=response_item.name,
call_id=response_item.call_id,
payload=response_item.arguments,
)
async def dispatch_tool_call(self, tool_call, context):
# ToolCall -> ToolInvocation
invocation = ToolInvocation(
tool_name=tool_call.tool_name,
call_id=tool_call.call_id,
payload=tool_call.payload,
context=context,
)
# 真正执行交给 ToolRegistry
result = await self.registry.dispatch(invocation)
return result
关于 Tool 的更新，基本上每次 step 运行都会更新一下 tool，如下所示。比较不同的是 skill 中涉及到的 Tool 处理方式。
# 每次准备发模型请求前
def build_tool_router(turn_context):
registry = ToolRegistry()
# Codex 自带工具
load_core_tools(registry)
# 当前 MCP server 提供的工具
load_mcp_tools(registry)
# extension / plugin 提供的工具
load_extension_tools(registry)
# 当前 turn 传入的 dynamic tools
load_dynamic_tools(registry)
# 根据 exposure / tool mode 筛选
visible_tools = build_model_visible_specs(registry)
# hosted tool，例如 provider 提供的 web search
visible_tools += load_hosted_tools()
return ToolRouter(
registry,
visible_tools,
)
4.2 MCP
总结：stepContext 中储存内部运行状态。
Prompts.tools中储存主要的 MCP 中的ToolInfo。Codex 根据当前 MCP 的状态（如是否可用）等生成world-state instruction储存在input中。
几个相关对象：
MCPManager：整合 Config，Plugin，Extension 等地方涉及到的 MCP 配置。McpRuntime：统一管理一个 Codex Thread 里所有 MCP 的状态。McpBinding：保存在StepContext里的 MCP 信息：The exact MCP connections, configuration, and catalog captured for this step。binding 信息不会直接发送给 LLM，是内部状态。McpConnectionSet：管理链接。McpHandler：根据 MCPToolInfo生成 Codex 的ToolSpec，通常放在Prompt.tools
class McpConnectionSet:
clients = {
"github": github_client,
"filesystem": filesystem_client,
...
}
async def call_tool(server, tool, args):
client = clients[server]
check_tool_filter(server, tool)
return await client.call_tool(tool, args)
4.3 Skill
根据下面的情况，添加新的 skill 会改变 developer message。
Codex 会扫描并加载 User / Repo / System / Admin 不同层级的 skill。扫描 skill 时，读取 skill.md 并保存其中的 SkillMetadata 信息，和 openai.yaml 信息，而非全部 skill.md 信息。
SkillMetadata {
name,
description,
short_description,
interface,
dependencies,
policy,
path_to_skills_md,
scope,
plugin_id,
...
}
yaml 可能提供
interface:
display_name: PDF
dependencies:
tools:
- type: mcp
value: ...
policy:
allow_implicit_invocation: true
- Available Skills Catalog
整合 SkillMetadata，所有可使用的 skill 的 metadata 通过 role="developer" 加入到同一个 message 当中。
<skills_instructions>
Available skills:
- pdf
description: Create/edit PDF...
path: ~/.codex/skills/pdf/SKILL.md
- slides
description: Create presentations...
path: ...
How to use skills:
...
</skills_instructions>
- 用户选择的 skill（Explicit Skill Invocation）
用户使用 $(skill-name) 符号来指定需要用哪个 skill。
codex 中使用 collect_explicit_skill_mentions(...) 来抓取 $ 信息。后续其他函数加载 SKILL.MD。
最后通过 role="user" 添加到 message 当中，作为单独的一个 ResponseItem.Message 加入到 history Messages 当中。
<skill>
<name>pdf</name>
<path>/.../pdf/SKILL.md</path>
[optional resource_access]
这里是 SKILL.md 的正文内容
</skill>
- 模型第一次回复
Codex 中设定了模型的回复逻辑，根据以下逻辑可以判断，模型会使用对应工具阅读 skill 内容。skill 内容加载到 input 中之后，后续就是各种 agent loop 了。
const SKILLS_HOW_TO_USE_WITH_SOURCE_LOCATORS: &str = r###"- Discovery: The list above is the skills available in this session (name + description + source locator). `file` entries live on the host filesystem, `executor package` and `orchestrator package` entries are accessed directly through `skills.read`, and `custom resource` entries use their provider's access mechanism.
- Trigger rules: If the user names a skill (with `$SkillName` or plain text) OR the task clearly matches a skill's description shown above, you must use that skill for that turn. Multiple mentions mean use them all. Do not carry skills across turns unless re-mentioned.
- Missing/blocked: If a named skill isn't in the list or its source can't be read, say so briefly and continue with the best fallback.
- How to use a skill (progressive disclosure):
1) After deciding to use a skill, the main agent must read its `SKILL.md` completely before taking task actions. For a `file` entry, open the listed path. For an `executor package` or `orchestrator package`, pass the listed locator directly to `skills.read` as `package`; root aliases are resolved automatically. Omit `resource` to read `SKILL.md` directly without calling `skills.list`. If a read is paginated, follow `next_cursor` until EOF.
2) When `SKILL.md` references another resource, use the same access mechanism. For executor and orchestrator skills, pass the complete package-contained resource identifier with the same package to `skills.read`; do not treat `skill://` identifiers as filesystem paths.
3) If `SKILL.md` points to extra folders such as `references/`, use its routing instructions to identify the resources required for the task. The main agent must read each required instruction or reference file itself before acting on it. Do not delegate reading, summarizing, or interpreting skill instructions to a subagent. Subagents may still perform task work when the selected skill allows it.
4) For filesystem-backed skills, prefer running or patching provided scripts instead of retyping large code blocks. For executor and orchestrator skills, use `skills.read` and the available tools; do not invent a local path.
5) Reuse provided assets or templates through the same source access mechanism instead of recreating them.
- Coordination and sequencing:
- If multiple skills apply, choose the minimal set that covers the request and state the order you'll use them.
- Announce which skill(s) you're using and why (one short line). If you skip an obvious skill, say why.
- Context hygiene:
- Progressive disclosure applies to selecting relevant files, not partially reading a selected instruction file. Do not load unrelated references, scripts, or assets.
- Avoid deep reference-chasing: prefer opening only files directly linked from `SKILL.md` unless you're blocked.
- When variants exist (frameworks, providers, domains), pick only the relevant reference file(s) and note that choice.
- Safety and fallback: If a skill can't be applied cleanly (missing files, unclear instructions), state the issue, pick the next-best approach, and continue."###;
对于 explicit（$pdf），skill.md 整个内容会被提前放在 context 当中，后续要调用 tool 或者回复 response 由模型决定。
Thread 初始化时已经进行了 skill discovery / snapshot warmup。
async def warm_plugins_and_skills(...):
plugins = await plugins_manager.plugins_for_config(...)
plugin_skill_roots = plugins.effective_plugin_skill_roots()
plugin_skill_snapshots = (
plugins_manager.plugin_skill_snapshots_for_config(...)
)
skills_input = skills_load_input_from_config(
config,
plugin_skill_roots,
)
skills_input.add_plugin_skill_snapshots(
plugin_skill_snapshots
)
return await skills_service.snapshot_for_config(
skills_input,
filesystem,
)
4.4 API 的 prompt
openai 透露了一些 Model Server 对 input 的整理方式，每个模型不同，取决于模型训练时候的排列方式。
五、权限和环境
5.1 Execution Environment
由 codex runtime 管理，LLM prompt 会包含部分环境信息。
struct TurnEnvironment {
environment_id: String,
environment: Arc<Environment>, // 可以 执行进程，读写文件，发网络请求，等
cwd: PathUri,
workspace_roots: Vec<PathUri>,
shell: Option<Shell>, // 如 { shell_type: Zsh, shell_path: "/bin/zsh" }
shell_snapshot: ..., // 捕捉用户 shell 初始化之后的状态，如 `.zscrc` 运行后的状态。
}
5.2 Permission / Approval
pub struct Permissions {
pub approval_policy: Constrained<AskForApproval>,
permission_profile_state: PermissionProfileState,
managed_deny_read_policy:
Option<Arc<FileSystemSandboxPolicy>>,
workspace_roots: Vec<AbsolutePathBuf>,
pub network: Option<NetworkProxySpec>,
pub allow_login_shell: bool,
pub shell_environment_policy:
ShellEnvironmentPolicy,
pub windows_sandbox_mode:
Option<WindowsSandboxModeToml>,
pub windows_sandbox_private_desktop: bool,
}
Permission Profile
有点像一套 chmod 755 file； chmod 600 key； iptables ... ；ACL ...； SELinux rules ... 组合。然后起个什么命名，比如 developer permission profile.
负责 codex 运行的进程可以访问什么内容。比如
- FileSystem Permission：Permission Profile 最终会产生类似：
FileSystemSandboxPolicy，然后传给 sandbox。 - Network Permission：管理是否能够联网，如
network = disabled。需要与 Network proxy 区分。
codex 自带 3 个 profile：
| Profile | File System | 含义 |
|---|---|---|
:read-only | 只读 | 看代码，不修改 |
:workspace | workspace 可写 | 正常 Agent 开发（请求批准和帮我批准） |
:danger-full-access | unrestricted | 基本不做本地 sandbox 限制（如 codex 中的完全访问权限） |
Permission Profile 可以在 Managed / Disabled / External 之间选择使用 codex 自带管理，关闭或者使用外放配置好的 sandbox。
Permission 决定“当前无需额外授权就能做什么”；Approval 决定“当操作超出当前权限，或者策略认为该操作需要额外确认时，能否获得授权继续执行”
Permission 中有 approval_policy，核心模式包括：
untrusted / UnlessTrusted # 通过 Exec Policy 判断是否为安全命令，然后问用户决定是否需要 approval
on-request # 超出范围问用户
never # 从不问用户，直接拒绝超出范围
granular # 哪些类型可以问、哪些不能问，分别配置
Permission 并不是一开始就全部配置好：
- Addition Permission 可以让模型在执行 tool call 发现权限不够时候，添加上新的 Permission。
apply_granted_turn_permissions或permissions_preapproved可以在一个 turn / session 范围内添加 Permission（类似批准这类操作）
5.3 Sandbox
LLM 生成了一条准备执行的命令之后，如何确保这条命令只能访问被允许的文件、目录、网络和系统资源。在真正执行 command 时，用操作系统机制强制执行权限边界。
可以粗略得把 sandbox 理解成：
class Sandbox:
filesystem: FileSystemPolicy
network: NetworkPolicy
enforcement: SandboxEnforcement
def execute(self, command):
env = build_sandbox_environment(
filesystem=self.filesystem,
network=self.network,
enforcement=self.enforcement,
)
return env.run(command)
permission 更改或者 approve 之后，sandbox 会添加上新的规则，然后被重新创建，类似：
base = turn.permission_profile
approved = AdditionalPermissionProfile(
file_system_write=["/foo"]
)
effective = merge(base, approved)
sandbox_policy = effective.to_runtime_permissions()
run_command(
command,
sandbox_policy=sandbox_policy
)
六、拓展功能
6.1 Hooks
几个常见的 Hook
- UserPromptSubmit：用户消息正式进入 conversation 之前就可以检查它
- PreToolUse：发生在
ToolRegistry生成 payload 之后，调用 tool 之前。也在 Permission / Approve / Sandbox 之前。逻辑可以简化成：
def dispatch_tool(invocation):
tool = registry.get(invocation.tool_name)
hook_result = pre_tool_use(invocation)
if hook_result.blocked:
return error
if hook_result.updated_input:
invocation = rewrite(invocation)
return tool.handle(invocation)
- PermissionRequest：在 Approve 后调用，逻辑可以简化成：
if approval_required:
decision = permission_request_hook(...)
if decision == ALLOW:
execute()
elif decision == DENY:
reject()
else:
guardian_or_user_approval()
- PostToolUse：Tool output 生成后，用于检查 Tool output 是否应该被加入到 context 当中。
- Stop：实际上用于 Agent 认为自己完成，但外部检查不成功，让 agent 继续工作。类型包扣
should_stop,stop_reason,should_block,block_reason,continuation_fragments。 - UserPromptSubmit：输入过滤
- PreCompact/PostCompact：compact 生命周期
- SubagentStart/Stop：multi-agent 生命周期
Hook 的结果可以影响整个流程：
- 控制流程：如
{
"continue": false,
"stopReason": "policy violation"
}
- block：禁止某些操作，如禁止某些 Tool Call
- 修改 Tool 输入：如 preToolUse 修改 Tool call 的输入
- 修改 Context：如 record_additional_contexts(...) 直接在 context 这个添加内容。
似乎可以添加自己的 Hook，可以考虑添加类似监控模型请求的 Hook，Tool 状态的 Hook，或者 sandbox 状态的 Hook。
6.2 Plugins
Openai 的 plugin 都要求包括：
# plugin 的 companion surfaces
plugin-root/
│
├── .codex-plugin/
│ └── plugin.json # required manifest
│
├── skills/
│ ├── skill-a/
│ │ └── SKILL.md
│ └── skill-b/
│ └── SKILL.md
│
├── .mcp.json
├── .app.json
│
├── hooks/
│ └── hooks.json
│
├── agents/
├── commands/
├── assets/
└── ...
- plugin.json 可以近似理解成：
class PluginManifest:
name: str
version: str | None
description: str | None
keywords: list[str]
paths: PluginManifestPaths
interface: PluginInterface | None
class PluginManifestPaths:
skills: list[Path]
mcp_servers: McpConfig | None
apps: Path | None
hooks: HookConfig | None
plugin.json 加载后变成类似
class LoadedPlugin:
config_name: str
manifest_name: str | None
description: str | None
root: Path
enabled: bool
skill_roots: list[Path]
disabled_skill_paths: set[Path]
has_enabled_skills: bool
mcp_servers: dict[str, McpServerConfig]
apps: list[AppDeclaration]
hook_sources: list[HookSource]
error: str | None
6.3 apps
An app is equivalent to a set of MCP tools within the codex_apps MCP
Apps 可以理解为 Codex 认证并进行管理的 MCP。Apps 用来连接外部 tools、information、actions；Plugin 可以把一个或多个 App 和 Skills 组合成完整 workflow。
apps 通过 .app.json 定义，如：
{
"apps": {
"github": {
"id": "some_connector_id"
}
}
}
6.4 Extensions
Extensions 不直接参与 codex harness 的构造，而是用来支持 vscode 这种 rich interface 的接口。整体关系可以看做：
┌─────────────────────────────┐
│ VS Code │
│ │
│ Codex Extension │
│ - UI │
│ - editor integration │
│ - approval UI │
│ - thread / turn control │
└──────────────┬──────────────┘
│
│ JSON-RPC
▼
┌─────────────────────────────┐
│ codex app-server │
│ │
│ thread/start │
│ thread/resume │
│ turn/start │
│ approvals │
│ events │
│ skills / apps / plugins API │
└──────────────┬──────────────┘
│
▼
┌─────────────────────────────┐
│ Codex Core / Harness │
│ │
│ Instructions │
│ Skills │
│ Tools / ToolRouter │
│ MCP │
│ Plugins │
│ Hooks │
│ Sandbox │
│ Permissions / Approval │
│ Agent loop │
└─────────────────────────────┘
七、Compaction
Codex 当前的 compaction prompt 非常直白：
“Create a handoff summary for another LLM that will resume the task.”
当 auto-compact limit 或者 model context window 达到条件时，就会进入 compaction。
compact 分成 auto, manual 还有 remote，local。
compact 检查点：
run_turn()刚开始，context 还没有更新前。（用户新一轮对话刚刚开始的时候）run_turn()loop 过程中，一次 sampling 结束之后进行 token check（auto compact limit 或 model context）。
token 检测方式：
- BodyAfterPrefix：系统 prefix 之后，用户真正占用的 token。
- total：全部 context 上下文
中间过程中有一个 should_compact_mid_turn，可以再用户提供新一轮信息后进行 compact。
should_compact_mid_turn = (
needs_follow_up # has_pending_input
and (
token_limit_reached
or new_context_window_requested
)
)
此外 compact 也因为模型降级，环境配置变化等触发。
八、持久化储存
这里不讨论 memory
8.1 Persistence、
把已经进入 persistence queue / recorder 内存中的待写内容，确保落到持久化存储。
persistence 机制类似：
Session
├─ in-memory conversation history
│
├─ LiveThread
│ │
│ └─ ThreadStore
│ │
│ ├─ LocalThreadStore
│ │ └─ LiveRecorder
│ │
│ └─ SQLite projections / metadata
│
└─ RolloutRecorder
│
└─ rollout-xxx.jsonl
其中 ThreadStore 包括：
append_items：把 raw rollout items 加入 live thread；persist_thread：如果 persistence 是 lazy 的，则 materialize，并写出 queued items；flush_thread：保证之前 queued 的数据已经 durable/readable；shutdown_thread：flush 后关闭 writer
8.2 Rollout
把需要持久化的 Session 事件/状态表示成一系列
RolloutItem，然后序列化成 JSONL。
rollout 记录各种信息，包括重要的 session-level 配置等，非仅仅只有聊天记录
# Rollout 伪代码
import asyncio
from dataclasses import dataclass
from enum import Enum
class CmdType(Enum):
ADD_ITEMS = "add_items"
PERSIST = "persist"
FLUSH = "flush"
SHUTDOWN = "shutdown"
@dataclass
class RolloutCmd:
type: CmdType
items: list | None = None
ack: asyncio.Future | None = None
class RolloutRecorder:
def __init__(self, rollout_path):
self.rollout_path = rollout_path
self._queue = asyncio.Queue()
self._writer = None
self._writer_task = asyncio.create_task(
self._writer_loop()
)
async def record_items(self, items):
"""
Session 有新的 RolloutItem 时调用。
注意：
这里通常不会直接写硬盘，
而是把“写这些 items”的命令发送给后台 writer。
"""
cmd = RolloutCmd(
type=CmdType.ADD_ITEMS,
items=items
)
await self._queue.put(cmd)
async def persist(self):
ack = asyncio.get_running_loop().create_future()
await self._queue.put(
RolloutCmd(
type=CmdType.PERSIST,
ack=ack
)
)
await ack
async def flush(self):
ack = asyncio.get_running_loop().create_future()
await self._queue.put(
RolloutCmd(
type=CmdType.FLUSH,
ack=ack
)
)
await ack
async def shutdown(self):
ack = asyncio.get_running_loop().create_future()
await self._queue.put(
RolloutCmd(
type=CmdType.SHUTDOWN,
ack=ack
)
)
await ack
await self._writer_task
async def _writer_loop(self):
pending_items = []
while True:
cmd = await self._queue.get()
if cmd.type == CmdType.ADD_ITEMS:
items = self._filter_persisted_items(
cmd.items
)
if self._writer is None:
pending_items.extend(items)
else:
self._append_items(items)
elif cmd.type == CmdType.PERSIST:
if self._writer is None:
self._open_rollout_file()
self._append_items(pending_items)
pending_items.clear()
cmd.ack.set_result(True)
elif cmd.type == CmdType.FLUSH:
if self._writer is not None:
self._writer.flush()
cmd.ack.set_result(True)
elif cmd.type == CmdType.SHUTDOWN:
if self._writer is not None:
self._writer.flush()
self._writer.close()
cmd.ack.set_result(True)
break
def _filter_persisted_items(self, items):
return [
item
for item in items
if self._should_persist(item)
]
def _should_persist(self, item):
return True
def _open_rollout_file(self):
self._writer = open(
self.rollout_path,
"a",
encoding="utf-8"
)
def _append_items(self, items):
for item in items:
json_line = serialize_to_json(item)
self._writer.write(json_line + "\n")
rollout 出来的文件类似：
SessionMeta
TurnStarted
UserMessage
TurnContext
ResponseItem(user)
ResponseItem(reasoning)
ResponseItem(tool_call)
ResponseItem(tool_result)
ResponseItem(assistant)
TokenUsage
TurnComplete
TurnStarted
UserMessage
TurnContext
...
例如 SessionMeta：
SessionMeta {
session_id,
id: conversation_id,
cwd,
originator,
cli_version,
source,
model_provider,
base_instructions,
dynamic_tools,
selected_capability_roots,
history_mode,
history_base,
multi_agent_version,
...
}
8.3 Resume
持久化 resume
Session 层 resume
Resume 过程中
- 基本原样保留：正常的
ResponseItem - 会发生明显改变：
Rollback和Compact。如果历史上下文太长，会寻找最优的 compact 节点来继续。 - 也有其他数据不进入 History，但会用于恢复其他状态
| Rollout 内容 | Resume reconstruct 后 |
|---|---|
普通有效 ResponseItem | 基本保留 |
| 被 rollback 的 ResponseItem | 删除/忽略 |
| Compact 之前的旧 history | 被 replacement_history 替换 |
| Compact 后的新 ResponseItem | 继续保留 |
TurnContext | 不直接进入 history， 提取部分恢复状态 |
WorldState | 重新 replay/merge |
| 被 rollback turn 的 WorldState | 忽略 |
SessionMeta | Rollout 中保留，但通常不进入 conversation history |
TurnStarted/Complete | 用于识别 turn 边界，不进入 history |
| TokenUsage / SecurityRisk 等 | 不进入 history |
| 原始 rollout JSONL | 完全不修改 |
九、Streaming 和 Retry
Streaming 使用 WS 失败后，还有 fallback 到 HTTP 的机制。
retry 分为 request retry 和 stream retry。stream retry 也只能进行一次新的 sampling request。
十、其他
需要理解启动时 thread 之间的任务安排，如
async def start_thread_inner(
options,
forked_from_thread_id=None,
):
# 1. 创建 multi-agent 控制器
agent_control = agent_control_for_config(
options.config
)
# 2. 确定 session/thread 的来源
resumed_session_source, resumed_thread_source = (
options.initial_history.get_resumed_session_sources()
or (
self.state.session_source,
None,
)
)
# 3. 如果调用方没有显式指定，就使用推导出的 source
if options.session_source is None:
options.session_source = resumed_session_source
if options.thread_source is None:
options.thread_source = resumed_thread_source
# 4. 包装成真正的 spawn request
request = ThreadSpawnRequest(
options=options,
auth_manager=self.state.auth_manager,
agent_control=agent_control,
)
request.forked_from_thread_id = forked_from_thread_id
# 5. 交给 ThreadManagerState
return await self.state.spawn_thread(request)