openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

logic cleanup

Browse Source
This commit is contained in:
Roy Han
2026-03-24 17:35:46 -07:00
parent 7f10658d93
commit ade9c6fc5b
4 changed files with 72 additions and 423 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/analytics_client.rs
+22 -112
View File
@@ -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<Config>,
tracking: TrackEventsContext,
turn_steer: CodexTurnSteerEvent,
}
struct TrackPluginUsedJob {
config: Arc<Config>,
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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
turn_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: String,
turn_id: String,
product_client_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
model_slug: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
model: Option<String>,
submission_type: Option<SubmissionType>,
#[serde(skip_serializing_if = "Option::is_none")]
model_provider: Option<String>,
model_provider: String,
sandbox_policy: Option<&'static str>,
#[serde(skip_serializing_if = "Option::is_none")]
reasoning_effort: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
reasoning_summary: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
service_tier: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
approval_policy: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
approvals_reviewer: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
sandbox_network_access: Option<bool>,
#[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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
num_input_images: Option<usize>,
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<Config>,
tracking: Option<TrackEventsContext>,
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<Config>,
@@ -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,
}
}
codex-rs/core/src/analytics_client_tests.rs
+2 -31
View File
@@ -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() {
codex-rs/core/src/codex.rs
+42 -47
View File
@@ -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<SkillsManager>,
pub(crate) plugins_manager: Arc<PluginsManager>,
pub(crate) mcp_manager: Arc<McpManager>,
pub(crate) file_watcher: Arc<FileWatcher>,
pub(crate) skills_watcher: Arc<SkillsWatcher>,
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<Self>) {
let mut rx = self.services.file_watcher.subscribe();
fn start_skills_watcher_listener(self: &Arc<Self>) {
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<SkillsManager>,
plugins_manager: Arc<PluginsManager>,
mcp_manager: Arc<McpManager>,
file_watcher: Arc<FileWatcher>,
skills_watcher: Arc<SkillsWatcher>,
agent_control: AgentControl,
) -> anyhow::Result<Arc<Self>> {
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<crate::analytics_client::TrackEventsContext> {
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<ResponseInputItem> {
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<ResponseInputItem>) {
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<ResponseInputItem> {
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<Session>,
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<Session>, 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,
codex-rs/core/tests/suite/items.rs
+6 -233
View File
@@ -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(()));