openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-13 11:47:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Restore typed exec JSON output coverage

Browse Source
This commit is contained in:
Eric Traut
2026-03-18 16:29:15 -06:00
parent cf6cc9043d
commit 8dc496ec47
2 changed files with 512 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
codex-rs/exec/src/event_processor.rs
View File

@@ -5,6 +5,7 @@ use codex_protocol::protocol::SessionConfiguredEvent;
use crate::typed_exec_event::TypedExecEvent;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CodexStatus {
Running,
InitiateShutdown,

536
codex-rs/exec/src/event_processor_with_jsonl_output.rs
View File

@@ -69,6 +69,12 @@ struct RunningTodoList {
items: Vec<TodoItem>,
}
#[derive(Debug, PartialEq)]
struct CollectedThreadEvents {
events: Vec<ThreadEvent>,
status: CodexStatus,
}
impl EventProcessorWithJsonOutput {
pub fn new(last_message_path: Option<PathBuf>) -> Self {
Self {
@@ -279,24 +285,18 @@ impl EventProcessorWithJsonOutput {
Some(ExecThreadItem { id, details })
}
}
impl EventProcessor for EventProcessorWithJsonOutput {
fn print_config_summary(
&mut self,
_: &Config,
_: &str,
session_configured: &SessionConfiguredEvent,
) {
self.emit(ThreadEvent::ThreadStarted(ThreadStartedEvent {
fn thread_started_event(session_configured: &SessionConfiguredEvent) -> ThreadEvent {
ThreadEvent::ThreadStarted(ThreadStartedEvent {
thread_id: session_configured.session_id.to_string(),
}));
})
}
fn process_event(&mut self, event: TypedExecEvent) -> CodexStatus {
match event {
fn collect_thread_events(&mut self, event: TypedExecEvent) -> CollectedThreadEvents {
let mut events = Vec::new();
let status = match event {
TypedExecEvent::Warning(message) => {
self.emit(ThreadEvent::ItemCompleted(ItemCompletedEvent {
events.push(ThreadEvent::ItemCompleted(ItemCompletedEvent {
item: ExecThreadItem {
id: self.next_item_id(),
details: ThreadItemDetails::Error(ErrorItem { message }),
@@ -311,7 +311,7 @@ impl EventProcessor for EventProcessorWithJsonOutput {
}
_ => notification.summary,
};
self.emit(ThreadEvent::ItemCompleted(ItemCompletedEvent {
events.push(ThreadEvent::ItemCompleted(ItemCompletedEvent {
item: ExecThreadItem {
id: self.next_item_id(),
details: ThreadItemDetails::Error(ErrorItem { message }),
@@ -328,7 +328,7 @@ impl EventProcessor for EventProcessorWithJsonOutput {
};
let error = ThreadErrorEvent { message };
self.last_critical_error = Some(error.clone());
self.emit(ThreadEvent::Error(error));
events.push(ThreadEvent::Error(error));
CodexStatus::Running
}
TypedExecEvent::DeprecationNotice(notification) => {
@@ -338,7 +338,7 @@ impl EventProcessor for EventProcessorWithJsonOutput {
}
_ => notification.summary,
};
self.emit(ThreadEvent::ItemCompleted(ItemCompletedEvent {
events.push(ThreadEvent::ItemCompleted(ItemCompletedEvent {
item: ExecThreadItem {
id: self.next_item_id(),
details: ThreadItemDetails::Error(ErrorItem { message }),
@@ -351,7 +351,7 @@ impl EventProcessor for EventProcessorWithJsonOutput {
}
TypedExecEvent::ItemStarted(notification) => {
if let Some(item) = Self::map_item(notification.item) {
self.emit(ThreadEvent::ItemStarted(ItemStartedEvent { item }));
events.push(ThreadEvent::ItemStarted(ItemStartedEvent { item }));
}
CodexStatus::Running
}
@@ -362,12 +362,12 @@ impl EventProcessor for EventProcessorWithJsonOutput {
{
self.final_message = Some(text.clone());
}
self.emit(ThreadEvent::ItemCompleted(ItemCompletedEvent { item }));
events.push(ThreadEvent::ItemCompleted(ItemCompletedEvent { item }));
}
CodexStatus::Running
}
TypedExecEvent::ModelRerouted(notification) => {
self.emit(ThreadEvent::ItemCompleted(ItemCompletedEvent {
events.push(ThreadEvent::ItemCompleted(ItemCompletedEvent {
item: ExecThreadItem {
id: self.next_item_id(),
details: ThreadItemDetails::Error(ErrorItem {
@@ -386,7 +386,7 @@ impl EventProcessor for EventProcessorWithJsonOutput {
}
TypedExecEvent::TurnCompleted(notification) => {
if let Some(running) = self.running_todo_list.take() {
self.emit(ThreadEvent::ItemCompleted(ItemCompletedEvent {
events.push(ThreadEvent::ItemCompleted(ItemCompletedEvent {
item: ExecThreadItem {
id: running.item_id,
details: ThreadItemDetails::TodoList(TodoListItem {
@@ -397,7 +397,7 @@ impl EventProcessor for EventProcessorWithJsonOutput {
}
match notification.turn.status {
TurnStatus::Completed => {
self.emit(ThreadEvent::TurnCompleted(TurnCompletedEvent {
events.push(ThreadEvent::TurnCompleted(TurnCompletedEvent {
usage: self.usage_from_last_total(),
}));
CodexStatus::InitiateShutdown
@@ -418,7 +418,7 @@ impl EventProcessor for EventProcessorWithJsonOutput {
.unwrap_or_else(|| ThreadErrorEvent {
message: "turn failed".to_string(),
});
self.emit(ThreadEvent::TurnFailed(TurnFailedEvent { error }));
events.push(ThreadEvent::TurnFailed(TurnFailedEvent { error }));
CodexStatus::InitiateShutdown
}
TurnStatus::Interrupted => CodexStatus::InitiateShutdown,
@@ -431,7 +431,7 @@ impl EventProcessor for EventProcessorWithJsonOutput {
if let Some(running) = self.running_todo_list.as_mut() {
running.items = items.clone();
let item_id = running.item_id.clone();
self.emit(ThreadEvent::ItemUpdated(ItemUpdatedEvent {
events.push(ThreadEvent::ItemUpdated(ItemUpdatedEvent {
item: ExecThreadItem {
id: item_id,
details: ThreadItemDetails::TodoList(TodoListItem { items }),
@@ -443,7 +443,7 @@ impl EventProcessor for EventProcessorWithJsonOutput {
item_id: item_id.clone(),
items: items.clone(),
});
self.emit(ThreadEvent::ItemStarted(ItemStartedEvent {
events.push(ThreadEvent::ItemStarted(ItemStartedEvent {
item: ExecThreadItem {
id: item_id,
details: ThreadItemDetails::TodoList(TodoListItem { items }),
@@ -453,10 +453,31 @@ impl EventProcessor for EventProcessorWithJsonOutput {
CodexStatus::Running
}
TypedExecEvent::TurnStarted => {
self.emit(ThreadEvent::TurnStarted(TurnStartedEvent {}));
events.push(ThreadEvent::TurnStarted(TurnStartedEvent {}));
CodexStatus::Running
}
};
CollectedThreadEvents { events, status }
}
}
impl EventProcessor for EventProcessorWithJsonOutput {
fn print_config_summary(
&mut self,
_: &Config,
_: &str,
session_configured: &SessionConfiguredEvent,
) {
self.emit(Self::thread_started_event(session_configured));
}
fn process_event(&mut self, event: TypedExecEvent) -> CodexStatus {
let collected = self.collect_thread_events(event);
for event in collected.events {
self.emit(event);
}
collected.status
}
fn print_final_output(&mut self) {
@@ -468,11 +489,49 @@ impl EventProcessor for EventProcessorWithJsonOutput {
#[cfg(test)]
mod tests {
use super::CollectedThreadEvents;
use super::EventProcessorWithJsonOutput;
use crate::event_processor::CodexStatus;
use crate::exec_events::CommandExecutionItem;
use crate::exec_events::CommandExecutionStatus;
use crate::exec_events::ItemCompletedEvent;
use crate::exec_events::ItemStartedEvent;
use crate::exec_events::ItemUpdatedEvent;
use crate::exec_events::ThreadErrorEvent;
use crate::exec_events::ThreadEvent;
use crate::exec_events::ThreadItem as ExecThreadItem;
use crate::exec_events::ThreadItemDetails;
use crate::exec_events::ThreadStartedEvent;
use crate::exec_events::TodoItem;
use crate::exec_events::TodoListItem;
use crate::exec_events::TurnCompletedEvent;
use crate::exec_events::TurnStartedEvent;
use crate::exec_events::Usage;
use crate::exec_events::WebSearchItem;
use crate::typed_exec_event::TypedExecEvent;
use codex_app_server_protocol::CommandAction;
use codex_app_server_protocol::CommandExecutionStatus as ApiCommandExecutionStatus;
use codex_app_server_protocol::ErrorNotification;
use codex_app_server_protocol::ItemCompletedNotification;
use codex_app_server_protocol::ItemStartedNotification;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadTokenUsage;
use codex_app_server_protocol::TokenUsageBreakdown;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnError;
use codex_app_server_protocol::TurnPlanStep;
use codex_app_server_protocol::TurnPlanStepStatus;
use codex_app_server_protocol::TurnPlanUpdatedNotification;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::WebSearchAction as ApiWebSearchAction;
use codex_protocol::ThreadId;
use codex_protocol::models::WebSearchAction;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionConfiguredEvent;
use pretty_assertions::assert_eq;
use std::path::PathBuf;
#[test]
fn map_todo_items_preserves_text_and_completion_state() {
@@ -501,4 +560,431 @@ mod tests {
]
);
}
#[test]
fn session_configured_produces_thread_started_event() {
let session_configured = SessionConfiguredEvent {
session_id: ThreadId::from_string("67e55044-10b1-426f-9247-bb680e5fe0c8")
.expect("thread id should parse"),
forked_from_id: None,
thread_name: None,
model: "codex-mini-latest".to_string(),
model_provider_id: "test-provider".to_string(),
service_tier: None,
approval_policy: AskForApproval::Never,
approvals_reviewer: codex_protocol::config_types::ApprovalsReviewer::User,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
cwd: PathBuf::from("/tmp/project"),
reasoning_effort: None,
history_log_id: 0,
history_entry_count: 0,
initial_messages: None,
network_proxy: None,
rollout_path: None,
};
assert_eq!(
EventProcessorWithJsonOutput::thread_started_event(&session_configured),
ThreadEvent::ThreadStarted(ThreadStartedEvent {
thread_id: "67e55044-10b1-426f-9247-bb680e5fe0c8".to_string(),
})
);
}
#[test]
fn turn_started_emits_turn_started_event() {
let mut processor = EventProcessorWithJsonOutput::new(None);
let collected = processor.collect_thread_events(TypedExecEvent::TurnStarted);
assert_eq!(
collected,
CollectedThreadEvents {
events: vec![ThreadEvent::TurnStarted(TurnStartedEvent {})],
status: CodexStatus::Running,
}
);
}
#[test]
fn command_execution_started_and_completed_translate_to_thread_events() {
let mut processor = EventProcessorWithJsonOutput::new(None);
let command_item = ThreadItem::CommandExecution {
id: "cmd-1".to_string(),
command: "ls".to_string(),
cwd: PathBuf::from("/tmp/project"),
process_id: Some("123".to_string()),
status: ApiCommandExecutionStatus::InProgress,
command_actions: Vec::<CommandAction>::new(),
aggregated_output: None,
exit_code: None,
duration_ms: None,
};
let started =
processor.collect_thread_events(TypedExecEvent::ItemStarted(ItemStartedNotification {
item: command_item,
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
}));
assert_eq!(
started,
CollectedThreadEvents {
events: vec![ThreadEvent::ItemStarted(ItemStartedEvent {
item: ExecThreadItem {
id: "cmd-1".to_string(),
details: ThreadItemDetails::CommandExecution(CommandExecutionItem {
command: "ls".to_string(),
aggregated_output: String::new(),
exit_code: None,
status: CommandExecutionStatus::InProgress,
}),
},
})],
status: CodexStatus::Running,
}
);
let completed = processor.collect_thread_events(TypedExecEvent::ItemCompleted(
ItemCompletedNotification {
item: ThreadItem::CommandExecution {
id: "cmd-1".to_string(),
command: "ls".to_string(),
cwd: PathBuf::from("/tmp/project"),
process_id: Some("123".to_string()),
status: ApiCommandExecutionStatus::Completed,
command_actions: Vec::<CommandAction>::new(),
aggregated_output: Some("a.txt\n".to_string()),
exit_code: Some(0),
duration_ms: Some(3),
},
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
},
));
assert_eq!(
completed,
CollectedThreadEvents {
events: vec![ThreadEvent::ItemCompleted(ItemCompletedEvent {
item: ExecThreadItem {
id: "cmd-1".to_string(),
details: ThreadItemDetails::CommandExecution(CommandExecutionItem {
command: "ls".to_string(),
aggregated_output: "a.txt\n".to_string(),
exit_code: Some(0),
status: CommandExecutionStatus::Completed,
}),
},
})],
status: CodexStatus::Running,
}
);
}
#[test]
fn web_search_completion_preserves_query_and_action() {
let mut processor = EventProcessorWithJsonOutput::new(None);
let collected = processor.collect_thread_events(TypedExecEvent::ItemCompleted(
ItemCompletedNotification {
item: ThreadItem::WebSearch {
id: "search-1".to_string(),
query: "rust async await".to_string(),
action: Some(ApiWebSearchAction::Search {
query: Some("rust async await".to_string()),
queries: None,
}),
},
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
},
));
assert_eq!(
collected,
CollectedThreadEvents {
events: vec![ThreadEvent::ItemCompleted(ItemCompletedEvent {
item: ExecThreadItem {
id: "search-1".to_string(),
details: ThreadItemDetails::WebSearch(WebSearchItem {
id: "search-1".to_string(),
query: "rust async await".to_string(),
action: WebSearchAction::Search {
query: Some("rust async await".to_string()),
queries: None,
},
}),
},
})],
status: CodexStatus::Running,
}
);
}
#[test]
fn plan_update_emits_started_then_updated_then_completed() {
let mut processor = EventProcessorWithJsonOutput::new(None);
let started = processor.collect_thread_events(TypedExecEvent::TurnPlanUpdated(
TurnPlanUpdatedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
explanation: None,
plan: vec![
TurnPlanStep {
step: "step one".to_string(),
status: TurnPlanStepStatus::Pending,
},
TurnPlanStep {
step: "step two".to_string(),
status: TurnPlanStepStatus::InProgress,
},
],
},
));
assert_eq!(
started,
CollectedThreadEvents {
events: vec![ThreadEvent::ItemStarted(ItemStartedEvent {
item: ExecThreadItem {
id: "item_0".to_string(),
details: ThreadItemDetails::TodoList(TodoListItem {
items: vec![
TodoItem {
text: "step one".to_string(),
completed: false,
},
TodoItem {
text: "step two".to_string(),
completed: false,
},
],
}),
},
})],
status: CodexStatus::Running,
}
);
let updated = processor.collect_thread_events(TypedExecEvent::TurnPlanUpdated(
TurnPlanUpdatedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
explanation: None,
plan: vec![
TurnPlanStep {
step: "step one".to_string(),
status: TurnPlanStepStatus::Completed,
},
TurnPlanStep {
step: "step two".to_string(),
status: TurnPlanStepStatus::InProgress,
},
],
},
));
assert_eq!(
updated,
CollectedThreadEvents {
events: vec![ThreadEvent::ItemUpdated(ItemUpdatedEvent {
item: ExecThreadItem {
id: "item_0".to_string(),
details: ThreadItemDetails::TodoList(TodoListItem {
items: vec![
TodoItem {
text: "step one".to_string(),
completed: true,
},
TodoItem {
text: "step two".to_string(),
completed: false,
},
],
}),
},
})],
status: CodexStatus::Running,
}
);
let completed = processor.collect_thread_events(TypedExecEvent::TurnCompleted(
TurnCompletedNotification {
thread_id: "thread-1".to_string(),
turn: Turn {
id: "turn-1".to_string(),
items: Vec::new(),
status: TurnStatus::Completed,
error: None,
},
},
));
assert_eq!(
completed,
CollectedThreadEvents {
events: vec![
ThreadEvent::ItemCompleted(ItemCompletedEvent {
item: ExecThreadItem {
id: "item_0".to_string(),
details: ThreadItemDetails::TodoList(TodoListItem {
items: vec![
TodoItem {
text: "step one".to_string(),
completed: true,
},
TodoItem {
text: "step two".to_string(),
completed: false,
},
],
}),
},
}),
ThreadEvent::TurnCompleted(TurnCompletedEvent {
usage: Usage::default(),
}),
],
status: CodexStatus::InitiateShutdown,
}
);
}
#[test]
fn token_usage_update_is_emitted_on_turn_completion() {
let mut processor = EventProcessorWithJsonOutput::new(None);
let usage_update =
processor.collect_thread_events(TypedExecEvent::ThreadTokenUsageUpdated(
codex_app_server_protocol::ThreadTokenUsageUpdatedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
token_usage: ThreadTokenUsage {
total: TokenUsageBreakdown {
total_tokens: 42,
input_tokens: 10,
cached_input_tokens: 3,
output_tokens: 29,
reasoning_output_tokens: 7,
},
last: TokenUsageBreakdown {
total_tokens: 42,
input_tokens: 10,
cached_input_tokens: 3,
output_tokens: 29,
reasoning_output_tokens: 7,
},
model_context_window: Some(128_000),
},
},
));
assert_eq!(
usage_update,
CollectedThreadEvents {
events: Vec::new(),
status: CodexStatus::Running,
}
);
let completed = processor.collect_thread_events(TypedExecEvent::TurnCompleted(
TurnCompletedNotification {
thread_id: "thread-1".to_string(),
turn: Turn {
id: "turn-1".to_string(),
items: Vec::new(),
status: TurnStatus::Completed,
error: None,
},
},
));
assert_eq!(
completed,
CollectedThreadEvents {
events: vec![ThreadEvent::TurnCompleted(TurnCompletedEvent {
usage: Usage {
input_tokens: 10,
cached_input_tokens: 3,
output_tokens: 29,
},
})],
status: CodexStatus::InitiateShutdown,
}
);
}
#[test]
fn turn_failure_prefers_structured_error_message() {
let mut processor = EventProcessorWithJsonOutput::new(None);
let error = processor.collect_thread_events(TypedExecEvent::Error(ErrorNotification {
error: TurnError {
message: "backend failed".to_string(),
codex_error_info: None,
additional_details: Some("request id abc".to_string()),
},
will_retry: false,
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
}));
assert_eq!(
error,
CollectedThreadEvents {
events: vec![ThreadEvent::Error(ThreadErrorEvent {
message: "backend failed (request id abc)".to_string(),
})],
status: CodexStatus::Running,
}
);
let failed = processor.collect_thread_events(TypedExecEvent::TurnCompleted(
TurnCompletedNotification {
thread_id: "thread-1".to_string(),
turn: Turn {
id: "turn-1".to_string(),
items: Vec::new(),
status: TurnStatus::Failed,
error: None,
},
},
));
assert_eq!(
failed,
CollectedThreadEvents {
events: vec![ThreadEvent::TurnFailed(
crate::exec_events::TurnFailedEvent {
error: ThreadErrorEvent {
message: "backend failed (request id abc)".to_string(),
},
}
)],
status: CodexStatus::InitiateShutdown,
}
);
}
#[test]
fn model_reroute_surfaces_as_error_item() {
let mut processor = EventProcessorWithJsonOutput::new(None);
let collected = processor.collect_thread_events(TypedExecEvent::ModelRerouted(
codex_app_server_protocol::ModelReroutedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
from_model: "gpt-5".to_string(),
to_model: "gpt-5-mini".to_string(),
reason: codex_app_server_protocol::ModelRerouteReason::HighRiskCyberActivity,
},
));
assert_eq!(collected.status, CodexStatus::Running);
assert_eq!(collected.events.len(), 1);
let ThreadEvent::ItemCompleted(ItemCompletedEvent { item }) = &collected.events[0] else {
panic!("expected ItemCompleted");
};
assert_eq!(item.id, "item_0");
assert_eq!(
item.details,
ThreadItemDetails::Error(crate::exec_events::ErrorItem {
message: "model rerouted: gpt-5 -> gpt-5-mini (HighRiskCyberActivity)".to_string(),
})
);
}
}