Exec事件与JSON输出

codex exec --json 输出的不是 app-server 原始 JSON-RPC。Exec crate 用 EventProcessorWithJsonOutput 把内部 notification 投影为稳定的 ThreadEvent：thread/turn 生命周期、item 开始/更新/完成、错误和 token usage 都有公开形状；原始 item ID 还会被转换成 CLI 自己的 item_N ID。投影是有选择的，不支持的内部 item 不会占用公开 ID。

本文承接非交互Exec CLI和工具框架测试策略，面向已理解 serde、trait 和事件循环的读者。范围是 exec_events.rs、JSONL processor 与 completion backfill，不展开 app-server notification 的生成过程。读完后，你应能解释一个 command item 如何保持 started/completed ID 一致、为什么部分 collab item 不出现、turn failed 为什么不写最后消息文件，以及 warning 为什么是 item.completed 而不是顶层 error。

1. 公开事件

1.1 ThreadEvent

顶层事件通过 serde tag 输出 type 字段。生命周期事件和 item 事件共享同一 JSONL 流，消费者不需要解析 app-server 私有 notification 类型。

源码位置：codex-rs/exec/src/exec_events.rs :: ThreadEvent

rust

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "type")]
pub enum ThreadEvent {
    #[serde(rename = "thread.started")]
    ThreadStarted(ThreadStartedEvent),
    #[serde(rename = "turn.started")]
    TurnStarted(TurnStartedEvent),
    #[serde(rename = "turn.completed")]
    TurnCompleted(TurnCompletedEvent),
    #[serde(rename = "turn.failed")]
    TurnFailed(TurnFailedEvent),
    #[serde(rename = "item.started")]
    ItemStarted(ItemStartedEvent),
    #[serde(rename = "item.updated")]
    ItemUpdated(ItemUpdatedEvent),
    #[serde(rename = "item.completed")]
    ItemCompleted(ItemCompletedEvent),
    #[serde(rename = "error")]
    Error(ThreadErrorEvent),
}

1.2 item详情

ThreadItemDetails 用内部 tag 区分 agent message、reasoning、command execution、MCP、协作、web search、todo 和 item error。事件 ID 与详情类型分离，便于 started/updated/completed 共享 ID。

源码位置：codex-rs/exec/src/exec_events.rs :: ThreadItem

rust

pub struct ThreadItem {
    pub id: String,
    #[serde(flatten)]
    pub details: ThreadItemDetails,
}

#[derive(Serialize, Deserialize, PartialEq)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ThreadItemDetails {
    AgentMessage(AgentMessageItem),
    Reasoning(ReasoningItem),
    CommandExecution(CommandExecutionItem),
    FileChange(FileChangeItem),
    McpToolCall(McpToolCallItem),
    CollabToolCall(CollabToolCallItem),
    WebSearch(WebSearchItem),
    TodoList(TodoListItem),
    Error(ErrorItem),
}

2. ID映射

2.1 生命周期ID

processor 用 raw_to_exec_item_id 保存 app-server 原始 item ID 到公开 ID 的映射。started 分配并保存，completed 取出并删除；如果 completed 没有对应 started，则生成新的公开 ID。

源码位置：codex-rs/exec/src/event_processor_with_jsonl_output.rs :: started_item_id、completed_item_id

rust

fn started_item_id(&mut self, raw_id: &str) -> String {
    if let Some(existing) = self.raw_to_exec_item_id.get(raw_id) {
        return existing.clone();
    }
    let exec_id = self.next_item_id();
    self.raw_to_exec_item_id
        .insert(raw_id.to_string(), exec_id.clone());
    exec_id
}

fn completed_item_id(&mut self, raw_id: &str) -> String {
    self.raw_to_exec_item_id
        .remove(raw_id)
        .unwrap_or_else(|| self.next_item_id())
}

2.2 started范围

agent message 与 reasoning 不产生 started item；command、MCP、patch 等可持续 item 才需要跨事件保持 ID。完成时空 reasoning summary 会被过滤，避免输出无意义事件。

源码位置：codex-rs/exec/src/event_processor_with_jsonl_output.rs :: map_started_item、map_completed_item_mut

rust

fn map_started_item(&mut self, item: ThreadItem) -> Option<ExecThreadItem> {
    match item {
        ThreadItem::AgentMessage { .. } | ThreadItem::Reasoning { .. } => None,
        other => {
            let raw_id = other.id().to_string();
            Self::map_item_with_id(other, || self.started_item_id(&raw_id))
        }
    }
}

2.3 过滤不消耗ID

map_item_with_id 的 ID closure 是惰性调用：只有成功映射的 item 才执行 make_id()。因此 unsupported item、空 reasoning、被过滤的 collab tool 或 interrupted collab completion 不会制造 synthetic ID 空洞。

源码位置：codex-rs/exec/src/event_processor_with_jsonl_output.rs :: map_item_with_id

rust

tool: match tool {
    CollabAgentTool::SendMessage
    | CollabAgentTool::FollowupTask
    | CollabAgentTool::InterruptAgent
    | CollabAgentTool::ListAgents => return None,
    CollabAgentTool::SpawnAgent => CollabTool::SpawnAgent,
    CollabAgentTool::SendInput => CollabTool::SendInput,
    CollabAgentTool::ResumeAgent => CollabTool::Wait,
    CollabAgentTool::Wait => CollabTool::Wait,
    CollabAgentTool::CloseAgent => CollabTool::CloseAgent,
},
status: match status {
    CollabAgentToolCallStatus::InProgress => CollabToolCallStatus::InProgress,
    CollabAgentToolCallStatus::Completed => CollabToolCallStatus::Completed,
    CollabAgentToolCallStatus::Failed => CollabToolCallStatus::Failed,
    CollabAgentToolCallStatus::Interrupted => return None,
},

过滤发生在公开 schema 适配层，不表示内部工具没有执行。尤其 MultiAgent v2 的 send_message、followup_task、interrupt_agent 和 list_agents 没有旧 CollabTool 对应枚举，因此不会强行伪装成其他工具。

3. notification投影

3.1 warning与error

warning、配置警告和弃用通知被包装为 item.completed 的 Error item，保持非致命语义；server Error 则输出顶层 error 并保存为 last_critical_error，供后续 turn failed 使用。

源码位置：codex-rs/exec/src/event_processor_with_jsonl_output.rs :: collect_thread_events

rust

ServerNotification::Warning(notification) => {
    let warning = self.collect_warning(notification.message);
    events.extend(warning.events);
    warning.status
}
ServerNotification::Error(notification) => {
    let message = match notification.error.additional_details {
        Some(details) if !details.is_empty() => {
            format!("{} ({details})", notification.error.message)
        }
        _ => notification.error.message,
    };
    let error = ThreadErrorEvent { message };
    self.last_critical_error = Some(error.clone());
    events.push(ThreadEvent::Error(error));
    CodexStatus::Running
}

Model reroute 不是 fatal stream error，而是一个已完成 Error item，消息包含原模型、目标模型与原因；ModelVerification 当前不产生 JSONL 事件。这样 automation 可以看到模型变化，又不会把正常 reroute 当作 turn failure。

3.2 turn完成

turn completed 先补齐仍未 completed 的 started item，再根据 TurnStatus 选择 completed、failed、interrupted 或继续运行。成功时保存最后 agent message 和 usage；失败时清除 final message，防止把部分答案当成最终答案。

源码位置：codex-rs/exec/src/event_processor_with_jsonl_output.rs :: ServerNotification::TurnCompleted

rust

events.extend(self.reconcile_unfinished_started_items(&notification.turn.items));
match notification.turn.status {
    TurnStatus::Completed => {
        if let Some(final_message) =
            Self::final_message_from_turn_items(notification.turn.items.as_slice())
        {
            self.final_message = Some(final_message);
        }
        self.emit_final_message_on_shutdown = true;
        events.push(ThreadEvent::TurnCompleted(TurnCompletedEvent {
            usage: self.usage_from_last_total(),
        }));
        CodexStatus::InitiateShutdown
    }
    TurnStatus::Failed => {
        self.final_message = None;
        self.emit_final_message_on_shutdown = false;
        let error = notification
            .turn
            .error
            .map(|error| ThreadErrorEvent {
                message: match error.additional_details {
                    Some(details) if !details.is_empty() => {
                        format!("{} ({details})", error.message)
                    }
                    _ => error.message,
                },
            })
            .or_else(|| self.last_critical_error.clone())
            .unwrap_or_else(|| ThreadErrorEvent {
                message: "turn failed".to_string(),
            });
        events.push(ThreadEvent::TurnFailed(TurnFailedEvent { error }));
        CodexStatus::InitiateShutdown
    }
    TurnStatus::Interrupted => {
        self.final_message = None;
        self.emit_final_message_on_shutdown = false;
        CodexStatus::InitiateShutdown
    }
    TurnStatus::InProgress => CodexStatus::Running,
}

成功 final message 优先取 completion items 中最后一个 agent message；没有 agent message 时退回最后一个 Plan text。若 completion items 为空，则保留此前 streamed item 更新得到的 final message。

3.3 Completion回填

in-process channel 在背压时可能丢非终态 item notification，但保证 primary turn/completed。Exec 只对非 ephemeral 且 items view 不完整的 completion 调用 thread/read(include_turns=true)，把目标 turn items 写回 notification，再交给 processor reconcile。

源码位置：codex-rs/exec/src/lib.rs :: maybe_backfill_turn_completed_items、should_backfill_turn_completed_items

rust

fn should_backfill_turn_completed_items(
    thread_ephemeral: bool,
    notification: &ServerNotification,
) -> bool {
    let ServerNotification::TurnCompleted(payload) = notification else {
        return false;
    };

    !thread_ephemeral && payload.turn.items_view != TurnItemsView::Full
}

事件循环先用 should_process_notification 过滤 primary thread/turn，之后才调用 backfill。子 agent 的 turn completion 不会触发 parent thread/read；这避免多代理任务中每个子 completion 都产生一次恢复请求。

4. 输出与文件

4.1 stdout

emit 每次输出一行序列化 ThreadEvent；序列化失败也转成 JSON error 行，避免破坏 JSONL。人类模式使用另一个 processor，不共享 JSONL 的 stdout 契约。

源码位置：codex-rs/exec/src/event_processor_with_jsonl_output.rs :: emit

rust

fn emit(&self, event: ThreadEvent) {
    println!(
        "{}",
        serde_json::to_string(&event).unwrap_or_else(|err| {
            json!({
                "type": "error",
                "message": format!("failed to serialize exec json event: {err}"),
            })
            .to_string()
        })
    );
}

4.2 最后消息文件

只有成功 turn 才设置 emit_final_message_on_shutdown，print_final_output 才写 last_message_file。失败 turn 会清除 final message，因此不会覆盖已有文件。

源码位置：codex-rs/exec/src/event_processor_with_jsonl_output.rs :: print_final_output

rust

fn print_final_output(&mut self) {
    if self.emit_final_message_on_shutdown
        && let Some(path) = self.last_message_path.as_deref()
    {
        handle_last_message(self.final_message.as_deref(), path);
    }
}

5. 验证

5.1 映射矩阵

完整 processor 测试的 30 项断言覆盖 thread/turn 生命周期、command、reasoning、MCP、collab、file change、web search、todo、usage、final message、structured error、model reroute 和 unsupported item ID。它们直接构造 notification，不启动 app-server。

源码位置：codex-rs/exec/tests/event_processor_with_json_output.rs

text

cd codex-rs
cargo test -p codex-exec --test all 'event_processor_with_json_output::' -- --test-threads=1

5.2 warning与meta

JSONL processor 测试断言 warning 变成 item.completed Error item，并断言 MCP result 的 meta 序列化为公开 _meta 字段。它证明内部协议到公开 JSON 的字段投影，不证明 MCP 服务业务结果。

源码位置：codex-rs/exec/src/event_processor_with_jsonl_output_tests.rs :: runtime_warning_emits_a_non_fatal_error_item、mcp_tool_call_result_preserves_meta_in_jsonl_event

text

cd codex-rs
cargo test -p codex-exec --lib event_processor_with_jsonl_output::tests -- --test-threads=1

5.3 失败不覆盖文件

failed_turn_does_not_overwrite_output_last_message_file 先写入已有文件，再处理部分 agent message 和失败 turn，断言 final message 被清空且文件内容不变。这证明失败收尾的文件边界。

源码位置：codex-rs/exec/src/event_processor_with_jsonl_output_tests.rs :: failed_turn_does_not_overwrite_output_last_message_file

text

cd codex-rs
cargo test -p codex-exec --lib event_processor_with_jsonl_output::tests::failed_turn_does_not_overwrite_output_last_message_file -- --test-threads=1

5.4 Primary backfill

ignores_unrelated_turn_completion_before_backfilling_primary_turn 启动 parent/child 多代理流程，让 child 先完成。测试检查 child completion 后没有 typed request，而 primary completion 后恰好出现 thread/read 与 thread/unsubscribe，最终 stdout 包含 parent final message。

源码位置：codex-rs/exec/tests/suite/completion_backfill_tests.rs :: ignores_unrelated_turn_completion_before_backfilling_primary_turn

text

cd codex-rs
cargo test -p codex-exec --test all ignores_unrelated_turn_completion_before_backfilling_primary_turn -- --test-threads=1

测试没有为新过滤的 SendMessage、FollowupTask、InterruptAgent、ListAgents 或 Interrupted collab completion 分别构造用例；这些过滤结论来自当前 map_item_with_id 分支。公开 schema 也不会承诺未来所有内部 item 都必须得到 JSONL 对应项。

6. 源码排查

text

rg -n "enum ThreadEvent|struct ThreadItem|ThreadItemDetails" codex-rs/exec/src/exec_events.rs
rg -n "raw_to_exec_item_id|map_item_with_id|final_message|last_total_token_usage" codex-rs/exec/src/event_processor_with_jsonl_output.rs
rg -n "maybe_backfill_turn_completed_items|should_process_notification" codex-rs/exec/src/lib.rs
rg -n "TurnCompleted|TurnFailed|_meta|last_message" codex-rs/exec/tests codex-rs/exec/src/*tests.rs

本篇的主线是：primary notification 先按需回填，再由 processor 分类并选择性映射 item ID 和详情；turn completed 负责 reconcile、usage 与最终消息；processor 最后把稳定事件逐行写入 stdout，并只在成功收尾时写最后消息文件。

Exec事件与JSON输出 ​

1. 公开事件 ​

1.1 ThreadEvent ​

1.2 item详情 ​

2. ID映射 ​

2.1 生命周期ID ​

2.2 started范围 ​

2.3 过滤不消耗ID ​

3. notification投影 ​

3.1 warning与error ​

3.2 turn完成 ​

3.3 Completion回填 ​

4. 输出与文件 ​

4.1 stdout ​

4.2 最后消息文件 ​

5. 验证 ​

5.1 映射矩阵 ​

5.2 warning与meta ​

5.3 失败不覆盖文件 ​

5.4 Primary backfill ​

6. 源码排查 ​

Exec事件与JSON输出

1. 公开事件

1.1 ThreadEvent

1.2 item详情

2. ID映射

2.1 生命周期ID

2.2 started范围

2.3 过滤不消耗ID

3. notification投影

3.1 warning与error

3.2 turn完成

3.3 Completion回填

4. 输出与文件

4.1 stdout

4.2 最后消息文件

5. 验证

5.1 映射矩阵

5.2 warning与meta

5.3 失败不覆盖文件

5.4 Primary backfill

6. 源码排查