openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-03 19:53:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplify exec event handling

Browse Source
This commit is contained in:
Eric Traut
2026-03-21 15:14:45 -06:00
parent 36ba70c2bc
commit efa0f04e6f
8 changed files with 287 additions and 337 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/exec/src/event_processor.rs
+6 -4
View File
@@ -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) {}
}
codex-rs/exec/src/event_processor_with_human_output.rs
+29 -27
View File
@@ -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(&notification.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 {
codex-rs/exec/src/event_processor_with_jsonl_output.rs
+47 -36
View File
@@ -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<PathBuf>,
@@ -70,9 +70,9 @@ struct RunningTodoList {
}
#[derive(Debug, PartialEq)]
struct CollectedThreadEvents {
events: Vec<ThreadEvent>,
status: CodexStatus,
pub struct CollectedThreadEvents {
pub events: Vec<ThreadEvent>,
pub status: CodexStatus,
}
impl EventProcessorWithJsonOutput {
@@ -116,7 +116,7 @@ impl EventProcessorWithJsonOutput {
}
}
fn map_todo_items(plan: &[codex_app_server_protocol::TurnPlanStep]) -> Vec<TodoItem> {
pub fn map_todo_items(plan: &[codex_app_server_protocol::TurnPlanStep]) -> Vec<TodoItem> {
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(&notification.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;
codex-rs/exec/src/lib.rs
+44 -8
View File
@@ -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<()> {
&notification,
&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<String>,
rollout_path: Option<PathBuf>,
model: String,
model_provider_id: String,
service_tier: Option<codex_protocol::config_types::ServiceTier>,
approval_policy: AskForApproval,
approvals_reviewer: codex_protocol::config_types::ApprovalsReviewer,
sandbox_policy: SandboxPolicy,
cwd: PathBuf,
reasoning_effort: Option<codex_protocol::openai_models::ReasoningEffort>,
) -> Result<SessionConfiguredEvent, String> {
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")
}
codex-rs/exec/src/typed_exec_event.rs
-124
View File
@@ -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<TypedExecEvent> {
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<String>,
rollout_path: Option<PathBuf>,
model: String,
model_provider_id: String,
service_tier: Option<ServiceTier>,
approval_policy: AskForApproval,
approvals_reviewer: ApprovalsReviewer,
sandbox_policy: SandboxPolicy,
cwd: PathBuf,
reasoning_effort: Option<ReasoningEffort>,
) -> Result<SessionConfiguredEvent, String> {
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,
})
}
codex-rs/exec/tests/all.rs
+1
View File
@@ -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;
codex-rs/exec/tests/event_processor_with_json_output.rs
+159 -129
View File
@@ -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(),
sdk/typescript/tests/runStreamed.test.ts
+1 -9
View File
@@ -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!",
},