openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

rebase

Browse Source
This commit is contained in:
Roy Han
2026-03-25 17:32:01 -07:00
parent c4bdf02473 ff52cac307
commit 2c012d44cf
6 changed files with 390 additions and 951 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex.rs
+13 -29
View File
@@ -351,7 +351,8 @@ use crate::util::backoff;
use crate::windows_sandbox::WindowsSandboxLevelExt;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::AppInvocation;
use codex_analytics::CodexThreadInitializedEvent;
use codex_analytics::CodexThreadContext;
use codex_analytics::CodexThreadInitializedInput;
use codex_analytics::CodexTurnEvent;
use codex_analytics::CodexTurnSteerEvent;
use codex_analytics::InitializationMode;
@@ -673,43 +674,26 @@ impl Codex {
session
.services
.analytics_events_client
.track_thread_initialized(CodexThreadInitializedEvent {
.track_thread_initialized(CodexThreadInitializedInput {
thread_id: thread_id.to_string(),
model: thread_initialized_configuration
.collaboration_mode
.model()
.to_string(),
model_provider: thread_initialized_configuration
.original_config_do_not_use
.model_provider_id
.clone(),
reasoning_effort: thread_initialized_configuration
.collaboration_mode
.reasoning_effort(),
reasoning_summary: thread_initialized_configuration.model_reasoning_summary,
service_tier: thread_initialized_configuration.service_tier,
approval_policy: thread_initialized_configuration.approval_policy.value(),
approvals_reviewer: thread_initialized_configuration.approvals_reviewer,
sandbox_policy: thread_initialized_configuration
.sandbox_policy
.get()
.clone(),
sandbox_network_access: thread_initialized_configuration
.network_sandbox_policy
.is_enabled(),
collaboration_mode: thread_initialized_configuration.collaboration_mode.mode,
personality: thread_initialized_configuration.personality,
ephemeral: thread_initialized_configuration
.original_config_do_not_use
.ephemeral,
initialization_mode,
subagent_source: session_source_subagent_source(&thread_session_source),
parent_thread_id: session_source_parent_thread_id(&thread_session_source),
session_source: thread_session_source,
product_client_id: crate::default_client::originator().value,
created_at: SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs(),
thread_context: CodexThreadContext {
ephemeral: thread_initialized_configuration
.original_config_do_not_use
.ephemeral,
session_source: thread_session_source,
initialization_mode,
subagent_source: session_source_subagent_source(&thread_session_source),
parent_thread_id: session_source_parent_thread_id(&thread_session_source),
},
});
// This task will run until Op::Shutdown is received.
codex-rs/core/tests/suite/items.rs
+21 -176
View File
@@ -34,7 +34,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;
@@ -48,7 +47,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 {
@@ -74,36 +72,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).ok()?;
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(()));
@@ -194,7 +162,7 @@ async fn user_turn_tracks_turn_metadata_analytics() -> anyhow::Result<()> {
text_elements: Vec::new(),
}],
final_output_json_schema: None,
cwd: config.cwd.clone().to_path_buf(),
cwd: config.cwd.to_path_buf(),
approval_policy: AskForApproval::Never,
approvals_reviewer: None,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
@@ -217,30 +185,30 @@ async fn user_turn_tracks_turn_metadata_analytics() -> anyhow::Result<()> {
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let deadline = Instant::now() + Duration::from_secs(10);
let analytics_request = loop {
let event = loop {
let requests = server.received_requests().await.unwrap_or_default();
if let Some(request) = requests
let turn_events = requests
.into_iter()
.find(|request| request.url.path() == "/codex/analytics-events/events")
{
break request;
.filter(|request| request.url.path() == "/codex/analytics-events/events")
.filter_map(|request| serde_json::from_slice::<Value>(&request.body).ok())
.flat_map(|payload| {
payload["events"]
.as_array()
.cloned()
.unwrap_or_default()
.into_iter()
})
.filter(|event| event["event_type"] == "codex_turn_event")
.collect::<Vec<_>>();
if let Some(event) = turn_events.last().cloned() {
break event;
}
if Instant::now() >= deadline {
panic!("timed out waiting for turn analytics request");
panic!("timed out waiting for turn analytics event");
}
tokio::time::sleep(Duration::from_millis(50)).await;
};
let payload: Value = serde_json::from_slice(&analytics_request.body)?;
let event = payload["events"]
.as_array()
.and_then(|events| {
events
.iter()
.find(|event| event["event_type"] == "codex_turn_event")
})
.expect("codex_turn_event should be present");
let event_params = &event["event_params"];
assert_eq!(event_params["sandbox_policy"], "read_only");
@@ -272,129 +240,6 @@ async fn user_turn_tracks_turn_metadata_analytics() -> anyhow::Result<()> {
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,
})
.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 steered_turn_id = codex
.steer_input(
vec![UserInput::Text {
text: "steering metadata check".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("model").is_some()
&& event["event_params"].get("sandbox_policy").is_none()
})
.await;
let event_params = &event["event_params"];
assert_eq!(
event_params["product_client_id"],
serde_json::json!(codex_core::default_client::originator().value)
);
assert_eq!(event_params["model"], "gpt-5");
assert!(event_params["thread_id"].as_str().is_some());
assert!(event_params["turn_id"].as_str().is_some());
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("approval_policy").is_none());
assert!(event_params.get("approvals_reviewer").is_none());
assert!(event_params.get("sandbox_network_access").is_none());
assert!(event_params.get("collaboration_mode").is_none());
assert!(event_params.get("personality").is_none());
assert!(event_params.get("num_input_images").is_none());
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn resumed_thread_turn_tracks_is_not_first_turn_analytics() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
@@ -480,10 +325,10 @@ async fn resumed_thread_turn_tracks_is_not_first_turn_analytics() -> anyhow::Res
})
.filter(|event| event["event_type"] == "codex_turn_event")
.collect::<Vec<_>>();
if let Some(event) = turn_events.last().cloned() {
if turn_events.len() >= 2 {
break event;
}
if let Some(event) = turn_events.last().cloned()
&& turn_events.len() >= 2
{
break event;
}
if Instant::now() >= deadline {
panic!("timed out waiting for resumed turn analytics event");
codex-rs/core/tests/suite/thread_metadata.rs
-130
View File
@@ -1,13 +1,6 @@
use anyhow::Result;
use codex_core::CodexAuth;
use codex_core::RolloutRecorder;
use codex_core::config::Constrained;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::Personality;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::config_types::ServiceTier;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::InitialHistory;
use core_test_support::responses::start_mock_server;
use core_test_support::test_codex::test_codex;
use std::time::Duration;
@@ -23,14 +16,9 @@ async fn thread_initialization_tracks_thread_initialized_analytics() -> Result<(
.with_config(move |config| {
config.chatgpt_base_url = chatgpt_base_url;
config.model = Some("gpt-5".to_string());
config.model_reasoning_effort = Some(ReasoningEffort::High);
config.model_reasoning_summary = Some(ReasoningSummary::Detailed);
config.service_tier = Some(ServiceTier::Flex);
config.approvals_reviewer = ApprovalsReviewer::GuardianSubagent;
config.permissions.sandbox_policy = Constrained::allow_any(
codex_protocol::protocol::SandboxPolicy::new_workspace_write_policy(),
);
config.personality = Some(Personality::Friendly);
config.ephemeral = true;
})
.build(&server)
@@ -69,19 +57,6 @@ async fn thread_initialization_tracks_thread_initialized_analytics() -> Result<(
serde_json::json!(codex_core::default_client::originator().value)
);
assert_eq!(event["event_params"]["model"], "gpt-5");
assert_eq!(event["event_params"]["model_provider"], "openai");
assert_eq!(event["event_params"]["reasoning_effort"], "high");
assert_eq!(event["event_params"]["reasoning_summary"], "detailed");
assert_eq!(event["event_params"]["service_tier"], "flex");
assert_eq!(event["event_params"]["approval_policy"], "on-request");
assert_eq!(
event["event_params"]["approvals_reviewer"],
"guardian_subagent"
);
assert_eq!(event["event_params"]["sandbox_policy"], "workspace_write");
assert_eq!(event["event_params"]["sandbox_network_access"], false);
assert_eq!(event["event_params"]["collaboration_mode"], "default");
assert_eq!(event["event_params"]["personality"], "friendly");
assert_eq!(event["event_params"]["ephemeral"], true);
assert_eq!(event["event_params"]["session_source"], "user");
assert_eq!(event["event_params"]["initialization_mode"], "new");
@@ -101,108 +76,3 @@ async fn thread_initialization_tracks_thread_initialized_analytics() -> Result<(
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn resumed_thread_emits_thread_initialized_analytics() -> Result<()> {
let server = start_mock_server().await;
let chatgpt_base_url = server.uri();
let initial = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config({
let chatgpt_base_url = chatgpt_base_url.clone();
move |config| {
config.chatgpt_base_url = chatgpt_base_url;
config.model = Some("gpt-5".to_string());
config.model_reasoning_effort = Some(ReasoningEffort::High);
config.model_reasoning_summary = Some(ReasoningSummary::Detailed);
config.service_tier = Some(ServiceTier::Flex);
config.approvals_reviewer = ApprovalsReviewer::GuardianSubagent;
config.permissions.sandbox_policy = Constrained::allow_any(
codex_protocol::protocol::SandboxPolicy::new_workspace_write_policy(),
);
config.personality = Some(Personality::Friendly);
}
})
.build(&server)
.await?;
let rollout_path = initial
.codex
.rollout_path()
.expect("rollout path for initial thread");
let resume_deadline = Instant::now() + Duration::from_secs(10);
loop {
if matches!(
RolloutRecorder::get_rollout_history(&rollout_path).await?,
InitialHistory::Resumed(_)
) {
break;
}
if Instant::now() >= resume_deadline {
panic!("timed out waiting for rollout to become resumable");
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
let _resumed = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(move |config| {
config.chatgpt_base_url = chatgpt_base_url;
config.model = Some("gpt-5".to_string());
config.model_reasoning_effort = Some(ReasoningEffort::High);
config.model_reasoning_summary = Some(ReasoningSummary::Detailed);
config.service_tier = Some(ServiceTier::Flex);
config.approvals_reviewer = ApprovalsReviewer::GuardianSubagent;
config.permissions.sandbox_policy = Constrained::allow_any(
codex_protocol::protocol::SandboxPolicy::new_workspace_write_policy(),
);
config.personality = Some(Personality::Friendly);
})
.resume(&server, initial.home.clone(), rollout_path)
.await?;
let deadline = Instant::now() + Duration::from_secs(10);
let analytics_request = loop {
let requests = server.received_requests().await.unwrap_or_default();
if let Some(request) = requests.into_iter().find(|request| {
if request.url.path() != "/codex/analytics-events/events" {
return false;
}
let Ok(payload) = serde_json::from_slice::<serde_json::Value>(&request.body) else {
return false;
};
payload["events"]
.as_array()
.into_iter()
.flatten()
.any(|event| {
event["event_type"] == "codex_thread_initialized"
&& event["event_params"]["initialization_mode"] == "resumed"
})
}) {
break request;
}
if Instant::now() >= deadline {
panic!("timed out waiting for resumed thread analytics request");
}
tokio::time::sleep(Duration::from_millis(50)).await;
};
let payload: serde_json::Value =
serde_json::from_slice(&analytics_request.body).expect("analytics payload");
let event = payload["events"]
.as_array()
.expect("events array")
.iter()
.find(|event| {
event["event_type"] == "codex_thread_initialized"
&& event["event_params"]["initialization_mode"] == "resumed"
})
.expect("codex_thread_initialized resumed event should be present");
assert_eq!(event["event_params"]["session_source"], "user");
assert_eq!(event["event_params"]["initialization_mode"], "resumed");
Ok(())
}