diff --git a/codex-rs/core/src/analytics_client.rs b/codex-rs/core/src/analytics_client.rs index d25f4a075d..9d96e5001b 100644 --- a/codex-rs/core/src/analytics_client.rs +++ b/codex-rs/core/src/analytics_client.rs @@ -48,9 +48,6 @@ pub(crate) struct CodexTurnEvent { pub(crate) num_input_images: usize, } -#[derive(Clone, Copy)] -pub(crate) struct CodexTurnSteerEvent; - pub(crate) fn build_track_events_context( model_slug: String, thread_id: String, @@ -115,9 +112,6 @@ impl AnalyticsEventsQueue { TrackEventsJob::TurnEvent(job) => { send_track_turn_event(&auth_manager, job).await; } - TrackEventsJob::TurnSteer(job) => { - send_track_turn_steer(&auth_manager, job).await; - } TrackEventsJob::PluginUsed(job) => { send_track_plugin_used(&auth_manager, job).await; } @@ -244,19 +238,6 @@ impl AnalyticsEventsClient { ); } - pub(crate) fn track_turn_steer( - &self, - tracking: TrackEventsContext, - turn_steer: CodexTurnSteerEvent, - ) { - track_turn_steer( - &self.queue, - Arc::clone(&self.config), - Some(tracking), - turn_steer, - ); - } - pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) { track_plugin_management( &self.queue, @@ -299,7 +280,6 @@ enum TrackEventsJob { AppMentioned(TrackAppMentionedJob), AppUsed(TrackAppUsedJob), TurnEvent(TrackTurnEventJob), - TurnSteer(TrackTurnSteerJob), PluginUsed(TrackPluginUsedJob), PluginInstalled(TrackPluginManagementJob), PluginUninstalled(TrackPluginManagementJob), @@ -331,12 +311,6 @@ struct TrackTurnEventJob { turn_event: CodexTurnEvent, } -struct TrackTurnSteerJob { - config: Arc, - tracking: TrackEventsContext, - turn_steer: CodexTurnSteerEvent, -} - struct TrackPluginUsedJob { config: Arc, tracking: TrackEventsContext, @@ -420,39 +394,24 @@ struct CodexAppUsedEventRequest { event_params: CodexAppMetadata, } -#[derive(Default, Serialize)] +#[derive(Serialize)] struct CodexTurnEventParams { - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - #[serde(skip_serializing_if = "Option::is_none")] - turn_id: Option, - #[serde(skip_serializing_if = "Option::is_none")] + thread_id: String, + turn_id: String, product_client_id: Option, - #[serde(skip_serializing_if = "Option::is_none")] - model_slug: Option, - #[serde(skip_serializing_if = "Option::is_none")] + model: Option, submission_type: Option, - #[serde(skip_serializing_if = "Option::is_none")] - model_provider: Option, + model_provider: String, sandbox_policy: Option<&'static str>, - #[serde(skip_serializing_if = "Option::is_none")] reasoning_effort: Option, - #[serde(skip_serializing_if = "Option::is_none")] reasoning_summary: Option, - #[serde(skip_serializing_if = "Option::is_none")] - service_tier: Option, - #[serde(skip_serializing_if = "Option::is_none")] - approval_policy: Option, - #[serde(skip_serializing_if = "Option::is_none")] - approvals_reviewer: Option, - #[serde(skip_serializing_if = "Option::is_none")] - sandbox_network_access: Option, - #[serde(skip_serializing_if = "Option::is_none")] + service_tier: String, + approval_policy: String, + approvals_reviewer: String, + sandbox_network_access: bool, collaboration_mode: Option<&'static str>, - #[serde(skip_serializing_if = "Option::is_none")] personality: Option, - #[serde(skip_serializing_if = "Option::is_none")] - num_input_images: Option, + num_input_images: usize, } #[derive(Serialize)] @@ -582,26 +541,6 @@ pub(crate) fn track_turn_event( queue.try_send(job); } -pub(crate) fn track_turn_steer( - queue: &AnalyticsEventsQueue, - config: Arc, - tracking: Option, - turn_steer: CodexTurnSteerEvent, -) { - if config.analytics_enabled == Some(false) { - return; - } - let Some(tracking) = tracking else { - return; - }; - let job = TrackEventsJob::TurnSteer(TrackTurnSteerJob { - config, - tracking, - turn_steer, - }); - queue.try_send(job); -} - pub(crate) fn track_plugin_used( queue: &AnalyticsEventsQueue, config: Arc, @@ -741,20 +680,6 @@ async fn send_track_turn_event(auth_manager: &AuthManager, job: TrackTurnEventJo send_track_events(auth_manager, config, events).await; } -async fn send_track_turn_steer(auth_manager: &AuthManager, job: TrackTurnSteerJob) { - let TrackTurnSteerJob { - config, - tracking, - turn_steer, - } = job; - let events = vec![TrackEventRequest::TurnEvent(CodexTurnEventRequest { - event_type: "codex_turn_event", - event_params: codex_turn_steer_event_params(&tracking, turn_steer), - })]; - - send_track_events(auth_manager, config, events).await; -} - async fn send_track_plugin_used(auth_manager: &AuthManager, job: TrackPluginUsedJob) { let TrackPluginUsedJob { config, @@ -824,40 +749,25 @@ fn codex_turn_event_params( turn_event: CodexTurnEvent, ) -> CodexTurnEventParams { CodexTurnEventParams { - thread_id: Some(tracking.thread_id.clone()), - turn_id: Some(tracking.turn_id.clone()), + thread_id: tracking.thread_id.clone(), + turn_id: tracking.turn_id.clone(), product_client_id: Some(crate::default_client::originator().value), - model_slug: Some(tracking.model_slug.clone()), + model: Some(tracking.model_slug.clone()), submission_type: turn_event.submission_type, - model_provider: Some(turn_event.model_provider), + model_provider: turn_event.model_provider, sandbox_policy: Some(sandbox_policy_mode(&turn_event.sandbox_policy)), reasoning_effort: turn_event.reasoning_effort.map(|value| value.to_string()), reasoning_summary: reasoning_summary_mode(turn_event.reasoning_summary), - service_tier: Some( - turn_event - .service_tier - .map(|value| value.to_string()) - .unwrap_or_else(|| "default".to_string()), - ), - approval_policy: Some(turn_event.approval_policy.to_string()), - approvals_reviewer: Some(turn_event.approvals_reviewer.to_string()), - sandbox_network_access: Some(turn_event.sandbox_network_access), + service_tier: turn_event + .service_tier + .map(|value| value.to_string()) + .unwrap_or_else(|| "default".to_string()), + approval_policy: turn_event.approval_policy.to_string(), + approvals_reviewer: turn_event.approvals_reviewer.to_string(), + sandbox_network_access: turn_event.sandbox_network_access, collaboration_mode: Some(collaboration_mode_mode(turn_event.collaboration_mode)), personality: personality_mode(turn_event.personality), - num_input_images: Some(turn_event.num_input_images), - } -} - -fn codex_turn_steer_event_params( - tracking: &TrackEventsContext, - _turn_steer: CodexTurnSteerEvent, -) -> CodexTurnEventParams { - CodexTurnEventParams { - thread_id: Some(tracking.thread_id.clone()), - turn_id: Some(tracking.turn_id.clone()), - product_client_id: Some(crate::default_client::originator().value), - model_slug: Some(tracking.model_slug.clone()), - ..Default::default() + num_input_images: turn_event.num_input_images, } } diff --git a/codex-rs/core/src/analytics_client_tests.rs b/codex-rs/core/src/analytics_client_tests.rs index 6b0f550c9f..4802f24dea 100644 --- a/codex-rs/core/src/analytics_client_tests.rs +++ b/codex-rs/core/src/analytics_client_tests.rs @@ -6,7 +6,6 @@ use super::CodexPluginEventRequest; use super::CodexPluginUsedEventRequest; use super::CodexTurnEvent; use super::CodexTurnEventRequest; -use super::CodexTurnSteerEvent; use super::InvocationType; use super::SubmissionType; use super::TrackEventRequest; @@ -15,7 +14,6 @@ use super::codex_app_metadata; use super::codex_plugin_metadata; use super::codex_plugin_used_metadata; use super::codex_turn_event_params; -use super::codex_turn_steer_event_params; use super::normalize_path_for_skill_id; use crate::plugins::AppConnectorId; use crate::plugins::PluginCapabilitySummary; @@ -227,7 +225,7 @@ fn turn_event_serializes_expected_shape() { ), }); - let payload = serde_json::to_value(&event).expect("serialize codex turn event"); + let payload = serde_json::to_value(&event).expect("serialize turn event"); assert_eq!( payload, @@ -237,7 +235,7 @@ fn turn_event_serializes_expected_shape() { "thread_id": "thread-2", "turn_id": "turn-2", "product_client_id": crate::default_client::originator().value, - "model_slug": "gpt-5", + "model": "gpt-5", "submission_type": "prompt", "model_provider": "openai", "sandbox_policy": "read_only", @@ -255,33 +253,6 @@ fn turn_event_serializes_expected_shape() { ); } -#[test] -fn turn_steer_event_serializes_expected_shape() { - let tracking = TrackEventsContext { - model_slug: "gpt-5".to_string(), - thread_id: "thread-2".to_string(), - turn_id: "turn-2".to_string(), - }; - let event = TrackEventRequest::TurnEvent(CodexTurnEventRequest { - event_type: "codex_turn_event", - event_params: codex_turn_steer_event_params(&tracking, CodexTurnSteerEvent), - }); - - let payload = serde_json::to_value(&event).expect("serialize codex turn steer event"); - - assert_eq!( - payload, - json!({ - "event_type": "codex_turn_event", - "event_params": { - "thread_id": "thread-2", - "turn_id": "turn-2", - "product_client_id": crate::default_client::originator().value, - "model_slug": "gpt-5" - } - }) - ); -} #[test] fn plugin_used_event_serializes_expected_shape() { diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index a8df661311..2e3d436fca 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -15,7 +15,6 @@ use crate::agent::agent_status_from_event; use crate::analytics_client::AnalyticsEventsClient; use crate::analytics_client::AppInvocation; use crate::analytics_client::CodexTurnEvent; -use crate::analytics_client::CodexTurnSteerEvent; use crate::analytics_client::InvocationType; use crate::analytics_client::build_track_events_context; use crate::apps::render_apps_section; @@ -238,9 +237,6 @@ pub(crate) struct PreviousTurnSettings { use crate::exec_policy::ExecPolicyUpdateError; use crate::feedback_tags; -use crate::file_watcher::FileWatcher; -use crate::file_watcher::FileWatcherEvent; -use crate::git_info::get_git_repo_root; use crate::guardian::GuardianReviewSessionManager; use crate::hook_runtime::PendingInputHookDisposition; use crate::hook_runtime::inspect_pending_input; @@ -325,6 +321,8 @@ use crate::skills::injection::ToolMentionKind; use crate::skills::injection::app_id_from_path; use crate::skills::injection::tool_kind_for_path; use crate::skills::resolve_skill_dependencies_for_turn; +use crate::skills_watcher::SkillsWatcher; +use crate::skills_watcher::SkillsWatcherEvent; use crate::state::ActiveTurn; use crate::state::SessionServices; use crate::state::SessionState; @@ -353,6 +351,7 @@ use crate::unified_exec::UnifiedExecProcessManager; use crate::util::backoff; use crate::windows_sandbox::WindowsSandboxLevelExt; use codex_async_utils::OrCancelExt; +use codex_git_utils::get_git_repo_root; use codex_otel::SessionTelemetry; use codex_otel::TelemetryAuthMode; use codex_otel::metrics::names::THREAD_STARTED_METRIC; @@ -406,7 +405,7 @@ pub(crate) struct CodexSpawnArgs { pub(crate) skills_manager: Arc, pub(crate) plugins_manager: Arc, pub(crate) mcp_manager: Arc, - pub(crate) file_watcher: Arc, + pub(crate) skills_watcher: Arc, pub(crate) conversation_history: InitialHistory, pub(crate) session_source: SessionSource, pub(crate) agent_control: AgentControl, @@ -459,7 +458,7 @@ impl Codex { skills_manager, plugins_manager, mcp_manager, - file_watcher, + skills_watcher, conversation_history, session_source, agent_control, @@ -647,7 +646,7 @@ impl Codex { skills_manager, plugins_manager, mcp_manager.clone(), - file_watcher, + skills_watcher, agent_control, ) .await @@ -1303,13 +1302,13 @@ impl Session { self.out_of_band_elicitation_paused.send_replace(paused); } - fn start_file_watcher_listener(self: &Arc) { - let mut rx = self.services.file_watcher.subscribe(); + fn start_skills_watcher_listener(self: &Arc) { + let mut rx = self.services.skills_watcher.subscribe(); let weak_sess = Arc::downgrade(self); tokio::spawn(async move { loop { match rx.recv().await { - Ok(FileWatcherEvent::SkillsChanged { .. }) => { + Ok(SkillsWatcherEvent::SkillsChanged { .. }) => { let Some(sess) = weak_sess.upgrade() else { break; }; @@ -1446,7 +1445,7 @@ impl Session { skills_manager: Arc, plugins_manager: Arc, mcp_manager: Arc, - file_watcher: Arc, + skills_watcher: Arc, agent_control: AgentControl, ) -> anyhow::Result> { debug!( @@ -1865,7 +1864,7 @@ impl Session { skills_manager, plugins_manager: Arc::clone(&plugins_manager), mcp_manager: Arc::clone(&mcp_manager), - file_watcher, + skills_watcher, agent_control, network_proxy, network_approval: Arc::clone(&network_approval), @@ -1944,7 +1943,7 @@ impl Session { } // Start the watcher after SessionConfigured so it cannot emit earlier events. - sess.start_file_watcher_listener(); + sess.start_skills_watcher_listener(); // Construct sandbox_state before MCP startup so it can be sent to each // MCP server immediately after it becomes ready (avoiding blocking). let sandbox_state = SandboxState { @@ -2749,16 +2748,6 @@ impl Session { )) } - async fn active_turn_tracking(&self) -> Option { - let active = self.active_turn.lock().await; - let (_, task) = active.as_ref()?.tasks.first()?; - Some(build_track_events_context( - task.turn_context.model_info.slug.clone(), - self.conversation_id.to_string(), - task.turn_context.sub_id.clone(), - )) - } - pub(crate) async fn record_execpolicy_amendment_message( &self, sub_id: &str, @@ -3906,7 +3895,6 @@ impl Session { let Some((active_turn_id, _)) = active_turn.tasks.first() else { return Err(SteerInputError::NoActiveTurn(input)); }; - let active_turn_id = active_turn_id.clone(); if let Some(expected_turn_id) = expected_turn_id && expected_turn_id != active_turn_id @@ -3934,14 +3922,7 @@ impl Session { let mut turn_state = active_turn.turn_state.lock().await; turn_state.push_pending_input(input.into()); - drop(turn_state); - drop(active); - if let Some(tracking) = self.active_turn_tracking().await { - self.services - .analytics_events_client - .track_turn_steer(tracking, CodexTurnSteerEvent); - } - Ok(active_turn_id) + Ok(active_turn_id.clone()) } /// Returns the input if there was no task running to inject into. @@ -3985,6 +3966,17 @@ impl Session { } } + pub(crate) async fn pending_input_snapshot(&self) -> Vec { + let active = self.active_turn.lock().await; + match active.as_ref() { + Some(at) => { + let ts = at.turn_state.lock().await; + ts.pending_input_snapshot() + } + None => Vec::with_capacity(0), + } + } + /// Queue response items to be injected into the next active turn created for this session. pub(crate) async fn queue_response_items_for_next_turn(&self, items: Vec) { if items.is_empty() { @@ -3999,6 +3991,12 @@ impl Session { std::mem::take(&mut *self.idle_pending_input.lock().await) } + pub(crate) async fn queued_response_items_for_next_turn_snapshot( + &self, + ) -> Vec { + self.idle_pending_input.lock().await.clone() + } + pub(crate) async fn has_queued_response_items_for_next_turn(&self) -> bool { !self.idle_pending_input.lock().await.is_empty() } @@ -4677,27 +4675,27 @@ mod handlers { } } + /// Records an inter-agent assistant envelope and, when requested, wakes the recipient by + /// starting a regular turn if the session is currently idle. pub async fn inter_agent_communication( sess: &Arc, sub_id: String, communication: InterAgentCommunication, ) { let pending_item = communication.to_response_input_item(); - if sess - .inject_response_items(vec![pending_item.clone()]) - .await - .is_ok() - { + if let Ok(()) = sess.inject_response_items(vec![pending_item.clone()]).await { return; } - let turn_context = sess.new_default_turn_with_sub_id(sub_id).await; - sess.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref()) - .await; sess.queue_response_items_for_next_turn(vec![pending_item]) .await; - sess.spawn_task(turn_context, Vec::new(), crate::tasks::RegularTask::new()) - .await; + if communication.trigger_turn { + let turn_context = sess.new_default_turn_with_sub_id(sub_id).await; + sess.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref()) + .await; + sess.spawn_task(turn_context, Vec::new(), crate::tasks::RegularTask::new()) + .await; + } } pub async fn run_user_shell_command(sess: &Arc, sub_id: String, command: String) { @@ -6025,14 +6023,11 @@ pub(crate) async fn run_turn( sess.services.analytics_events_client.track_turn_event( tracking, CodexTurnEvent { - submission_type: match submission_type { - SubmissionType::Prompt => Some(SubmissionType::Prompt), - SubmissionType::PromptQueued => Some(SubmissionType::PromptQueued), - }, + submission_type: Some(submission_type), model_provider: turn_context.config.model_provider_id.clone(), sandbox_policy: turn_context.sandbox_policy.get().clone(), reasoning_effort: turn_context.reasoning_effort, - reasoning_summary: turn_context.reasoning_summary, + reasoning_summary: Some(turn_context.reasoning_summary), service_tier: turn_context.config.service_tier, approval_policy: turn_context.approval_policy.value(), approvals_reviewer: turn_context.config.approvals_reviewer, diff --git a/codex-rs/core/tests/suite/items.rs b/codex-rs/core/tests/suite/items.rs index 8f91366fee..b51f6443ec 100644 --- a/codex-rs/core/tests/suite/items.rs +++ b/codex-rs/core/tests/suite/items.rs @@ -33,7 +33,6 @@ use core_test_support::responses::ev_response_created; use core_test_support::responses::ev_web_search_call_added_partial; use core_test_support::responses::ev_web_search_call_done; use core_test_support::responses::mount_sse_once; -use core_test_support::responses::mount_sse_sequence; use core_test_support::responses::sse; use core_test_support::responses::start_mock_server; use core_test_support::skip_if_no_network; @@ -47,7 +46,6 @@ use std::path::Path; use std::path::PathBuf; use std::time::Duration; use std::time::Instant; -use tempfile::tempdir; fn image_generation_artifact_path(codex_home: &Path, session_id: &str, call_id: &str) -> PathBuf { fn sanitize(value: &str) -> String { @@ -73,37 +71,6 @@ fn image_generation_artifact_path(codex_home: &Path, session_id: &str, call_id: .join(format!("{}.png", sanitize(call_id))) } -async fn wait_for_analytics_event( - server: &wiremock::MockServer, - event_type: &str, - event_match: impl Fn(&Value) -> bool, -) -> Value { - let deadline = Instant::now() + Duration::from_secs(10); - loop { - let requests = server.received_requests().await.unwrap_or_default(); - if let Some(event) = requests - .into_iter() - .filter(|request| request.url.path() == "/codex/analytics-events/events") - .find_map(|request| { - let payload: Value = - serde_json::from_slice(&request.body).expect("analytics payload"); - payload["events"].as_array().and_then(|events| { - events - .iter() - .find(|event| event["event_type"] == event_type && event_match(event)) - .cloned() - }) - }) - { - return event; - } - if Instant::now() >= deadline { - panic!("timed out waiting for analytics event {event_type}"); - } - tokio::time::sleep(Duration::from_millis(50)).await; - } -} - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn user_message_item_is_emitted() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); @@ -163,7 +130,7 @@ async fn user_message_item_is_emitted() -> anyhow::Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn user_turn_tracks_turn_event_analytics() -> anyhow::Result<()> { +async fn user_turn_tracks_turn_metadata_analytics() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; @@ -235,21 +202,20 @@ async fn user_turn_tracks_turn_event_analytics() -> anyhow::Result<()> { let event = payload["events"] .as_array() .and_then(|events| { - events.iter().find(|event| { - event["event_type"] == "codex_turn_event" - && event["event_params"]["submission_type"] == "prompt" - }) + events + .iter() + .find(|event| event["event_type"] == "codex_turn_event") }) - .expect("codex_turn_event prompt event should be present"); + .expect("codex_turn_event should be present"); let event_params = &event["event_params"]; - assert_eq!(event_params["submission_type"], "prompt"); assert_eq!(event_params["sandbox_policy"], "read_only"); assert_eq!( event_params["product_client_id"], serde_json::json!(codex_core::default_client::originator().value) ); + assert_eq!(event_params["submission_type"], "prompt"); assert_eq!(event_params["model"], session_configured.model); assert_eq!(event_params["model_provider"], config.model_provider_id); assert_eq!(event_params["reasoning_effort"], "high"); @@ -273,199 +239,6 @@ async fn user_turn_tracks_turn_event_analytics() -> anyhow::Result<()> { Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn user_turn_tracks_turn_event_prompt_type_analytics() -> anyhow::Result<()> { - skip_if_no_network!(Ok(())); - - let server = start_mock_server().await; - mount_sse_once( - &server, - sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), - ) - .await; - - let chatgpt_base_url = server.uri(); - let mut builder = test_codex() - .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) - .with_config(move |config| { - config.chatgpt_base_url = chatgpt_base_url; - }); - let TestCodex { - codex, - session_configured, - config, - .. - } = builder.build(&server).await?; - - codex - .submit(Op::UserTurn { - items: vec![UserInput::Text { - text: "hello input message analytics".into(), - text_elements: Vec::new(), - }], - final_output_json_schema: None, - cwd: config.cwd.clone(), - approval_policy: AskForApproval::Never, - approvals_reviewer: None, - sandbox_policy: SandboxPolicy::new_read_only_policy(), - model: session_configured.model.clone(), - effort: Some(ReasoningEffort::High), - summary: Some(ReasoningSummary::Detailed), - service_tier: Some(Some(ServiceTier::Flex)), - collaboration_mode: Some(CollaborationMode { - mode: ModeKind::Plan, - settings: Settings { - model: session_configured.model.clone(), - reasoning_effort: Some(ReasoningEffort::High), - developer_instructions: None, - }, - }), - personality: None, - submission_type: None, - }) - .await?; - - wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; - - let event = wait_for_analytics_event(&server, "codex_turn_event", |event| { - event["event_params"]["submission_type"] == "prompt" - }) - .await; - let event_params = &event["event_params"]; - - assert_eq!(event_params["submission_type"], "prompt"); - assert_eq!( - event_params["product_client_id"], - serde_json::json!(codex_core::default_client::originator().value) - ); - assert_eq!(event_params["model_slug"], session_configured.model); - assert!(event_params["thread_id"].as_str().is_some()); - assert!(event_params["turn_id"].as_str().is_some()); - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn user_turn_tracks_turn_steer_analytics() -> anyhow::Result<()> { - skip_if_no_network!(Ok(())); - - let server = start_mock_server().await; - let temp = tempdir()?; - let unblock_path = temp.path().join("unblock-steering"); - let command = format!( - "while [ ! -f \"{}\" ]; do sleep 0.01; done; echo done", - unblock_path.display() - ); - let call_id = "shell-steering-call"; - - mount_sse_sequence( - &server, - vec![ - sse(vec![ - ev_response_created("resp-1"), - core_test_support::responses::ev_function_call( - call_id, - "shell", - &serde_json::to_string(&serde_json::json!({ - "command": ["/bin/sh", "-c", command], - }))?, - ), - ev_completed("resp-1"), - ]), - sse(vec![ - ev_assistant_message("msg-2", "done"), - ev_completed("resp-2"), - ]), - ], - ) - .await; - - let chatgpt_base_url = server.uri(); - let test = test_codex() - .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) - .with_model("gpt-5") - .with_config(move |config| { - config.chatgpt_base_url = chatgpt_base_url; - }) - .build(&server) - .await?; - let codex = test.codex.clone(); - let turn_model = test.session_configured.model.clone(); - - codex - .submit(Op::UserTurn { - items: vec![UserInput::Text { - text: "start steering flow".into(), - text_elements: Vec::new(), - }], - final_output_json_schema: None, - cwd: test.cwd_path().to_path_buf(), - approval_policy: AskForApproval::Never, - approvals_reviewer: None, - sandbox_policy: SandboxPolicy::DangerFullAccess, - model: turn_model, - effort: None, - summary: None, - service_tier: None, - collaboration_mode: None, - personality: None, - submission_type: None, - }) - .await?; - - let turn_id = wait_for_event_match(&codex, |ev| match ev { - EventMsg::TurnStarted(event) => Some(event.turn_id.clone()), - _ => None, - }) - .await; - - wait_for_event_match(&codex, |ev| match ev { - EventMsg::ExecCommandBegin(event) if event.call_id == call_id => Some(()), - _ => None, - }) - .await; - - let steering_text = "steering metadata check"; - let steered_turn_id = codex - .steer_input( - vec![UserInput::Text { - text: steering_text.into(), - text_elements: Vec::new(), - }], - Some(turn_id.as_str()), - ) - .await - .expect("steer should succeed on active turn"); - assert_eq!(steered_turn_id, turn_id); - - std::fs::write(&unblock_path, "go")?; - - wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; - - let event = wait_for_analytics_event(&server, "codex_turn_event", |event| { - event["event_params"].get("submission_type").is_none() - && event["event_params"].get("sandbox_policy").is_none() - }) - .await; - let event_params = &event["event_params"]; - - assert!(event_params.get("submission_type").is_none()); - assert!(event_params.get("sandbox_policy").is_none()); - assert!(event_params.get("reasoning_effort").is_none()); - assert!(event_params.get("reasoning_summary").is_none()); - assert!(event_params.get("service_tier").is_none()); - assert!(event_params.get("collaboration_mode").is_none()); - assert_eq!( - event_params["product_client_id"], - serde_json::json!(codex_core::default_client::originator().value) - ); - assert_eq!(event_params["model_slug"], "gpt-5"); - assert!(event_params["thread_id"].as_str().is_some()); - assert!(event_params["turn_id"].as_str().is_some()); - - Ok(()) -} - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn assistant_message_item_is_emitted() -> anyhow::Result<()> { skip_if_no_network!(Ok(()));