diff --git a/codex-rs/exec/src/event_processor.rs b/codex-rs/exec/src/event_processor.rs index 87505b0160..16c2c7a9b2 100644 --- a/codex-rs/exec/src/event_processor.rs +++ b/codex-rs/exec/src/event_processor.rs @@ -1,10 +1,9 @@ use std::path::Path; +use codex_app_server_protocol::ServerNotification; use codex_core::config::Config; use codex_protocol::protocol::SessionConfiguredEvent; -use crate::typed_exec_event::TypedExecEvent; - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum CodexStatus { Running, @@ -20,8 +19,11 @@ pub(crate) trait EventProcessor { session_configured: &SessionConfiguredEvent, ); - /// Handle a single event emitted by the agent. - fn process_event(&mut self, event: TypedExecEvent) -> CodexStatus; + /// Handle a single typed app-server notification emitted by the agent. + fn process_server_notification(&mut self, notification: ServerNotification) -> CodexStatus; + + /// Handle a local exec warning that is not represented as an app-server notification. + fn process_warning(&mut self, message: String) -> CodexStatus; fn print_final_output(&mut self) {} } diff --git a/codex-rs/exec/src/event_processor_with_human_output.rs b/codex-rs/exec/src/event_processor_with_human_output.rs index 83628bab4d..fed86b2b13 100644 --- a/codex-rs/exec/src/event_processor_with_human_output.rs +++ b/codex-rs/exec/src/event_processor_with_human_output.rs @@ -4,6 +4,7 @@ use std::path::PathBuf; use codex_app_server_protocol::CommandExecutionStatus; use codex_app_server_protocol::McpToolCallStatus; use codex_app_server_protocol::PatchApplyStatus; +use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadTokenUsage; use codex_app_server_protocol::TurnStatus; @@ -16,7 +17,6 @@ use owo_colors::Style; use crate::event_processor::CodexStatus; use crate::event_processor::EventProcessor; use crate::event_processor::handle_last_message; -use crate::typed_exec_event::TypedExecEvent; pub(crate) struct EventProcessorWithHumanOutput { bold: Style, @@ -232,16 +232,9 @@ impl EventProcessor for EventProcessorWithHumanOutput { eprintln!("{}\n{}", "user".style(self.cyan), prompt); } - fn process_event(&mut self, event: TypedExecEvent) -> CodexStatus { - match event { - TypedExecEvent::Warning(message) => { - eprintln!( - "{} {message}", - "warning:".style(self.yellow).style(self.bold) - ); - CodexStatus::Running - } - TypedExecEvent::ConfigWarning(notification) => { + fn process_server_notification(&mut self, notification: ServerNotification) -> CodexStatus { + match notification { + ServerNotification::ConfigWarning(notification) => { let details = notification .details .map(|details| format!(" ({details})")) @@ -254,7 +247,7 @@ impl EventProcessor for EventProcessorWithHumanOutput { ); CodexStatus::Running } - TypedExecEvent::Error(notification) => { + ServerNotification::Error(notification) => { eprintln!( "{} {}", "ERROR:".style(self.red).style(self.bold), @@ -262,7 +255,7 @@ impl EventProcessor for EventProcessorWithHumanOutput { ); CodexStatus::Running } - TypedExecEvent::DeprecationNotice(notification) => { + ServerNotification::DeprecationNotice(notification) => { eprintln!( "{} {}", "deprecated:".style(self.yellow).style(self.bold), @@ -273,7 +266,7 @@ impl EventProcessor for EventProcessorWithHumanOutput { } CodexStatus::Running } - TypedExecEvent::HookStarted(notification) => { + ServerNotification::HookStarted(notification) => { eprintln!( "{} {}", "hook:".style(self.bold), @@ -281,7 +274,7 @@ impl EventProcessor for EventProcessorWithHumanOutput { ); CodexStatus::Running } - TypedExecEvent::HookCompleted(notification) => { + ServerNotification::HookCompleted(notification) => { eprintln!( "{} {} {:?}", "hook:".style(self.bold), @@ -290,15 +283,15 @@ impl EventProcessor for EventProcessorWithHumanOutput { ); CodexStatus::Running } - TypedExecEvent::ItemStarted(notification) => { + ServerNotification::ItemStarted(notification) => { self.render_item_started(¬ification.item); CodexStatus::Running } - TypedExecEvent::ItemCompleted(notification) => { + ServerNotification::ItemCompleted(notification) => { self.render_item_completed(notification.item); CodexStatus::Running } - TypedExecEvent::ModelRerouted(notification) => { + ServerNotification::ModelRerouted(notification) => { eprintln!( "{} {} -> {}", "model rerouted:".style(self.yellow).style(self.bold), @@ -307,11 +300,11 @@ impl EventProcessor for EventProcessorWithHumanOutput { ); CodexStatus::Running } - TypedExecEvent::ThreadTokenUsageUpdated(notification) => { + ServerNotification::ThreadTokenUsageUpdated(notification) => { self.last_total_token_usage = Some(notification.token_usage); CodexStatus::Running } - TypedExecEvent::TurnCompleted(notification) => match notification.turn.status { + ServerNotification::TurnCompleted(notification) => match notification.turn.status { TurnStatus::Completed => { if let Some(final_message) = final_message_from_turn_items(notification.turn.items.as_slice()) @@ -333,13 +326,13 @@ impl EventProcessor for EventProcessorWithHumanOutput { } TurnStatus::InProgress => CodexStatus::Running, }, - TypedExecEvent::TurnDiffUpdated(notification) => { + ServerNotification::TurnDiffUpdated(notification) => { if !notification.diff.trim().is_empty() { eprintln!("{}", notification.diff); } CodexStatus::Running } - TypedExecEvent::TurnPlanUpdated(notification) => { + ServerNotification::TurnPlanUpdated(notification) => { if let Some(explanation) = notification.explanation { eprintln!("{}", explanation.style(self.dimmed)); } @@ -348,10 +341,19 @@ impl EventProcessor for EventProcessorWithHumanOutput { } CodexStatus::Running } - TypedExecEvent::TurnStarted => CodexStatus::Running, + ServerNotification::TurnStarted(_) => CodexStatus::Running, + _ => CodexStatus::Running, } } + fn process_warning(&mut self, message: String) -> CodexStatus { + eprintln!( + "{} {message}", + "warning:".style(self.yellow).style(self.bold) + ); + CodexStatus::Running + } + fn print_final_output(&mut self) { if let Some(path) = self.last_message_path.as_deref() { handle_last_message(self.final_message.as_deref(), path); @@ -422,7 +424,7 @@ mod tests { use super::reasoning_text; use super::should_print_final_message_to_stdout; use crate::event_processor::EventProcessor; - use crate::typed_exec_event::TypedExecEvent; + use codex_app_server_protocol::ServerNotification; #[test] fn suppresses_final_stdout_message_when_both_streams_are_terminals() { @@ -539,7 +541,7 @@ mod tests { last_total_token_usage: None, }; - let status = processor.process_event(TypedExecEvent::TurnCompleted( + let status = processor.process_server_notification(ServerNotification::TurnCompleted( codex_app_server_protocol::TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { @@ -579,7 +581,7 @@ mod tests { last_total_token_usage: None, }; - let status = processor.process_event(TypedExecEvent::TurnCompleted( + let status = processor.process_server_notification(ServerNotification::TurnCompleted( codex_app_server_protocol::TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { @@ -619,7 +621,7 @@ mod tests { last_total_token_usage: None, }; - let status = processor.process_event(TypedExecEvent::TurnCompleted( + let status = processor.process_server_notification(ServerNotification::TurnCompleted( codex_app_server_protocol::TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { diff --git a/codex-rs/exec/src/event_processor_with_jsonl_output.rs b/codex-rs/exec/src/event_processor_with_jsonl_output.rs index 4097fb4097..4adc6db5d4 100644 --- a/codex-rs/exec/src/event_processor_with_jsonl_output.rs +++ b/codex-rs/exec/src/event_processor_with_jsonl_output.rs @@ -8,6 +8,7 @@ use codex_app_server_protocol::CommandExecutionStatus; use codex_app_server_protocol::McpToolCallStatus; use codex_app_server_protocol::PatchApplyStatus; use codex_app_server_protocol::PatchChangeKind; +use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadTokenUsage; use codex_app_server_protocol::TurnStatus; @@ -16,7 +17,7 @@ use codex_protocol::models::WebSearchAction; use codex_protocol::protocol::SessionConfiguredEvent; use serde_json::json; -use crate::event_processor::CodexStatus; +pub use crate::event_processor::CodexStatus; use crate::event_processor::EventProcessor; use crate::event_processor::handle_last_message; use crate::exec_events::AgentMessageItem; @@ -52,7 +53,6 @@ use crate::exec_events::TurnFailedEvent; use crate::exec_events::TurnStartedEvent; use crate::exec_events::Usage; use crate::exec_events::WebSearchItem; -use crate::typed_exec_event::TypedExecEvent; pub struct EventProcessorWithJsonOutput { last_message_path: Option, @@ -70,9 +70,9 @@ struct RunningTodoList { } #[derive(Debug, PartialEq)] -struct CollectedThreadEvents { - events: Vec, - status: CodexStatus, +pub struct CollectedThreadEvents { + pub events: Vec, + pub status: CodexStatus, } impl EventProcessorWithJsonOutput { @@ -116,7 +116,7 @@ impl EventProcessorWithJsonOutput { } } - fn map_todo_items(plan: &[codex_app_server_protocol::TurnPlanStep]) -> Vec { + pub fn map_todo_items(plan: &[codex_app_server_protocol::TurnPlanStep]) -> Vec { plan.iter() .map(|step| TodoItem { text: step.step.clone(), @@ -295,25 +295,31 @@ impl EventProcessorWithJsonOutput { }) } - fn thread_started_event(session_configured: &SessionConfiguredEvent) -> ThreadEvent { + pub fn thread_started_event(session_configured: &SessionConfiguredEvent) -> ThreadEvent { ThreadEvent::ThreadStarted(ThreadStartedEvent { thread_id: session_configured.session_id.to_string(), }) } - fn collect_thread_events(&mut self, event: TypedExecEvent) -> CollectedThreadEvents { + pub fn collect_warning(&mut self, message: String) -> CollectedThreadEvents { + CollectedThreadEvents { + events: vec![ThreadEvent::ItemCompleted(ItemCompletedEvent { + item: ExecThreadItem { + id: self.next_item_id(), + details: ThreadItemDetails::Error(ErrorItem { message }), + }, + })], + status: CodexStatus::Running, + } + } + + pub fn collect_thread_events( + &mut self, + notification: ServerNotification, + ) -> CollectedThreadEvents { let mut events = Vec::new(); - let status = match event { - TypedExecEvent::Warning(message) => { - events.push(ThreadEvent::ItemCompleted(ItemCompletedEvent { - item: ExecThreadItem { - id: self.next_item_id(), - details: ThreadItemDetails::Error(ErrorItem { message }), - }, - })); - CodexStatus::Running - } - TypedExecEvent::ConfigWarning(notification) => { + let status = match notification { + ServerNotification::ConfigWarning(notification) => { let message = match notification.details { Some(details) if !details.is_empty() => { format!("{} ({details})", notification.summary) @@ -328,7 +334,7 @@ impl EventProcessorWithJsonOutput { })); CodexStatus::Running } - TypedExecEvent::Error(notification) => { + ServerNotification::Error(notification) => { let message = match notification.error.additional_details { Some(details) if !details.is_empty() => { format!("{} ({details})", notification.error.message) @@ -340,7 +346,7 @@ impl EventProcessorWithJsonOutput { events.push(ThreadEvent::Error(error)); CodexStatus::Running } - TypedExecEvent::DeprecationNotice(notification) => { + ServerNotification::DeprecationNotice(notification) => { let message = match notification.details { Some(details) if !details.is_empty() => { format!("{} ({details})", notification.summary) @@ -355,16 +361,16 @@ impl EventProcessorWithJsonOutput { })); CodexStatus::Running } - TypedExecEvent::HookStarted(_) | TypedExecEvent::HookCompleted(_) => { + ServerNotification::HookStarted(_) | ServerNotification::HookCompleted(_) => { CodexStatus::Running } - TypedExecEvent::ItemStarted(notification) => { + ServerNotification::ItemStarted(notification) => { if let Some(item) = Self::map_item(notification.item) { events.push(ThreadEvent::ItemStarted(ItemStartedEvent { item })); } CodexStatus::Running } - TypedExecEvent::ItemCompleted(notification) => { + ServerNotification::ItemCompleted(notification) => { if let Some(item) = Self::map_item(notification.item) { if let ThreadItemDetails::AgentMessage(AgentMessageItem { text }) = &item.details @@ -375,7 +381,7 @@ impl EventProcessorWithJsonOutput { } CodexStatus::Running } - TypedExecEvent::ModelRerouted(notification) => { + ServerNotification::ModelRerouted(notification) => { events.push(ThreadEvent::ItemCompleted(ItemCompletedEvent { item: ExecThreadItem { id: self.next_item_id(), @@ -389,11 +395,11 @@ impl EventProcessorWithJsonOutput { })); CodexStatus::Running } - TypedExecEvent::ThreadTokenUsageUpdated(notification) => { + ServerNotification::ThreadTokenUsageUpdated(notification) => { self.last_total_token_usage = Some(notification.token_usage); CodexStatus::Running } - TypedExecEvent::TurnCompleted(notification) => { + ServerNotification::TurnCompleted(notification) => { if let Some(running) = self.running_todo_list.take() { events.push(ThreadEvent::ItemCompleted(ItemCompletedEvent { item: ExecThreadItem { @@ -439,8 +445,8 @@ impl EventProcessorWithJsonOutput { TurnStatus::InProgress => CodexStatus::Running, } } - TypedExecEvent::TurnDiffUpdated(_) => CodexStatus::Running, - TypedExecEvent::TurnPlanUpdated(notification) => { + ServerNotification::TurnDiffUpdated(_) => CodexStatus::Running, + ServerNotification::TurnPlanUpdated(notification) => { let items = Self::map_todo_items(¬ification.plan); if let Some(running) = self.running_todo_list.as_mut() { running.items = items.clone(); @@ -466,10 +472,11 @@ impl EventProcessorWithJsonOutput { } CodexStatus::Running } - TypedExecEvent::TurnStarted => { + ServerNotification::TurnStarted(_) => { events.push(ThreadEvent::TurnStarted(TurnStartedEvent {})); CodexStatus::Running } + _ => CodexStatus::Running, }; CollectedThreadEvents { events, status } @@ -486,8 +493,16 @@ impl EventProcessor for EventProcessorWithJsonOutput { self.emit(Self::thread_started_event(session_configured)); } - fn process_event(&mut self, event: TypedExecEvent) -> CodexStatus { - let collected = self.collect_thread_events(event); + fn process_server_notification(&mut self, notification: ServerNotification) -> CodexStatus { + let collected = self.collect_thread_events(notification); + for event in collected.events { + self.emit(event); + } + collected.status + } + + fn process_warning(&mut self, message: String) -> CodexStatus { + let collected = self.collect_warning(message); for event in collected.events { self.emit(event); } @@ -500,7 +515,3 @@ impl EventProcessor for EventProcessorWithJsonOutput { } } } - -#[cfg(test)] -#[path = "../tests/event_processor_with_json_output.rs"] -mod tests; diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 0b44b79c19..89612dab0b 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -9,7 +9,6 @@ mod event_processor; mod event_processor_with_human_output; pub mod event_processor_with_jsonl_output; pub mod exec_events; -mod typed_exec_event; pub use cli::Cli; pub use cli::Command; @@ -101,8 +100,6 @@ use uuid::Uuid; use crate::cli::Command as ExecCommand; use crate::event_processor::CodexStatus; use crate::event_processor::EventProcessor; -use crate::typed_exec_event::TypedExecEvent; -use crate::typed_exec_event::session_configured_from_thread_response; use codex_core::default_client::set_default_client_residency_requirement; use codex_core::default_client::set_default_originator; @@ -661,7 +658,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { // is using. event_processor.print_config_summary(&config, &prompt_summary, &session_configured); if !json_mode && let Some(message) = codex_core::config::missing_system_bwrap_warning() { - event_processor.process_event(TypedExecEvent::Warning(message)); + event_processor.process_warning(message); } info!("Codex initialized with event: {session_configured:?}"); @@ -794,9 +791,8 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { ¬ification, &primary_thread_id_for_requests, &task_id, - ) && let Some(event) = TypedExecEvent::from_server_notification(notification) - { - match event_processor.process_event(event) { + ) { + match event_processor.process_server_notification(notification) { CodexStatus::Running => {} CodexStatus::InitiateShutdown => { if let Err(err) = request_shutdown( @@ -816,7 +812,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { InProcessServerEvent::Lagged { skipped } => { let message = lagged_event_warning_message(skipped); warn!("{message}"); - event_processor.process_event(TypedExecEvent::Warning(message)); + event_processor.process_warning(message); } } } @@ -952,6 +948,46 @@ fn review_target_to_api(target: ReviewTarget) -> ApiReviewTarget { } } +#[expect( + clippy::too_many_arguments, + reason = "session mapping keeps explicit fields" +)] +fn session_configured_from_thread_response( + thread_id: &str, + thread_name: Option, + rollout_path: Option, + model: String, + model_provider_id: String, + service_tier: Option, + approval_policy: AskForApproval, + approvals_reviewer: codex_protocol::config_types::ApprovalsReviewer, + sandbox_policy: SandboxPolicy, + cwd: PathBuf, + reasoning_effort: Option, +) -> Result { + let session_id = codex_protocol::ThreadId::from_string(thread_id) + .map_err(|err| format!("thread id `{thread_id}` is invalid: {err}"))?; + + Ok(SessionConfiguredEvent { + session_id, + forked_from_id: None, + thread_name, + model, + model_provider_id, + service_tier, + approval_policy, + approvals_reviewer, + sandbox_policy, + cwd, + reasoning_effort, + history_log_id: 0, + history_entry_count: 0, + initial_messages: None, + network_proxy: None, + rollout_path, + }) +} + fn lagged_event_warning_message(skipped: usize) -> String { format!("in-process app-server event stream lagged; dropped {skipped} events") } diff --git a/codex-rs/exec/src/typed_exec_event.rs b/codex-rs/exec/src/typed_exec_event.rs deleted file mode 100644 index a4e95bd1ae..0000000000 --- a/codex-rs/exec/src/typed_exec_event.rs +++ /dev/null @@ -1,124 +0,0 @@ -use codex_app_server_protocol::ConfigWarningNotification; -use codex_app_server_protocol::DeprecationNoticeNotification; -use codex_app_server_protocol::ErrorNotification; -use codex_app_server_protocol::HookCompletedNotification; -use codex_app_server_protocol::HookStartedNotification; -use codex_app_server_protocol::ItemCompletedNotification; -use codex_app_server_protocol::ItemStartedNotification; -use codex_app_server_protocol::ModelReroutedNotification; -use codex_app_server_protocol::ServerNotification; -use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification; -use codex_app_server_protocol::TurnCompletedNotification; -use codex_app_server_protocol::TurnDiffUpdatedNotification; -use codex_app_server_protocol::TurnPlanUpdatedNotification; -use codex_protocol::ThreadId; -use codex_protocol::config_types::ApprovalsReviewer; -use codex_protocol::config_types::ServiceTier; -use codex_protocol::openai_models::ReasoningEffort; -use codex_protocol::protocol::AskForApproval; -use codex_protocol::protocol::SandboxPolicy; -use codex_protocol::protocol::SessionConfiguredEvent; -use std::path::PathBuf; - -#[derive(Debug, Clone)] -pub(crate) enum TypedExecEvent { - Warning(String), - ConfigWarning(ConfigWarningNotification), - Error(ErrorNotification), - DeprecationNotice(DeprecationNoticeNotification), - HookStarted(HookStartedNotification), - HookCompleted(HookCompletedNotification), - ItemStarted(ItemStartedNotification), - ItemCompleted(ItemCompletedNotification), - ModelRerouted(ModelReroutedNotification), - ThreadTokenUsageUpdated(ThreadTokenUsageUpdatedNotification), - TurnCompleted(TurnCompletedNotification), - TurnDiffUpdated(TurnDiffUpdatedNotification), - TurnPlanUpdated(TurnPlanUpdatedNotification), - TurnStarted, -} - -impl TypedExecEvent { - pub(crate) fn from_server_notification( - notification: ServerNotification, - ) -> Option { - match notification { - ServerNotification::ConfigWarning(notification) => { - Some(TypedExecEvent::ConfigWarning(notification)) - } - ServerNotification::DeprecationNotice(notification) => { - Some(TypedExecEvent::DeprecationNotice(notification)) - } - ServerNotification::Error(notification) => Some(TypedExecEvent::Error(notification)), - ServerNotification::HookCompleted(notification) => { - Some(TypedExecEvent::HookCompleted(notification)) - } - ServerNotification::HookStarted(notification) => { - Some(TypedExecEvent::HookStarted(notification)) - } - ServerNotification::ItemCompleted(notification) => { - Some(TypedExecEvent::ItemCompleted(notification)) - } - ServerNotification::ItemStarted(notification) => { - Some(TypedExecEvent::ItemStarted(notification)) - } - ServerNotification::ModelRerouted(notification) => { - Some(TypedExecEvent::ModelRerouted(notification)) - } - ServerNotification::ThreadTokenUsageUpdated(notification) => { - Some(TypedExecEvent::ThreadTokenUsageUpdated(notification)) - } - ServerNotification::TurnCompleted(notification) => { - Some(TypedExecEvent::TurnCompleted(notification)) - } - ServerNotification::TurnDiffUpdated(notification) => { - Some(TypedExecEvent::TurnDiffUpdated(notification)) - } - ServerNotification::TurnPlanUpdated(notification) => { - Some(TypedExecEvent::TurnPlanUpdated(notification)) - } - ServerNotification::TurnStarted(_) => Some(TypedExecEvent::TurnStarted), - _ => None, - } - } -} - -#[expect( - clippy::too_many_arguments, - reason = "session mapping keeps explicit fields" -)] -pub(crate) fn session_configured_from_thread_response( - thread_id: &str, - thread_name: Option, - rollout_path: Option, - model: String, - model_provider_id: String, - service_tier: Option, - approval_policy: AskForApproval, - approvals_reviewer: ApprovalsReviewer, - sandbox_policy: SandboxPolicy, - cwd: PathBuf, - reasoning_effort: Option, -) -> Result { - let session_id = ThreadId::from_string(thread_id) - .map_err(|err| format!("thread id `{thread_id}` is invalid: {err}"))?; - - Ok(SessionConfiguredEvent { - session_id, - forked_from_id: None, - thread_name, - model, - model_provider_id, - service_tier, - approval_policy, - approvals_reviewer, - sandbox_policy, - cwd, - reasoning_effort, - history_log_id: 0, - history_entry_count: 0, - initial_messages: None, - network_proxy: None, - rollout_path, - }) -} diff --git a/codex-rs/exec/tests/all.rs b/codex-rs/exec/tests/all.rs index 7e136e4cce..dcb58b683a 100644 --- a/codex-rs/exec/tests/all.rs +++ b/codex-rs/exec/tests/all.rs @@ -1,3 +1,4 @@ // Single integration test binary that aggregates all test modules. // The submodules live in `tests/suite/`. +mod event_processor_with_json_output; mod suite; diff --git a/codex-rs/exec/tests/event_processor_with_json_output.rs b/codex-rs/exec/tests/event_processor_with_json_output.rs index 964bad3ddf..4576b329f1 100644 --- a/codex-rs/exec/tests/event_processor_with_json_output.rs +++ b/codex-rs/exec/tests/event_processor_with_json_output.rs @@ -16,6 +16,7 @@ use codex_app_server_protocol::McpToolCallResult; use codex_app_server_protocol::McpToolCallStatus as ApiMcpToolCallStatus; use codex_app_server_protocol::PatchApplyStatus as ApiPatchApplyStatus; use codex_app_server_protocol::PatchChangeKind as ApiPatchChangeKind; +use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadTokenUsage; use codex_app_server_protocol::TokenUsageBreakdown; @@ -25,6 +26,7 @@ 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::TurnStartedNotification; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::WebSearchAction as ApiWebSearchAction; use codex_protocol::ThreadId; @@ -35,43 +37,42 @@ use codex_protocol::protocol::SessionConfiguredEvent; use pretty_assertions::assert_eq; use serde_json::json; -use super::CollectedThreadEvents; -use super::EventProcessorWithJsonOutput; -use crate::event_processor::CodexStatus; -use crate::exec_events::AgentMessageItem; -use crate::exec_events::CollabAgentState; -use crate::exec_events::CollabAgentStatus; -use crate::exec_events::CollabTool; -use crate::exec_events::CollabToolCallItem; -use crate::exec_events::CollabToolCallStatus; -use crate::exec_events::CommandExecutionItem; -use crate::exec_events::CommandExecutionStatus; -use crate::exec_events::ErrorItem; -use crate::exec_events::FileChangeItem; -use crate::exec_events::FileUpdateChange as ExecFileUpdateChange; -use crate::exec_events::ItemCompletedEvent; -use crate::exec_events::ItemStartedEvent; -use crate::exec_events::ItemUpdatedEvent; -use crate::exec_events::McpToolCallItem; -use crate::exec_events::McpToolCallItemError; -use crate::exec_events::McpToolCallItemResult; -use crate::exec_events::McpToolCallStatus; -use crate::exec_events::PatchApplyStatus; -use crate::exec_events::PatchChangeKind; -use crate::exec_events::ReasoningItem; -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::TurnFailedEvent; -use crate::exec_events::TurnStartedEvent; -use crate::exec_events::Usage; -use crate::exec_events::WebSearchItem; -use crate::typed_exec_event::TypedExecEvent; +use codex_exec::event_processor_with_jsonl_output::CodexStatus; +use codex_exec::event_processor_with_jsonl_output::CollectedThreadEvents; +use codex_exec::event_processor_with_jsonl_output::EventProcessorWithJsonOutput; +use codex_exec::exec_events::AgentMessageItem; +use codex_exec::exec_events::CollabAgentState; +use codex_exec::exec_events::CollabAgentStatus; +use codex_exec::exec_events::CollabTool; +use codex_exec::exec_events::CollabToolCallItem; +use codex_exec::exec_events::CollabToolCallStatus; +use codex_exec::exec_events::CommandExecutionItem; +use codex_exec::exec_events::CommandExecutionStatus; +use codex_exec::exec_events::ErrorItem; +use codex_exec::exec_events::FileChangeItem; +use codex_exec::exec_events::FileUpdateChange as ExecFileUpdateChange; +use codex_exec::exec_events::ItemCompletedEvent; +use codex_exec::exec_events::ItemStartedEvent; +use codex_exec::exec_events::ItemUpdatedEvent; +use codex_exec::exec_events::McpToolCallItem; +use codex_exec::exec_events::McpToolCallItemError; +use codex_exec::exec_events::McpToolCallItemResult; +use codex_exec::exec_events::McpToolCallStatus; +use codex_exec::exec_events::PatchApplyStatus; +use codex_exec::exec_events::PatchChangeKind; +use codex_exec::exec_events::ReasoningItem; +use codex_exec::exec_events::ThreadErrorEvent; +use codex_exec::exec_events::ThreadEvent; +use codex_exec::exec_events::ThreadItem as ExecThreadItem; +use codex_exec::exec_events::ThreadItemDetails; +use codex_exec::exec_events::ThreadStartedEvent; +use codex_exec::exec_events::TodoItem; +use codex_exec::exec_events::TodoListItem; +use codex_exec::exec_events::TurnCompletedEvent; +use codex_exec::exec_events::TurnFailedEvent; +use codex_exec::exec_events::TurnStartedEvent; +use codex_exec::exec_events::Usage; +use codex_exec::exec_events::WebSearchItem; #[test] fn map_todo_items_preserves_text_and_completion_state() { @@ -135,7 +136,16 @@ fn session_configured_produces_thread_started_event() { fn turn_started_emits_turn_started_event() { let mut processor = EventProcessorWithJsonOutput::new(None); - let collected = processor.collect_thread_events(TypedExecEvent::TurnStarted); + let collected = + processor.collect_thread_events(ServerNotification::TurnStarted(TurnStartedNotification { + thread_id: "thread-1".to_string(), + turn: Turn { + id: "turn-1".to_string(), + items: Vec::new(), + status: TurnStatus::InProgress, + error: None, + }, + })); assert_eq!( collected, @@ -163,7 +173,7 @@ fn command_execution_started_and_completed_translate_to_thread_events() { }; let started = - processor.collect_thread_events(TypedExecEvent::ItemStarted(ItemStartedNotification { + processor.collect_thread_events(ServerNotification::ItemStarted(ItemStartedNotification { item: command_item, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), @@ -186,8 +196,8 @@ fn command_execution_started_and_completed_translate_to_thread_events() { } ); - let completed = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::CommandExecution { id: "cmd-1".to_string(), command: "ls".to_string(), @@ -202,7 +212,8 @@ fn command_execution_started_and_completed_translate_to_thread_events() { }, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( completed, CollectedThreadEvents { @@ -226,8 +237,8 @@ fn command_execution_started_and_completed_translate_to_thread_events() { fn reasoning_items_emit_summary_not_raw_content() { let mut processor = EventProcessorWithJsonOutput::new(None); - let collected = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let collected = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::Reasoning { id: "reasoning-1".to_string(), summary: vec!["safe summary".to_string()], @@ -235,7 +246,8 @@ fn reasoning_items_emit_summary_not_raw_content() { }, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( collected, @@ -257,8 +269,8 @@ fn reasoning_items_emit_summary_not_raw_content() { fn web_search_completion_preserves_query_and_action() { let mut processor = EventProcessorWithJsonOutput::new(None); - let collected = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let collected = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::WebSearch { id: "search-1".to_string(), query: "rust async await".to_string(), @@ -269,7 +281,8 @@ fn web_search_completion_preserves_query_and_action() { }, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( collected, @@ -297,7 +310,7 @@ fn web_search_start_and_completion_reuse_item_id() { let mut processor = EventProcessorWithJsonOutput::new(None); let started = - processor.collect_thread_events(TypedExecEvent::ItemStarted(ItemStartedNotification { + processor.collect_thread_events(ServerNotification::ItemStarted(ItemStartedNotification { item: ThreadItem::WebSearch { id: "search-1".to_string(), query: String::new(), @@ -307,8 +320,8 @@ fn web_search_start_and_completion_reuse_item_id() { turn_id: "turn-1".to_string(), })); - let completed = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::WebSearch { id: "search-1".to_string(), query: "rust async await".to_string(), @@ -319,7 +332,8 @@ fn web_search_start_and_completion_reuse_item_id() { }, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( started, @@ -363,7 +377,7 @@ fn mcp_tool_call_begin_and_end_emit_item_events() { let mut processor = EventProcessorWithJsonOutput::new(None); let started = - processor.collect_thread_events(TypedExecEvent::ItemStarted(ItemStartedNotification { + processor.collect_thread_events(ServerNotification::ItemStarted(ItemStartedNotification { item: ThreadItem::McpToolCall { id: "mcp-1".to_string(), server: "server_a".to_string(), @@ -377,8 +391,8 @@ fn mcp_tool_call_begin_and_end_emit_item_events() { thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), })); - let completed = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::McpToolCall { id: "mcp-1".to_string(), server: "server_a".to_string(), @@ -394,7 +408,8 @@ fn mcp_tool_call_begin_and_end_emit_item_events() { }, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( started, @@ -443,8 +458,8 @@ fn mcp_tool_call_begin_and_end_emit_item_events() { fn mcp_tool_call_failure_sets_failed_status() { let mut processor = EventProcessorWithJsonOutput::new(None); - let collected = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let collected = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::McpToolCall { id: "mcp-2".to_string(), server: "server_b".to_string(), @@ -459,7 +474,8 @@ fn mcp_tool_call_failure_sets_failed_status() { }, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( collected, @@ -489,7 +505,7 @@ fn mcp_tool_call_defaults_arguments_and_preserves_structured_content() { let mut processor = EventProcessorWithJsonOutput::new(None); let started = - processor.collect_thread_events(TypedExecEvent::ItemStarted(ItemStartedNotification { + processor.collect_thread_events(ServerNotification::ItemStarted(ItemStartedNotification { item: ThreadItem::McpToolCall { id: "mcp-3".to_string(), server: "server_c".to_string(), @@ -503,8 +519,8 @@ fn mcp_tool_call_defaults_arguments_and_preserves_structured_content() { thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), })); - let completed = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::McpToolCall { id: "mcp-3".to_string(), server: "server_c".to_string(), @@ -523,7 +539,8 @@ fn mcp_tool_call_defaults_arguments_and_preserves_structured_content() { }, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( started, @@ -576,7 +593,7 @@ fn collab_spawn_begin_and_end_emit_item_events() { let mut processor = EventProcessorWithJsonOutput::new(None); let started = - processor.collect_thread_events(TypedExecEvent::ItemStarted(ItemStartedNotification { + processor.collect_thread_events(ServerNotification::ItemStarted(ItemStartedNotification { item: ThreadItem::CollabAgentToolCall { id: "collab-1".to_string(), tool: CollabAgentTool::SpawnAgent, @@ -591,8 +608,8 @@ fn collab_spawn_begin_and_end_emit_item_events() { thread_id: "thread-parent".to_string(), turn_id: "turn-1".to_string(), })); - let completed = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::CollabAgentToolCall { id: "collab-1".to_string(), tool: CollabAgentTool::SpawnAgent, @@ -612,7 +629,8 @@ fn collab_spawn_begin_and_end_emit_item_events() { }, thread_id: "thread-parent".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( started, @@ -664,8 +682,8 @@ fn collab_spawn_begin_and_end_emit_item_events() { fn file_change_completion_maps_change_kinds() { let mut processor = EventProcessorWithJsonOutput::new(None); - let collected = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let collected = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::FileChange { id: "patch-1".to_string(), changes: vec![ @@ -689,7 +707,8 @@ fn file_change_completion_maps_change_kinds() { }, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( collected, @@ -725,8 +744,8 @@ fn file_change_completion_maps_change_kinds() { fn file_change_declined_maps_to_failed_status() { let mut processor = EventProcessorWithJsonOutput::new(None); - let collected = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let collected = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::FileChange { id: "patch-2".to_string(), changes: vec![ApiFileUpdateChange { @@ -738,7 +757,8 @@ fn file_change_declined_maps_to_failed_status() { }, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( collected, @@ -764,8 +784,8 @@ fn file_change_declined_maps_to_failed_status() { fn agent_message_item_updates_final_message() { let mut processor = EventProcessorWithJsonOutput::new(None); - let collected = - processor.collect_thread_events(TypedExecEvent::ItemCompleted(ItemCompletedNotification { + let collected = processor.collect_thread_events(ServerNotification::ItemCompleted( + ItemCompletedNotification { item: ThreadItem::AgentMessage { id: "msg-1".to_string(), text: "hello".to_string(), @@ -774,7 +794,8 @@ fn agent_message_item_updates_final_message() { }, thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), - })); + }, + )); assert_eq!( collected, @@ -797,9 +818,9 @@ fn agent_message_item_updates_final_message() { fn warning_event_produces_error_item() { let mut processor = EventProcessorWithJsonOutput::new(None); - let collected = processor.collect_thread_events(TypedExecEvent::Warning( + let collected = processor.collect_warning( "Heads up: Long conversations and multiple compactions can cause the model to be less accurate. Start a new conversation when possible to keep conversations small and targeted.".to_string(), - )); + ); assert_eq!( collected, @@ -821,7 +842,7 @@ fn warning_event_produces_error_item() { fn plan_update_emits_started_then_updated_then_completed() { let mut processor = EventProcessorWithJsonOutput::new(None); - let started = processor.collect_thread_events(TypedExecEvent::TurnPlanUpdated( + let started = processor.collect_thread_events(ServerNotification::TurnPlanUpdated( TurnPlanUpdatedNotification { thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), @@ -862,7 +883,7 @@ fn plan_update_emits_started_then_updated_then_completed() { } ); - let updated = processor.collect_thread_events(TypedExecEvent::TurnPlanUpdated( + let updated = processor.collect_thread_events(ServerNotification::TurnPlanUpdated( TurnPlanUpdatedNotification { thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), @@ -903,8 +924,8 @@ fn plan_update_emits_started_then_updated_then_completed() { } ); - let completed = - processor.collect_thread_events(TypedExecEvent::TurnCompleted(TurnCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::TurnCompleted( + TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { id: "turn-1".to_string(), @@ -912,7 +933,8 @@ fn plan_update_emits_started_then_updated_then_completed() { status: TurnStatus::Completed, error: None, }, - })); + }, + )); assert_eq!( completed, CollectedThreadEvents { @@ -947,7 +969,7 @@ fn plan_update_emits_started_then_updated_then_completed() { fn plan_update_after_completion_starts_new_todo_list_with_new_id() { let mut processor = EventProcessorWithJsonOutput::new(None); - let _ = processor.collect_thread_events(TypedExecEvent::TurnPlanUpdated( + let _ = processor.collect_thread_events(ServerNotification::TurnPlanUpdated( TurnPlanUpdatedNotification { thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), @@ -958,8 +980,8 @@ fn plan_update_after_completion_starts_new_todo_list_with_new_id() { }], }, )); - let _ = - processor.collect_thread_events(TypedExecEvent::TurnCompleted(TurnCompletedNotification { + let _ = processor.collect_thread_events(ServerNotification::TurnCompleted( + TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { id: "turn-1".to_string(), @@ -967,9 +989,10 @@ fn plan_update_after_completion_starts_new_todo_list_with_new_id() { status: TurnStatus::Completed, error: None, }, - })); + }, + )); - let restarted = processor.collect_thread_events(TypedExecEvent::TurnPlanUpdated( + let restarted = processor.collect_thread_events(ServerNotification::TurnPlanUpdated( TurnPlanUpdatedNotification { thread_id: "thread-1".to_string(), turn_id: "turn-2".to_string(), @@ -1004,29 +1027,30 @@ fn plan_update_after_completion_starts_new_todo_list_with_new_id() { 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, + let usage_update = + processor.collect_thread_events(ServerNotification::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), }, - 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 { @@ -1035,8 +1059,8 @@ fn token_usage_update_is_emitted_on_turn_completion() { } ); - let completed = - processor.collect_thread_events(TypedExecEvent::TurnCompleted(TurnCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::TurnCompleted( + TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { id: "turn-1".to_string(), @@ -1044,7 +1068,8 @@ fn token_usage_update_is_emitted_on_turn_completion() { status: TurnStatus::Completed, error: None, }, - })); + }, + )); assert_eq!( completed, CollectedThreadEvents { @@ -1064,8 +1089,8 @@ fn token_usage_update_is_emitted_on_turn_completion() { fn turn_completion_recovers_final_message_from_turn_items() { let mut processor = EventProcessorWithJsonOutput::new(None); - let completed = - processor.collect_thread_events(TypedExecEvent::TurnCompleted(TurnCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::TurnCompleted( + TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { id: "turn-1".to_string(), @@ -1078,7 +1103,8 @@ fn turn_completion_recovers_final_message_from_turn_items() { status: TurnStatus::Completed, error: None, }, - })); + }, + )); assert_eq!( completed, @@ -1097,8 +1123,8 @@ fn turn_completion_overwrites_stale_final_message_from_turn_items() { let mut processor = EventProcessorWithJsonOutput::new(None); processor.final_message = Some("stale answer".to_string()); - let completed = - processor.collect_thread_events(TypedExecEvent::TurnCompleted(TurnCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::TurnCompleted( + TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { id: "turn-1".to_string(), @@ -1111,7 +1137,8 @@ fn turn_completion_overwrites_stale_final_message_from_turn_items() { status: TurnStatus::Completed, error: None, }, - })); + }, + )); assert_eq!( completed, @@ -1130,8 +1157,8 @@ fn turn_completion_preserves_streamed_final_message_when_turn_items_are_empty() let mut processor = EventProcessorWithJsonOutput::new(None); processor.final_message = Some("streamed answer".to_string()); - let completed = - processor.collect_thread_events(TypedExecEvent::TurnCompleted(TurnCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::TurnCompleted( + TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { id: "turn-1".to_string(), @@ -1139,7 +1166,8 @@ fn turn_completion_preserves_streamed_final_message_when_turn_items_are_empty() status: TurnStatus::Completed, error: None, }, - })); + }, + )); assert_eq!( completed, @@ -1157,8 +1185,8 @@ fn turn_completion_preserves_streamed_final_message_when_turn_items_are_empty() fn turn_completion_falls_back_to_final_plan_text() { let mut processor = EventProcessorWithJsonOutput::new(None); - let completed = - processor.collect_thread_events(TypedExecEvent::TurnCompleted(TurnCompletedNotification { + let completed = processor.collect_thread_events(ServerNotification::TurnCompleted( + TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { id: "turn-1".to_string(), @@ -1169,7 +1197,8 @@ fn turn_completion_falls_back_to_final_plan_text() { status: TurnStatus::Completed, error: None, }, - })); + }, + )); assert_eq!( completed, @@ -1190,7 +1219,7 @@ fn turn_completion_falls_back_to_final_plan_text() { fn turn_failure_prefers_structured_error_message() { let mut processor = EventProcessorWithJsonOutput::new(None); - let error = processor.collect_thread_events(TypedExecEvent::Error(ErrorNotification { + let error = processor.collect_thread_events(ServerNotification::Error(ErrorNotification { error: TurnError { message: "backend failed".to_string(), codex_error_info: None, @@ -1210,8 +1239,8 @@ fn turn_failure_prefers_structured_error_message() { } ); - let failed = - processor.collect_thread_events(TypedExecEvent::TurnCompleted(TurnCompletedNotification { + let failed = processor.collect_thread_events(ServerNotification::TurnCompleted( + TurnCompletedNotification { thread_id: "thread-1".to_string(), turn: Turn { id: "turn-1".to_string(), @@ -1219,7 +1248,8 @@ fn turn_failure_prefers_structured_error_message() { status: TurnStatus::Failed, error: None, }, - })); + }, + )); assert_eq!( failed, CollectedThreadEvents { @@ -1237,7 +1267,7 @@ fn turn_failure_prefers_structured_error_message() { fn model_reroute_surfaces_as_error_item() { let mut processor = EventProcessorWithJsonOutput::new(None); - let collected = processor.collect_thread_events(TypedExecEvent::ModelRerouted( + let collected = processor.collect_thread_events(ServerNotification::ModelRerouted( codex_app_server_protocol::ModelReroutedNotification { thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), diff --git a/sdk/typescript/tests/runStreamed.test.ts b/sdk/typescript/tests/runStreamed.test.ts index 8f7907bcc0..3eb0552d38 100644 --- a/sdk/typescript/tests/runStreamed.test.ts +++ b/sdk/typescript/tests/runStreamed.test.ts @@ -36,18 +36,10 @@ describe("Codex", () => { { type: "turn.started", }, - { - type: "item.started", - item: { - id: "msg_mock", - type: "agent_message", - text: "Hi!", - }, - }, { type: "item.completed", item: { - id: "msg_mock", + id: "item_0", type: "agent_message", text: "Hi!", },