use super::turn_context::TurnEnvironment;use super::*;use crate::codex_thread::TryStartTurnIfIdleRejectionReason;use crate::config::ConfigBuilder;use crate::config::ConfigOverrides;use crate::config::test_config;use crate::context::ContextualUserFragment;use crate::context::TurnAborted;use crate::environment_selection::ThreadEnvironments;use crate::function_tool::FunctionCallError;use crate::shell::default_user_shell;use crate::shell_snapshot::ShellSnapshot;use crate::skills::SkillRenderSideEffects;use crate::skills::render::SkillMetadataBudget;use crate::test_support::models_manager_with_provider;use crate::tools::format_exec_output_str;use codex_config::ConfigLayerStack;use codex_config::ConfigLayerStackOrdering;use codex_config::LoaderOverrides;use codex_config::NetworkConstraints;use codex_config::NetworkDomainPermissionToml;use codex_config::NetworkDomainPermissionsToml;use codex_config::RequirementSource;use codex_config::Sourced;use codex_config::loader::project_trust_key;use codex_config::types::ToolSuggestDisabledTool;use core_test_support::test_codex::local_selections;use codex_features::Feature;use codex_login::CodexAuth;use codex_model_provider_info::ModelProviderInfo;use codex_models_manager::bundled_models_response;use codex_models_manager::model_info;use codex_models_manager::test_support::construct_model_info_offline_for_tests;use codex_models_manager::test_support::get_model_offline_for_tests;use codex_protocol::AgentPath;use codex_protocol::SessionId;use codex_protocol::ThreadId;use codex_protocol::config_types::SERVICE_TIER_DEFAULT_REQUEST_VALUE;use codex_protocol::config_types::ServiceTier;use codex_protocol::config_types::TrustLevel;use codex_protocol::exec_output::ExecToolCallOutput;use codex_protocol::models::ActivePermissionProfile;use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_WORKSPACE;use codex_protocol::models::FileSystemPermissions;use codex_protocol::models::FunctionCallOutputBody;use codex_protocol::models::FunctionCallOutputContentItem;use codex_protocol::models::FunctionCallOutputPayload;use codex_protocol::models::ImageDetail;use codex_protocol::models::PermissionProfile;use codex_protocol::models::SandboxEnforcement;use codex_protocol::openai_models::ModelServiceTier;use codex_protocol::permissions::FileSystemAccessMode;use codex_protocol::permissions::FileSystemPath;use codex_protocol::permissions::FileSystemSandboxEntry;use codex_protocol::permissions::FileSystemSandboxPolicy;use codex_protocol::permissions::FileSystemSpecialPath;use codex_protocol::protocol::NonSteerableTurnKind;use codex_protocol::protocol::SandboxPolicy;use codex_protocol::protocol::TurnEnvironmentSelections;use codex_protocol::request_permissions::PermissionGrantScope;use codex_protocol::request_permissions::RequestPermissionProfile;use codex_utils_path_uri::PathUri;use tracing::Span;use crate::rollout::recorder::RolloutRecorder;use crate::state::ActiveTurn;use crate::state::TaskKind;use crate::tasks::SessionTask;use crate::tasks::SessionTaskContext;use crate::tasks::UserShellCommandMode;use crate::tasks::execute_user_shell_command;use crate::tools::ToolRouter;use crate::tools::context::ToolInvocation;use crate::tools::context::ToolPayload;use crate::tools::handlers::ExecCommandHandler;use crate::tools::handlers::RequestPermissionsHandler;use crate::tools::handlers::ShellCommandHandler;use crate::tools::registry::ToolExecutor;use crate::tools::router::ToolCallSource;use crate::turn_diff_tracker::TurnDiffTracker;use codex_app_server_protocol::AppInfo;use codex_app_server_protocol::McpElicitationSchema;use codex_config::config_toml::ConfigToml;use codex_config::config_toml::ProjectConfig;use codex_config::permissions_toml::FilesystemPermissionToml;use codex_config::permissions_toml::FilesystemPermissionsToml;use codex_config::permissions_toml::NetworkToml;use codex_config::permissions_toml::PermissionProfileToml;use codex_config::permissions_toml::PermissionsToml;use codex_execpolicy::Decision;use codex_execpolicy::NetworkRuleProtocol;use codex_execpolicy::Policy;use codex_network_proxy::NetworkProxyConfig;use codex_otel::MetricsClient;use codex_otel::MetricsConfig;use codex_otel::THREAD_SKILLS_DESCRIPTION_TRUNCATED_CHARS_METRIC;use codex_otel::THREAD_SKILLS_ENABLED_TOTAL_METRIC;use codex_otel::THREAD_SKILLS_KEPT_TOTAL_METRIC;use codex_otel::THREAD_SKILLS_TRUNCATED_METRIC;use codex_otel::TelemetryAuthMode;use codex_protocol::config_types::CollaborationMode;use codex_protocol::config_types::ModeKind;use codex_protocol::config_types::Settings;use codex_protocol::models::BaseInstructions;use codex_protocol::models::ContentItem;use codex_protocol::models::ResponseItem;use codex_protocol::models::ResponseItemMetadata;use codex_protocol::protocol::AskForApproval;use codex_protocol::protocol::CodexErrorInfo;use codex_protocol::protocol::CompactedItem;use codex_protocol::protocol::ConversationAudioParams;use codex_protocol::protocol::CreditsSnapshot;use codex_protocol::protocol::GranularApprovalConfig;use codex_protocol::protocol::InitialHistory;use codex_protocol::protocol::InterAgentCommunication;use codex_protocol::protocol::MultiAgentVersion;use codex_protocol::protocol::NetworkApprovalProtocol;use codex_protocol::protocol::RateLimitSnapshot;use codex_protocol::protocol::RateLimitWindow;use codex_protocol::protocol::RealtimeAudioFrame;use codex_protocol::protocol::RealtimeConversationListVoicesResponseEvent;use codex_protocol::protocol::RealtimeVoice;use codex_protocol::protocol::RealtimeVoicesList;use codex_protocol::protocol::ResumedHistory;use codex_protocol::protocol::RolloutItem;use codex_protocol::protocol::SessionMeta;use codex_protocol::protocol::SessionMetaLine;use codex_protocol::protocol::SkillScope;use codex_protocol::protocol::Submission;use codex_protocol::protocol::ThreadRolledBackEvent;use codex_protocol::protocol::ThreadSettingsOverrides;use codex_protocol::protocol::TokenCountEvent;use codex_protocol::protocol::TokenUsage;use codex_protocol::protocol::TokenUsageInfo;use codex_protocol::protocol::TurnAbortedEvent;use codex_protocol::protocol::TurnCompleteEvent;use codex_protocol::protocol::TurnStartedEvent;use codex_protocol::protocol::UserMessageEvent;use codex_protocol::protocol::W3cTraceContext;use codex_rmcp_client::ElicitationAction;use core_test_support::PathBufExt;use core_test_support::PathExt;use core_test_support::context_snapshot;use core_test_support::context_snapshot::ContextSnapshotOptions;use core_test_support::context_snapshot::ContextSnapshotRenderMode;use core_test_support::responses::ev_completed;use core_test_support::responses::ev_response_created;use core_test_support::responses::mount_sse_once;use core_test_support::responses::sse;use core_test_support::responses::start_mock_server;use core_test_support::test_codex::local;use core_test_support::test_codex::test_codex;use core_test_support::test_path_buf;use core_test_support::tracing::install_test_tracing;use core_test_support::wait_for_event;use opentelemetry::trace::TraceContextExt;use opentelemetry::trace::TraceId;use opentelemetry_sdk::metrics::InMemoryMetricExporter;use opentelemetry_sdk::metrics::data::AggregatedMetrics;use opentelemetry_sdk::metrics::data::Metric;use opentelemetry_sdk::metrics::data::MetricData;use opentelemetry_sdk::metrics::data::ResourceMetrics;use std::path::Path;use std::time::Duration;use tokio::sync::Semaphore;use tokio::time::sleep;use tokio::time::timeout;use tracing_opentelemetry::OpenTelemetrySpanExt;use codex_protocol::mcp::CallToolResult as McpCallToolResult;use pretty_assertions::assert_eq;use serde::Deserialize;use serde_json::json;use std::path::PathBuf;use std::sync::Arc;use std::sync::OnceLock;use std::time::Duration as StdDuration;mod guardian_tests;struct InstructionsTestCase { slug: &'static str, expects_apply_patch_description: bool,}fn user_message(text: &str) -> ResponseItem { ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: text.to_string(), }], phase: None, metadata: None, }}fn assistant_message(text: &str) -> ResponseItem { ResponseItem::Message { id: None, role: "assistant".to_string(), content: vec![ContentItem::OutputText { text: text.to_string(), }], phase: None, metadata: None, }}fn test_session_telemetry_without_metadata() -> SessionTelemetry { let exporter = InMemoryMetricExporter::default(); let metrics = MetricsClient::new( MetricsConfig::in_memory("test", "codex-core", env!("CARGO_PKG_VERSION"), exporter) .with_runtime_reader(), ) .expect("in-memory metrics client"); SessionTelemetry::new( ThreadId::new(), "gpt-5.4", "gpt-5.4", /*account_id*/ None, /*account_email*/ None, /*auth_mode*/ None, "test_originator".to_string(), /*log_user_prompts*/ false, "tty".to_string(), SessionSource::Cli, ) .with_metrics_without_metadata_tags(metrics)}fn find_metric<'a>(resource_metrics: &'a ResourceMetrics, name: &str) -> &'a Metric { for scope_metrics in resource_metrics.scope_metrics() { for metric in scope_metrics.metrics() { if metric.name() == name { return metric; } } } panic!("metric {name} missing");}fn histogram_sum(resource_metrics: &ResourceMetrics, name: &str) -> u64 { let metric = find_metric(resource_metrics, name); match metric.data() { AggregatedMetrics::F64(data) => match data { MetricData::Histogram(histogram) => { let points: Vec<_> = histogram.data_points().collect(); assert_eq!(points.len(), 1); points[0].sum().round() as u64 } _ => panic!("unexpected histogram aggregation"), }, _ => panic!("unexpected metric data type"), }}fn skill_message(text: &str) -> ResponseItem { ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: text.to_string(), }], phase: None, metadata: None, }}#[tokio::test]async fn regular_turn_emits_turn_started_with_trace_id_without_waiting_for_startup_prewarm() { let _trace_test_context = install_test_tracing("codex-core-tests"); let request_parent = W3cTraceContext { traceparent: Some("00-00000000000000000000000000000011-0000000000000022-01".into()), tracestate: Some("vendor=value".into()), }; let request_span = info_span!("app_server.request"); assert!(set_parent_from_w3c_trace_context( &request_span, &request_parent )); let (sess, tc, rx) = make_session_and_context_with_rx() .instrument(request_span) .await; assert_eq!( tc.trace_id.as_deref(), Some("00000000000000000000000000000011") ); let (_tx, startup_prewarm_rx) = tokio::sync::oneshot::channel::<()>(); let handle = tokio::spawn(async move { let _ = startup_prewarm_rx.await; Ok(test_model_client_session()) }); sess.set_session_startup_prewarm( crate::session_startup_prewarm::SessionStartupPrewarmHandle::new( handle, std::time::Instant::now(), crate::client::WEBSOCKET_CONNECT_TIMEOUT, ), ) .await; sess.spawn_task( Arc::clone(&tc), Vec::new(), crate::tasks::RegularTask::new(), ) .await; let first = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv()) .await .expect("expected turn started event without waiting for startup prewarm") .expect("channel open"); let EventMsg::TurnStarted(turn_started) = first.msg else { panic!("expected turn started event"); }; assert_eq!(turn_started.turn_id, tc.sub_id); assert_eq!(turn_started.trace_id, tc.trace_id); sess.abort_all_tasks(TurnAbortReason::Interrupted).await;}#[tokio::test]async fn request_mcp_server_elicitation_auto_accepts_when_auto_deny_is_enabled() { let (session, turn_context, rx) = make_session_and_context_with_rx().await; session .services .mcp_connection_manager .load_full() .set_elicitations_auto_deny(/*auto_deny*/ true); let requested_schema: McpElicitationSchema = serde_json::from_value(json!({ "type": "object", "properties": {}, })) .expect("schema should deserialize"); let response = session .request_mcp_server_elicitation( turn_context.as_ref(), RequestId::String("request-1".into()), McpServerElicitationRequestParams { thread_id: session.thread_id.to_string(), turn_id: Some(turn_context.sub_id.clone()), server_name: "codex_apps".to_string(), request: McpServerElicitationRequest::Form { meta: None, message: "Allow this request?".to_string(), requested_schema, }, }, ) .await; assert_eq!( response.response, Some(ElicitationResponse { action: ElicitationAction::Accept, content: Some(json!({})), meta: None, }) ); assert!(!response.sent); assert!(rx.try_recv().is_err());}#[tokio::test]async fn interrupting_regular_turn_waiting_on_startup_prewarm_emits_turn_aborted() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let (_tx, startup_prewarm_rx) = tokio::sync::oneshot::channel::<()>(); let handle = tokio::spawn(async move { let _ = startup_prewarm_rx.await; Ok(test_model_client_session()) }); sess.set_session_startup_prewarm( crate::session_startup_prewarm::SessionStartupPrewarmHandle::new( handle, std::time::Instant::now(), crate::client::WEBSOCKET_CONNECT_TIMEOUT, ), ) .await; sess.spawn_task( Arc::clone(&tc), Vec::new(), crate::tasks::RegularTask::new(), ) .await; let first = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv()) .await .expect("expected turn started event without waiting for startup prewarm") .expect("channel open"); assert!(matches!( first.msg, EventMsg::TurnStarted(TurnStartedEvent { turn_id, .. }) if turn_id == tc.sub_id )); sess.abort_all_tasks(TurnAbortReason::Interrupted).await; let marker_evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("expected turn aborted marker event") .expect("channel open"); assert!(matches!(marker_evt.msg, EventMsg::RawResponseItem(_))); let second = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("expected turn aborted event") .expect("channel open"); let EventMsg::TurnAborted(TurnAbortedEvent { turn_id, reason, completed_at, duration_ms, }) = second.msg else { panic!("expected turn aborted event"); }; assert_eq!(turn_id, Some(tc.sub_id.clone())); assert_eq!(reason, TurnAbortReason::Interrupted); assert!(completed_at.is_some()); assert!(duration_ms.is_some());}fn test_model_client_session() -> crate::client::ModelClientSession { let thread_id = ThreadId::try_from("00000000-0000-4000-8000-000000000001") .expect("test thread id should be valid"); crate::client::ModelClient::new( /*auth_manager*/ None, thread_id, ModelProviderInfo::create_openai_provider(/* base_url */ /*base_url*/ None), codex_protocol::protocol::SessionSource::Exec, /*model_verbosity*/ None, /*enable_request_compression*/ false, /*include_timing_metrics*/ false, /*beta_features_header*/ None, /*attestation_provider*/ None, ) .new_session()}fn developer_input_texts(items: &[ResponseItem]) -> Vec<&str> { items .iter() .filter_map(|item| match item { ResponseItem::Message { role, content, .. } if role == "developer" => { Some(content.as_slice()) } _ => None, }) .flat_map(|content| content.iter()) .filter_map(|item| match item { ContentItem::InputText { text } => Some(text.as_str()), _ => None, }) .collect()}fn developer_message_texts(items: &[ResponseItem]) -> Vec<Vec<&str>> { items .iter() .filter_map(|item| match item { ResponseItem::Message { role, content, .. } if role == "developer" => { Some(content.as_slice()) } _ => None, }) .map(|content| { content .iter() .filter_map(|item| match item { ContentItem::InputText { text } => Some(text.as_str()), _ => None, }) .collect() }) .collect()}fn user_input_texts(items: &[ResponseItem]) -> Vec<&str> { items .iter() .filter_map(|item| match item { ResponseItem::Message { role, content, .. } if role == "user" => { Some(content.as_slice()) } _ => None, }) .flat_map(|content| content.iter()) .filter_map(|item| match item { ContentItem::InputText { text } => Some(text.as_str()), _ => None, }) .collect()}fn write_project_hooks(dot_codex: &Path) -> std::io::Result<()> { std::fs::create_dir_all(dot_codex)?; std::fs::write( dot_codex.join("hooks.json"), r#"{ "hooks": { "SessionStart": [ { "hooks": [ { "type": "command", "command": "echo hello from hook" } ] } ] }}"#, )}async fn write_project_trust_config( codex_home: &Path, trusted_projects: &[(&Path, TrustLevel)],) -> std::io::Result<()> { tokio::fs::write( codex_home.join(codex_config::CONFIG_TOML_FILE), toml::to_string(&ConfigToml { projects: Some( trusted_projects .iter() .map(|(project, trust_level)| { ( project_trust_key(project), ProjectConfig { trust_level: Some(*trust_level), }, ) }) .collect::<std::collections::HashMap<_, _>>(), ), ..Default::default() }) .expect("serialize config"), ) .await}async fn preview_session_start_hooks( config: &crate::config::Config,) -> std::io::Result<Vec<codex_protocol::protocol::HookRunSummary>> { let hooks = Hooks::new(HooksConfig { feature_enabled: true, config_layer_stack: Some(config.config_layer_stack.clone()), ..HooksConfig::default() }); Ok( hooks.preview_session_start(&codex_hooks::SessionStartRequest { session_id: ThreadId::new(), cwd: config.cwd.clone(), transcript_path: None, model: "gpt-5.2".to_string(), permission_mode: "default".to_string(), target: codex_hooks::StartHookTarget::SessionStart { source: codex_hooks::SessionStartSource::Startup, }, }), )}fn test_tool_runtime(session: Arc<Session>, turn_context: Arc<TurnContext>) -> ToolCallRuntime { let router = Arc::new(ToolRouter::from_turn_context( &turn_context, crate::tools::router::ToolRouterParams { mcp_tools: None, deferred_mcp_tools: None, discoverable_tools: None, extension_tool_executors: Vec::new(), dynamic_tools: turn_context.dynamic_tools.as_slice(), }, &Default::default(), )); let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); ToolCallRuntime::new(router, session, turn_context, tracker)}fn make_connector(id: &str, name: &str) -> AppInfo { AppInfo { id: id.to_string(), name: name.to_string(), description: None, logo_url: None, logo_url_dark: None, distribution_channel: None, branding: None, app_metadata: None, labels: None, install_url: None, is_accessible: true, is_enabled: true, plugin_display_names: Vec::new(), }}#[test]fn assistant_message_stream_parsers_can_be_seeded_from_output_item_added_text() { let mut parsers = AssistantMessageStreamParsers::new(/*plan_mode*/ false); let item_id = "msg-1"; let seeded = parsers.seed_item_text(item_id, "hello <oai-mem-citation>doc"); let parsed = parsers.parse_delta(item_id, "1</oai-mem-citation> world"); let tail = parsers.finish_item(item_id); assert_eq!(seeded.visible_text, "hello "); assert_eq!(seeded.citations, Vec::<String>::new()); assert_eq!(parsed.visible_text, " world"); assert_eq!(parsed.citations, vec!["doc1".to_string()]); assert_eq!(tail.visible_text, ""); assert_eq!(tail.citations, Vec::<String>::new());}#[test]fn assistant_message_stream_parsers_seed_buffered_prefix_stays_out_of_finish_tail() { let mut parsers = AssistantMessageStreamParsers::new(/*plan_mode*/ false); let item_id = "msg-1"; let seeded = parsers.seed_item_text(item_id, "hello <oai-mem-"); let parsed = parsers.parse_delta(item_id, "citation>doc</oai-mem-citation> world"); let tail = parsers.finish_item(item_id); assert_eq!(seeded.visible_text, "hello "); assert_eq!(seeded.citations, Vec::<String>::new()); assert_eq!(parsed.visible_text, " world"); assert_eq!(parsed.citations, vec!["doc".to_string()]); assert_eq!(tail.visible_text, ""); assert_eq!(tail.citations, Vec::<String>::new());}#[test]fn assistant_message_stream_parsers_seed_plan_parser_across_added_and_delta_boundaries() { let mut parsers = AssistantMessageStreamParsers::new(/*plan_mode*/ true); let item_id = "msg-1"; let seeded = parsers.seed_item_text(item_id, "Intro\n<proposed"); let parsed = parsers.parse_delta(item_id, "_plan>\n- step\n</proposed_plan>\nOutro"); let tail = parsers.finish_item(item_id); assert_eq!(seeded.visible_text, "Intro\n"); assert_eq!( seeded.plan_segments, vec![ProposedPlanSegment::Normal("Intro\n".to_string())] ); assert_eq!(parsed.visible_text, "Outro"); assert_eq!( parsed.plan_segments, vec![ ProposedPlanSegment::ProposedPlanStart, ProposedPlanSegment::ProposedPlanDelta("- step\n".to_string()), ProposedPlanSegment::ProposedPlanEnd, ProposedPlanSegment::Normal("Outro".to_string()), ] ); assert_eq!(tail.visible_text, ""); assert!(tail.plan_segments.is_empty());}#[test]fn validated_network_policy_amendment_host_allows_normalized_match() { let amendment = NetworkPolicyAmendment { host: "ExAmPlE.Com.:443".to_string(), action: NetworkPolicyRuleAction::Allow, }; let context = NetworkApprovalContext { host: "example.com".to_string(), protocol: NetworkApprovalProtocol::Https, }; let host = Session::validated_network_policy_amendment_host(&amendment, &context) .expect("normalized hosts should match"); assert_eq!(host, "example.com");}#[test]fn validated_network_policy_amendment_host_rejects_mismatch() { let amendment = NetworkPolicyAmendment { host: "evil.example.com".to_string(), action: NetworkPolicyRuleAction::Deny, }; let context = NetworkApprovalContext { host: "api.example.com".to_string(), protocol: NetworkApprovalProtocol::Https, }; let err = Session::validated_network_policy_amendment_host(&amendment, &context) .expect_err("mismatched hosts should be rejected"); let message = err.to_string(); assert!(message.contains("does not match approved host"));}#[tokio::test]async fn start_managed_network_proxy_applies_execpolicy_network_rules() -> anyhow::Result<()> { let permission_profile = PermissionProfile::workspace_write(); let spec = crate::config::NetworkProxySpec::from_config_and_constraints( NetworkProxyConfig::default(), /*requirements*/ None, &permission_profile, )?; let mut exec_policy = Policy::empty(); exec_policy.add_network_rule( "example.com", NetworkRuleProtocol::Https, Decision::Allow, /*justification*/ None, )?; let (started_proxy, _) = Session::start_managed_network_proxy( &spec, &exec_policy, &permission_profile, /*network_policy_decider*/ None, /*blocked_request_observer*/ None, /*managed_network_requirements_enabled*/ false, crate::config::NetworkProxyAuditMetadata::default(), ) .await?; let current_cfg = started_proxy.proxy().current_cfg().await?; assert_eq!( current_cfg.network.allowed_domains(), Some(vec!["example.com".to_string()]) ); Ok(())}#[tokio::test]async fn start_managed_network_proxy_ignores_invalid_execpolicy_network_rules() -> anyhow::Result<()>{ let permission_profile = PermissionProfile::workspace_write(); let spec = crate::config::NetworkProxySpec::from_config_and_constraints( NetworkProxyConfig::default(), Some(NetworkConstraints { domains: Some(NetworkDomainPermissionsToml { entries: std::collections::BTreeMap::from([( "managed.example.com".to_string(), NetworkDomainPermissionToml::Allow, )]), }), managed_allowed_domains_only: Some(true), ..Default::default() }), &permission_profile, )?; let mut exec_policy = Policy::empty(); exec_policy.add_network_rule( "example.com", NetworkRuleProtocol::Https, Decision::Allow, /*justification*/ None, )?; let (started_proxy, _) = Session::start_managed_network_proxy( &spec, &exec_policy, &permission_profile, /*network_policy_decider*/ None, /*blocked_request_observer*/ None, /*managed_network_requirements_enabled*/ false, crate::config::NetworkProxyAuditMetadata::default(), ) .await?; let current_cfg = started_proxy.proxy().current_cfg().await?; assert_eq!( current_cfg.network.allowed_domains(), Some(vec!["managed.example.com".to_string()]) ); Ok(())}#[tokio::test]async fn managed_network_proxy_decider_survives_full_access_start() -> anyhow::Result<()> { let full_access_permission_profile = PermissionProfile::Disabled; let spec = crate::config::NetworkProxySpec::from_config_and_constraints( NetworkProxyConfig::default(), Some(NetworkConstraints { enabled: Some(true), ..Default::default() }), &full_access_permission_profile, )?; let exec_policy = Policy::empty(); let decider_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let network_policy_decider: Arc<dyn codex_network_proxy::NetworkPolicyDecider> = Arc::new({ let decider_calls = Arc::clone(&decider_calls); move |_request| { decider_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); async { codex_network_proxy::NetworkDecision::ask("not_allowed") } } }); let (started_proxy, _) = Session::start_managed_network_proxy( &spec, &exec_policy, &full_access_permission_profile, Some(network_policy_decider), /*blocked_request_observer*/ None, /*managed_network_requirements_enabled*/ true, crate::config::NetworkProxyAuditMetadata::default(), ) .await?; let spec = spec.recompute_for_permission_profile(&PermissionProfile::workspace_write())?; spec.apply_to_started_proxy(&started_proxy).await?; let current_cfg = started_proxy.proxy().current_cfg().await?; assert_eq!(current_cfg.network.allowed_domains(), None); use tokio::io::AsyncReadExt as _; use tokio::io::AsyncWriteExt as _; let mut stream = tokio::net::TcpStream::connect(started_proxy.proxy().http_addr()).await?; stream .write_all( b"GET http://example.com/ HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n", ) .await?; let mut buffer = [0_u8; 4096]; let bytes_read = tokio::time::timeout(StdDuration::from_secs(2), stream.read(&mut buffer)) .await .expect("timed out waiting for proxy response")?; let response = String::from_utf8_lossy(&buffer[..bytes_read]); assert!( response.starts_with("HTTP/1.1 403 Forbidden"), "unexpected proxy response: {response}" ); assert!( response.contains("x-proxy-error: blocked-by-allowlist"), "unexpected proxy response: {response}" ); assert_eq!( decider_calls.load(std::sync::atomic::Ordering::SeqCst), 1, "unexpected proxy response: {response}" ); Ok(())}#[tokio::test]async fn new_turn_refreshes_managed_network_proxy_for_sandbox_change() -> anyhow::Result<()> { let (session, _turn_context) = make_session_and_context().await; let initial_permission_profile = PermissionProfile::workspace_write(); let mut network_config = NetworkProxyConfig::default(); network_config .network .set_allowed_domains(vec!["evil.com".to_string()]); let requirements = NetworkConstraints { domains: Some(NetworkDomainPermissionsToml { entries: std::collections::BTreeMap::from([( "*.example.com".to_string(), NetworkDomainPermissionToml::Allow, )]), }), ..Default::default() }; let spec = crate::config::NetworkProxySpec::from_config_and_constraints( network_config, Some(requirements), &initial_permission_profile, )?; let (started_proxy, _) = Session::start_managed_network_proxy( &spec, &Policy::empty(), &initial_permission_profile, /*network_policy_decider*/ None, /*blocked_request_observer*/ None, /*managed_network_requirements_enabled*/ false, crate::config::NetworkProxyAuditMetadata::default(), ) .await?; assert_eq!( started_proxy .proxy() .current_cfg() .await? .network .allowed_domains(), Some(vec!["*.example.com".to_string(), "evil.com".to_string()]) ); { let mut state = session.state.lock().await; let mut config = (*state.session_configuration.original_config_do_not_use).clone(); config.permissions.network = Some(spec); config .permissions .set_permission_profile(initial_permission_profile.clone()) .expect("test setup should allow permission profile"); state.session_configuration.original_config_do_not_use = Arc::new(config); state .session_configuration .set_permission_profile_for_tests(initial_permission_profile) .expect("test setup should allow permission profile"); } session .services .network_proxy .store(Some(Arc::new(started_proxy))); session .new_turn_with_sub_id( "sandbox-policy-change".to_string(), SessionSettingsUpdate { sandbox_policy: Some(SandboxPolicy::DangerFullAccess), ..Default::default() }, ) .await?; let started_proxy = session .services .network_proxy .load_full() .expect("managed network proxy should be present"); assert_eq!( started_proxy .proxy() .current_cfg() .await? .network .allowed_domains(), Some(vec!["*.example.com".to_string()]) ); Ok(())}#[tokio::test]async fn danger_full_access_turns_do_not_expose_managed_network_proxy() -> anyhow::Result<()> { let network_spec = crate::config::NetworkProxySpec::from_config_and_constraints( NetworkProxyConfig::default(), Some(NetworkConstraints { enabled: Some(true), ..Default::default() }), &PermissionProfile::Disabled, )?; let session = make_session_with_config(move |config| { config .permissions .set_permission_profile(PermissionProfile::Disabled) .expect("test setup should allow permission profile"); config.permissions.network = Some(network_spec); }) .await?; let turn_context = session.new_default_turn().await; assert!(turn_context.network.is_none()); Ok(())}#[tokio::test]async fn danger_full_access_tool_attempts_do_not_enforce_managed_network() -> anyhow::Result<()> { #[derive(Default)] struct ProbeToolRuntime { enforce_managed_network: Vec<bool>, } impl crate::tools::sandboxing::Approvable<()> for ProbeToolRuntime { type ApprovalKey = String; fn approval_keys(&self, _req: &()) -> Vec<Self::ApprovalKey> { vec!["probe".to_string()] } fn start_approval_async<'a>( &'a mut self, _req: &'a (), _ctx: crate::tools::sandboxing::ApprovalCtx<'a>, ) -> futures::future::BoxFuture<'a, ReviewDecision> { Box::pin(async { ReviewDecision::Approved }) } } impl crate::tools::sandboxing::Sandboxable for ProbeToolRuntime { fn sandbox_preference(&self) -> codex_sandboxing::SandboxablePreference { codex_sandboxing::SandboxablePreference::Auto } } impl crate::tools::sandboxing::ToolRuntime<(), ()> for ProbeToolRuntime { async fn run( &mut self, _req: &(), attempt: &crate::tools::sandboxing::SandboxAttempt<'_>, _ctx: &crate::tools::sandboxing::ToolCtx, ) -> Result<(), crate::tools::sandboxing::ToolError> { self.enforce_managed_network .push(attempt.enforce_managed_network); Ok(()) } } let network_spec = crate::config::NetworkProxySpec::from_config_and_constraints( NetworkProxyConfig::default(), Some(NetworkConstraints { enabled: Some(true), ..Default::default() }), &PermissionProfile::Disabled, )?; let session = make_session_with_config(move |config| { config .permissions .set_permission_profile(PermissionProfile::Disabled) .expect("test setup should allow permission profile"); config.permissions.network = Some(network_spec); let layers = config .config_layer_stack .get_layers( ConfigLayerStackOrdering::LowestPrecedenceFirst, /*include_disabled*/ true, ) .into_iter() .cloned() .collect(); let mut requirements = config.config_layer_stack.requirements().clone(); requirements.network = Some(Sourced::new( NetworkConstraints { enabled: Some(true), ..Default::default() }, RequirementSource::LegacyManagedConfigTomlFromMdm, )); let mut requirements_toml = config.config_layer_stack.requirements_toml().clone(); requirements_toml.network = Some(codex_config::NetworkRequirementsToml { enabled: Some(true), ..Default::default() }); config.config_layer_stack = ConfigLayerStack::new(layers, requirements, requirements_toml) .expect("rebuild config layer stack with network requirements"); }) .await?; let turn = session.new_default_turn().await; assert!(turn.network.is_none()); let mut orchestrator = crate::tools::orchestrator::ToolOrchestrator::new(); let mut tool = ProbeToolRuntime::default(); let tool_ctx = crate::tools::sandboxing::ToolCtx { session: Arc::clone(&session), turn: Arc::clone(&turn), call_id: "probe-call".to_string(), tool_name: codex_tools::ToolName::plain("probe"), }; orchestrator .run( &mut tool, &(), &tool_ctx, turn.as_ref(), AskForApproval::Never, ) .await .expect("probe runtime should succeed"); assert_eq!(tool.enforce_managed_network, vec![false]); Ok(())}#[tokio::test]async fn workspace_write_turns_continue_to_expose_managed_network_proxy() -> anyhow::Result<()> { let permission_profile = PermissionProfile::workspace_write(); let network_spec = crate::config::NetworkProxySpec::from_config_and_constraints( NetworkProxyConfig::default(), Some(NetworkConstraints { enabled: Some(true), ..Default::default() }), &permission_profile, )?; let session = make_session_with_config(move |config| { config .permissions .set_permission_profile(permission_profile) .expect("test setup should allow permission profile"); config.permissions.network = Some(network_spec); }) .await?; let turn_context = session.new_default_turn().await; assert!(turn_context.network.is_some()); Ok(())}#[tokio::test]async fn user_shell_commands_do_not_inherit_managed_network_proxy() -> anyhow::Result<()> { let permission_profile = PermissionProfile::workspace_write(); let network_spec = crate::config::NetworkProxySpec::from_config_and_constraints( NetworkProxyConfig::default(), Some(NetworkConstraints { enabled: Some(true), ..Default::default() }), &permission_profile, )?; let (session, rx) = make_session_with_config_and_rx(move |config| { config .permissions .set_permission_profile(permission_profile) .expect("test setup should allow permission profile"); config.permissions.network = Some(network_spec); }) .await?; let turn_context = session.new_default_turn().await; assert!(turn_context.network.is_some()); #[cfg(windows)] let command = r#"$val = $env:HTTP_PROXY; if ([string]::IsNullOrEmpty($val)) { $val = 'not-set' } ; [System.Console]::Write($val)"#.to_string(); #[cfg(not(windows))] let command = r#"sh -c "printf '%s' \"${HTTP_PROXY:-not-set}\"""#.to_string(); execute_user_shell_command( Arc::clone(&session), turn_context, command, CancellationToken::new(), UserShellCommandMode::StandaloneTurn, ) .await; loop { let event = rx.recv().await.expect("channel open"); if let EventMsg::ExecCommandEnd(event) = event.msg { assert_eq!(event.exit_code, 0); assert_eq!(event.stdout.trim(), "not-set"); break; } } Ok(())}#[tokio::test]async fn get_base_instructions_no_user_content() { let prompt_with_apply_patch_instructions = include_str!("../../prompt_with_apply_patch_instructions.md"); let models_response = bundled_models_response() .unwrap_or_else(|err| panic!("bundled models.json should parse: {err}")); let model_info_for_slug = |slug: &str, config: &Config| { let model = models_response .models .iter() .find(|candidate| candidate.slug == slug) .cloned() .unwrap_or_else(|| panic!("model slug {slug} is missing from models.json")); model_info::with_config_overrides(model, &config.to_models_manager_config()) }; let test_cases = vec![ InstructionsTestCase { slug: "gpt-5.4", expects_apply_patch_description: false, }, InstructionsTestCase { slug: "gpt-5.4-mini", expects_apply_patch_description: false, }, InstructionsTestCase { slug: "gpt-5.3-codex", expects_apply_patch_description: false, }, InstructionsTestCase { slug: "gpt-5.2", expects_apply_patch_description: false, }, ]; let (session, _turn_context) = make_session_and_context().await; let config = test_config().await; for test_case in test_cases { let model_info = model_info_for_slug(test_case.slug, &config); if test_case.expects_apply_patch_description { assert_eq!( model_info.base_instructions.as_str(), prompt_with_apply_patch_instructions ); } { let mut state = session.state.lock().await; state.session_configuration.base_instructions = model_info.base_instructions.clone(); } let base_instructions = session.get_base_instructions().await; assert_eq!(base_instructions.text, model_info.base_instructions); }}#[tokio::test]async fn reload_user_config_layer_updates_effective_apps_config() { let (session, _turn_context) = make_session_and_context().await; let codex_home = session.codex_home().await; std::fs::create_dir_all(&codex_home).expect("create codex home"); let config_toml_path = codex_home.join(CONFIG_TOML_FILE); std::fs::write( &config_toml_path, "[apps.calendar]\nenabled = false\ndestructive_enabled = false\n", ) .expect("write user config"); session.reload_user_config_layer().await; let config = session.get_config().await; let apps_toml = config .config_layer_stack .effective_config() .as_table() .and_then(|table| table.get("apps")) .cloned() .expect("apps table"); let apps = codex_config::types::AppsConfigToml::deserialize(apps_toml) .expect("deserialize apps config"); let app = apps .apps .get("calendar") .expect("calendar app config exists"); assert!(!app.enabled); assert_eq!(app.destructive_enabled, Some(false));}#[tokio::test]async fn reload_user_config_layer_updates_base_and_selected_profile_layers() { let (session, _turn_context) = make_session_and_context().await; let codex_home = session.codex_home().await; std::fs::create_dir_all(&codex_home).expect("create codex home"); let base_config_path = codex_home.join(CONFIG_TOML_FILE); let profile_config_path = codex_home.join("work.config.toml"); std::fs::write( &base_config_path, "model = \"base\"\napproval_policy = \"on-failure\"\n", ) .expect("write base user config"); std::fs::write(&profile_config_path, "model = \"profile-old\"\n") .expect("write profile user config"); let config = ConfigBuilder::without_managed_config_for_tests() .codex_home(codex_home.to_path_buf()) .loader_overrides(LoaderOverrides { user_config_path: Some(profile_config_path.abs()), user_config_profile: Some("work".parse().expect("profile-v2 name")), ..LoaderOverrides::without_managed_config_for_tests() }) .build() .await .expect("load profile config"); { let mut state = session.state.lock().await; state.session_configuration.original_config_do_not_use = Arc::new(config); } std::fs::write( &base_config_path, "model = \"base\"\napproval_policy = \"never\"\n", ) .expect("update base user config"); std::fs::write(&profile_config_path, "model = \"profile-new\"\n") .expect("update profile user config"); session.reload_user_config_layer().await; let config = session.get_config().await; assert_eq!( config .config_layer_stack .get_user_config_file() .map(codex_utils_absolute_path::AbsolutePathBuf::as_path), Some(profile_config_path.as_path()) ); let effective_user_config = config .config_layer_stack .effective_user_config() .expect("merged user config"); assert_eq!( effective_user_config .get("model") .and_then(toml::Value::as_str), Some("profile-new") ); assert_eq!( effective_user_config .get("approval_policy") .and_then(toml::Value::as_str), Some("never") );}#[tokio::test]async fn reload_user_config_layer_refreshes_hooks() -> anyhow::Result<()> { let session = make_session_with_config(|config| { config .features .enable(Feature::CodexHooks) .expect("enable Codex hooks"); }) .await?; let codex_home = session.codex_home().await; std::fs::create_dir_all(&codex_home)?; let config_toml_path = codex_home.join(CONFIG_TOML_FILE); let user_config: codex_config::TomlValue = serde_json::from_value(serde_json::json!({ "hooks": { "SessionStart": [{ "hooks": [{ "type": "command", "command": "python3 /tmp/user.py", }], }], }, }))?; let request = codex_hooks::SessionStartRequest { session_id: session.thread_id, cwd: session.get_config().await.cwd.clone(), transcript_path: None, model: "gpt-5.2".to_string(), permission_mode: "default".to_string(), target: codex_hooks::StartHookTarget::SessionStart { source: codex_hooks::SessionStartSource::Startup, }, }; assert!(session.hooks().preview_session_start(&request).is_empty()); let config = session.get_config().await; let hook_list = codex_hooks::list_hooks(codex_hooks::HooksConfig { feature_enabled: true, config_layer_stack: Some( config .config_layer_stack .with_user_config(&config_toml_path, user_config.clone()), ), ..codex_hooks::HooksConfig::default() }); assert_eq!(hook_list.hooks.len(), 1); assert_eq!( hook_list.hooks[0].trust_status, codex_protocol::protocol::HookTrustStatus::Untrusted ); let trusted_user_config: codex_config::TomlValue = serde_json::from_value(serde_json::json!({ "hooks": { "SessionStart": [{ "hooks": [{ "type": "command", "command": "python3 /tmp/user.py", }], }], "state": { hook_list.hooks[0].key.clone(): { "trusted_hash": hook_list.hooks[0].current_hash.clone(), }, }, }, }))?; std::fs::write(&config_toml_path, toml::to_string(&trusted_user_config)?)?; session.reload_user_config_layer().await; assert_eq!(session.hooks().preview_session_start(&request).len(), 1); Ok(())}#[tokio::test]async fn refresh_runtime_config_refreshes_hooks() -> anyhow::Result<()> { let (session, _turn_context) = make_session_and_context().await; { let mut state = session.state.lock().await; let mut config = (*state.session_configuration.original_config_do_not_use).clone(); config .features .enable(Feature::CodexHooks) .expect("enable Codex hooks"); state.session_configuration.original_config_do_not_use = Arc::new(config); } let codex_home = session.codex_home().await; std::fs::create_dir_all(&codex_home)?; let config_toml_path = codex_home.join(CONFIG_TOML_FILE); #[derive(serde::Serialize)] struct NormalizedHookIdentity { event_name: &'static str, #[serde(flatten)] group: codex_config::MatcherGroup, } let trusted_hash = { let identity = NormalizedHookIdentity { event_name: "session_start", group: codex_config::MatcherGroup { matcher: None, hooks: vec![codex_config::HookHandlerConfig::Command { command: "python3 /tmp/user.py".to_string(), command_windows: None, timeout_sec: Some(600), r#async: false, status_message: None, }], }, }; let identity = codex_config::TomlValue::try_from(identity)?; codex_config::version_for_toml(&identity) }; let hook_key = format!("{}:session_start:0:0", config_toml_path.display()); let trusted_user_config: codex_config::TomlValue = serde_json::from_value(serde_json::json!({ "hooks": { "SessionStart": [{ "hooks": [{ "type": "command", "command": "python3 /tmp/user.py", }], }], "state": { hook_key: { "trusted_hash": trusted_hash, }, }, }, }))?; std::fs::write(&config_toml_path, toml::to_string(&trusted_user_config)?)?; let request = codex_hooks::SessionStartRequest { session_id: session.thread_id, cwd: session.get_config().await.cwd.clone(), transcript_path: None, model: "gpt-5.2".to_string(), permission_mode: "default".to_string(), target: codex_hooks::StartHookTarget::SessionStart { source: codex_hooks::SessionStartSource::Startup, }, }; assert!(session.hooks().preview_session_start(&request).is_empty()); let next_config = load_latest_config_for_session(&session).await; session.refresh_runtime_config(next_config).await; assert_eq!(session.hooks().preview_session_start(&request).len(), 1); Ok(())}#[tokio::test]async fn reload_user_config_layer_updates_effective_tool_suggest_config() { let (session, _turn_context) = make_session_and_context().await; let codex_home = session.codex_home().await; std::fs::create_dir_all(&codex_home).expect("create codex home"); let config_toml_path = codex_home.join(CONFIG_TOML_FILE); std::fs::write( &config_toml_path, r#"[tool_suggest]disabled_tools = [ { type = "connector", id = " calendar " }, { type = "plugin", id = "slack@openai-curated" },]"#, ) .expect("write user config"); session.reload_user_config_layer().await; let config = session.get_config().await; assert_eq!( config.tool_suggest.disabled_tools, vec![ ToolSuggestDisabledTool::connector("calendar"), ToolSuggestDisabledTool::plugin("slack@openai-curated"), ] );}#[tokio::test]async fn refresh_runtime_config_updates_runtime_refreshable_fields_and_keeps_session_static_settings() { let (session, _turn_context) = make_session_and_context().await; let codex_home = session.codex_home().await; std::fs::create_dir_all(&codex_home).expect("create codex home"); std::fs::write( codex_home.join(CONFIG_TOML_FILE), r#"[apps.calendar]enabled = falsedestructive_enabled = false[tool_suggest]disabled_tools = [ { type = "connector", id = " calendar " }, { type = "plugin", id = "slack@openai-curated" },]"#, ) .expect("write user config"); let original = session.get_config().await; let mut next_config = load_latest_config_for_session(&session).await; next_config.model = Some("gpt-5.4".to_string()); next_config.notify = Some(vec!["echo".to_string()]); session.refresh_runtime_config(next_config).await; let config = session.get_config().await; let apps_toml = config .config_layer_stack .effective_config() .as_table() .and_then(|table| table.get("apps")) .cloned() .expect("apps table"); let apps = codex_config::types::AppsConfigToml::deserialize(apps_toml) .expect("deserialize apps config"); let app = apps .apps .get("calendar") .expect("calendar app config exists"); assert!(!app.enabled); assert_eq!(app.destructive_enabled, Some(false)); assert_eq!(config.model, original.model); assert_eq!(config.notify, original.notify); assert_eq!( config.tool_suggest.disabled_tools, vec![ ToolSuggestDisabledTool::connector("calendar"), ToolSuggestDisabledTool::plugin("slack@openai-curated"), ] );}#[test]fn collect_explicit_app_ids_from_skill_items_includes_linked_mentions() { let connectors = vec![make_connector("calendar", "Calendar")]; let skill_items = vec\n</skill>", )]; let connector_ids = collect_explicit_app_ids_from_skill_items(&skill_items, &connectors, &HashMap::new()); assert_eq!(connector_ids, HashSet::from(["calendar".to_string()]));}#[test]fn collect_explicit_app_ids_from_skill_items_resolves_unambiguous_plain_mentions() { let connectors = vec![make_connector("calendar", "Calendar")]; let skill_items = vec![skill_message( "<skill>\n<name>demo</name>\n<path>/tmp/skills/demo/SKILL.md</path>\nuse $calendar\n</skill>", )]; let connector_ids = collect_explicit_app_ids_from_skill_items(&skill_items, &connectors, &HashMap::new()); assert_eq!(connector_ids, HashSet::from(["calendar".to_string()]));}#[test]fn collect_explicit_app_ids_from_skill_items_skips_plain_mentions_with_skill_conflicts() { let connectors = vec![make_connector("calendar", "Calendar")]; let skill_items = vec![skill_message( "<skill>\n<name>demo</name>\n<path>/tmp/skills/demo/SKILL.md</path>\nuse $calendar\n</skill>", )]; let skill_name_counts_lower = HashMap::from([("calendar".to_string(), 1)]); let connector_ids = collect_explicit_app_ids_from_skill_items( &skill_items, &connectors, &skill_name_counts_lower, ); assert_eq!(connector_ids, HashSet::<String>::new());}#[tokio::test]async fn reconstruct_history_matches_live_compactions() { let (session, turn_context) = make_session_and_context().await; let (rollout_items, expected) = sample_rollout(&session, &turn_context).await; let reconstruction_turn = session.new_default_turn().await; let reconstructed = session .reconstruct_history_from_rollout(reconstruction_turn.as_ref(), &rollout_items) .await; assert_eq!(expected, reconstructed.history); assert_eq!(2, reconstructed.window_id);}#[tokio::test]async fn reconstruct_history_uses_replacement_history_verbatim() { let (session, turn_context) = make_session_and_context().await; let summary_item = ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "summary".to_string(), }], phase: None, metadata: Some(ResponseItemMetadata { turn_id: Some("compact-turn".to_string()), }), }; let replacement_history = vec![ summary_item.clone(), ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ContentItem::InputText { text: "stale developer instructions".to_string(), }], phase: None, metadata: None, }, ]; let rollout_items = vec![RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(replacement_history.clone()), window_id: Some(42), })]; let reconstructed = session .reconstruct_history_from_rollout(&turn_context, &rollout_items) .await; assert_eq!(reconstructed.history, replacement_history); assert_eq!(42, reconstructed.window_id);}#[tokio::test]async fn record_initial_history_reconstructs_resumed_transcript() { let (session, turn_context) = make_session_and_context().await; let (rollout_items, expected) = sample_rollout(&session, &turn_context).await; session .record_initial_history(InitialHistory::Resumed(ResumedHistory { conversation_id: ThreadId::default(), history: rollout_items, rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")), })) .await; let history = session.state.lock().await.clone_history(); assert_eq!(expected, history.raw_items());}#[tokio::test]async fn resize_all_images_prepares_failures_before_history_insertion() { let (session, turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx( CodexAuth::from_api_key("Test API Key"), Vec::new(), |config| { let _ = config.features.enable(Feature::ResizeAllImages); }, ) .await; let item = ResponseItem::FunctionCallOutput { call_id: "call-1".to_string(), output: FunctionCallOutputPayload { body: FunctionCallOutputBody::ContentItems(vec![ FunctionCallOutputContentItem::InputText { text: "before".to_string(), }, FunctionCallOutputContentItem::InputImage { image_url: "data:image/png;base64,%%%".to_string(), detail: Some(ImageDetail::High), }, FunctionCallOutputContentItem::InputImage { image_url: "https://example.com/image.png".to_string(), detail: Some(ImageDetail::High), }, ]), success: Some(true), }, metadata: None, }; session .record_conversation_items(turn_context.as_ref(), std::slice::from_ref(&item)) .await; let expected = vec![ResponseItem::FunctionCallOutput { call_id: "call-1".to_string(), output: FunctionCallOutputPayload { body: FunctionCallOutputBody::ContentItems(vec![ FunctionCallOutputContentItem::InputText { text: "before".to_string(), }, FunctionCallOutputContentItem::InputText { text: "image content omitted because it could not be processed".to_string(), }, FunctionCallOutputContentItem::InputImage { image_url: "https://example.com/image.png".to_string(), detail: Some(ImageDetail::High), }, ]), success: Some(true), }, metadata: None, }]; assert_eq!( session.state.lock().await.clone_history().raw_items(), expected.as_slice() );}#[tokio::test]async fn resize_all_images_prepares_resumed_history_before_installing_it() { let (session, _turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx( CodexAuth::from_api_key("Test API Key"), Vec::new(), |config| { let _ = config.features.enable(Feature::ResizeAllImages); }, ) .await; let resumed_item = ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ ContentItem::InputImage { image_url: "data:image/png;base64,%%%".to_string(), detail: Some(ImageDetail::High), }, ContentItem::InputText { text: "keep me".to_string(), }, ], phase: None, metadata: None, }; session .record_initial_history(InitialHistory::Resumed(ResumedHistory { conversation_id: ThreadId::default(), history: vec![RolloutItem::ResponseItem(resumed_item)], rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")), })) .await; assert_eq!( session.state.lock().await.clone_history().raw_items(), &[ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ ContentItem::InputText { text: "image content omitted because it could not be processed".to_string(), }, ContentItem::InputText { text: "keep me".to_string(), }, ], phase: None, metadata: None, }] );}#[test]fn resolve_multi_agent_version_handles_unset_and_legacy_history() { let thread_id = ThreadId::default(); assert_eq!( resolve_multi_agent_version( &InitialHistory::New, /*inherited_multi_agent_version*/ None ), None ); assert_eq!( resolve_multi_agent_version( &InitialHistory::Resumed(ResumedHistory { conversation_id: thread_id, history: Vec::new(), rollout_path: None, }), /*inherited_multi_agent_version*/ None, ), Some(MultiAgentVersion::V1) ); assert_eq!( resolve_multi_agent_version( &InitialHistory::Resumed(ResumedHistory { conversation_id: thread_id, history: Vec::new(), rollout_path: None, }), Some(MultiAgentVersion::V2), ), Some(MultiAgentVersion::V2) ); assert_eq!( resolve_multi_agent_version( &InitialHistory::Resumed(ResumedHistory { conversation_id: thread_id, history: vec![session_meta_item( thread_id, Some(MultiAgentVersion::Disabled) )], rollout_path: None, }), Some(MultiAgentVersion::V2), ), Some(MultiAgentVersion::Disabled) ); assert_eq!( resolve_multi_agent_version( &InitialHistory::Forked(vec![session_meta_item( thread_id, Some(MultiAgentVersion::V2) )]), Some(MultiAgentVersion::Disabled), ), Some(MultiAgentVersion::Disabled) ); assert_eq!( resolve_multi_agent_version( &InitialHistory::Forked(Vec::new()), /*inherited_multi_agent_version*/ None ), Some(MultiAgentVersion::V1) );}#[tokio::test]async fn record_initial_history_new_defers_initial_context_until_first_turn() { let (session, _turn_context) = make_session_and_context().await; session.record_initial_history(InitialHistory::New).await; let history = session.clone_history().await; assert_eq!(history.raw_items().to_vec(), Vec::<ResponseItem>::new()); assert!(session.reference_context_item().await.is_none()); assert_eq!(session.previous_turn_settings().await, None);}fn session_meta_item( thread_id: ThreadId, multi_agent_version: Option<MultiAgentVersion>,) -> RolloutItem { RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { id: thread_id, multi_agent_version, ..SessionMeta::default() }, git: None, })}#[tokio::test]async fn resumed_history_injects_initial_context_on_first_context_update_only() { let (session, turn_context) = make_session_and_context().await; let (rollout_items, mut expected) = sample_rollout(&session, &turn_context).await; session .record_initial_history(InitialHistory::Resumed(ResumedHistory { conversation_id: ThreadId::default(), history: rollout_items, rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")), })) .await; let history_before_seed = session.state.lock().await.clone_history(); assert_eq!(expected, history_before_seed.raw_items()); session .record_context_updates_and_set_reference_context_item(&turn_context) .await; let initial_context = session.build_initial_context(&turn_context).await; expected.extend(initial_context); let history_after_seed = session.clone_history().await; assert_eq!(expected, history_after_seed.raw_items()); session .record_context_updates_and_set_reference_context_item(&turn_context) .await; let history_after_second_seed = session.clone_history().await; assert_eq!( history_after_seed.raw_items(), history_after_second_seed.raw_items() );}#[tokio::test]async fn record_initial_history_seeds_token_info_from_rollout() { let (session, turn_context) = make_session_and_context().await; let (mut rollout_items, _expected) = sample_rollout(&session, &turn_context).await; let info1 = TokenUsageInfo { total_token_usage: TokenUsage { input_tokens: 10, cached_input_tokens: 0, output_tokens: 20, reasoning_output_tokens: 0, total_tokens: 30, }, last_token_usage: TokenUsage { input_tokens: 3, cached_input_tokens: 0, output_tokens: 4, reasoning_output_tokens: 0, total_tokens: 7, }, model_context_window: Some(1_000), }; let info2 = TokenUsageInfo { total_token_usage: TokenUsage { input_tokens: 100, cached_input_tokens: 50, output_tokens: 200, reasoning_output_tokens: 25, total_tokens: 375, }, last_token_usage: TokenUsage { input_tokens: 10, cached_input_tokens: 0, output_tokens: 20, reasoning_output_tokens: 5, total_tokens: 35, }, model_context_window: Some(2_000), }; rollout_items.push(RolloutItem::EventMsg(EventMsg::TokenCount( TokenCountEvent { info: Some(info1), rate_limits: None, }, ))); rollout_items.push(RolloutItem::EventMsg(EventMsg::TokenCount( TokenCountEvent { info: None, rate_limits: None, }, ))); rollout_items.push(RolloutItem::EventMsg(EventMsg::TokenCount( TokenCountEvent { info: Some(info2.clone()), rate_limits: None, }, ))); rollout_items.push(RolloutItem::EventMsg(EventMsg::TokenCount( TokenCountEvent { info: None, rate_limits: None, }, ))); session .record_initial_history(InitialHistory::Resumed(ResumedHistory { conversation_id: ThreadId::default(), history: rollout_items, rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")), })) .await; let actual = session.state.lock().await.token_info(); assert_eq!(actual, Some(info2));}#[tokio::test]async fn recompute_token_usage_uses_session_base_instructions() { let (session, turn_context) = make_session_and_context().await; let override_instructions = "SESSION_OVERRIDE_INSTRUCTIONS_ONLY".repeat(120); { let mut state = session.state.lock().await; state.session_configuration.base_instructions = override_instructions.clone(); } let item = user_message("hello"); session .record_conversation_items(&turn_context, std::slice::from_ref(&item)) .await; let history = session.clone_history().await; let session_base_instructions = BaseInstructions { text: override_instructions, }; let expected_tokens = history .estimate_token_count_with_base_instructions(&session_base_instructions) .expect("estimate with session base instructions"); let model_estimated_tokens = history .estimate_token_count(&turn_context) .expect("estimate with model instructions"); assert_ne!(expected_tokens, model_estimated_tokens); session.recompute_token_usage(&turn_context).await; let actual_tokens = session .state .lock() .await .token_info() .expect("token info") .last_token_usage .total_tokens; assert_eq!(actual_tokens, expected_tokens.max(0));}#[tokio::test]async fn recompute_token_usage_updates_model_context_window() { let (session, mut turn_context) = make_session_and_context().await; { let mut state = session.state.lock().await; state.set_token_info(Some(TokenUsageInfo { total_token_usage: TokenUsage::default(), last_token_usage: TokenUsage::default(), model_context_window: Some(258_400), })); } turn_context.model_info.context_window = Some(128_000); turn_context.model_info.effective_context_window_percent = 100; session.recompute_token_usage(&turn_context).await; let actual = session.state.lock().await.token_info().expect("token info"); assert_eq!(actual.model_context_window, Some(128_000));}#[tokio::test]async fn record_token_usage_info_notifies_extension_contributors() { struct SessionTokenUsageMarker; struct ThreadTokenUsageMarker; #[derive(Debug, PartialEq, Eq)] struct RecordedTokenUsage { session_level_id: String, thread_level_id: String, turn_level_id: String, token_usage: TokenUsageInfo, saw_session_store: bool, saw_thread_store: bool, } struct TokenUsageRecorder { records: Arc<std::sync::Mutex<Vec<RecordedTokenUsage>>>, } impl codex_extension_api::TokenUsageContributor for TokenUsageRecorder { fn on_token_usage<'a>( &'a self, session_store: &'a codex_extension_api::ExtensionData, thread_store: &'a codex_extension_api::ExtensionData, turn_store: &'a codex_extension_api::ExtensionData, token_usage: &'a TokenUsageInfo, ) -> codex_extension_api::ExtensionFuture<'a, ()> { Box::pin(async move { self.records .lock() .expect("token usage records lock") .push(RecordedTokenUsage { session_level_id: session_store.level_id().to_string(), thread_level_id: thread_store.level_id().to_string(), turn_level_id: turn_store.level_id().to_string(), token_usage: token_usage.clone(), saw_session_store: session_store.get::<SessionTokenUsageMarker>().is_some(), saw_thread_store: thread_store.get::<ThreadTokenUsageMarker>().is_some(), }); }) } } let (mut session, turn_context) = make_session_and_context().await; let records = Arc::new(std::sync::Mutex::new(Vec::new())); let mut builder = codex_extension_api::ExtensionRegistryBuilder::<crate::config::Config>::new(); builder.token_usage_contributor(Arc::new(TokenUsageRecorder { records: Arc::clone(&records), })); session.services.extensions = Arc::new(builder.build()); session .services .session_extension_data .insert(SessionTokenUsageMarker); session .services .thread_extension_data .insert(ThreadTokenUsageMarker); let first_usage = TokenUsage { input_tokens: 10, cached_input_tokens: 2, output_tokens: 20, reasoning_output_tokens: 3, total_tokens: 33, }; let second_usage = TokenUsage { input_tokens: 7, cached_input_tokens: 1, output_tokens: 8, reasoning_output_tokens: 5, total_tokens: 20, }; session .record_token_usage_info(&turn_context, Some(&first_usage)) .await; session .record_token_usage_info(&turn_context, Some(&second_usage)) .await; let mut expected_total_usage = first_usage.clone(); expected_total_usage.add_assign(&second_usage); let expected = vec![ RecordedTokenUsage { session_level_id: session.session_id().to_string(), thread_level_id: session.thread_id.to_string(), turn_level_id: turn_context.sub_id.clone(), token_usage: TokenUsageInfo { total_token_usage: first_usage.clone(), last_token_usage: first_usage, model_context_window: turn_context.model_context_window(), }, saw_session_store: true, saw_thread_store: true, }, RecordedTokenUsage { session_level_id: session.session_id().to_string(), thread_level_id: session.thread_id.to_string(), turn_level_id: turn_context.sub_id.clone(), token_usage: TokenUsageInfo { total_token_usage: expected_total_usage, last_token_usage: second_usage, model_context_window: turn_context.model_context_window(), }, saw_session_store: true, saw_thread_store: true, }, ]; let actual = records .lock() .expect("token usage records lock") .drain(..) .collect::<Vec<_>>(); assert_eq!(expected, actual);}#[tokio::test]async fn turn_start_lifecycle_exposes_turn_metadata_and_token_baseline() { struct SessionTurnStartMarker; struct ThreadTurnStartMarker; #[derive(Debug, PartialEq, Eq)] struct RecordedTurnStart { session_level_id: String, thread_level_id: String, turn_level_id: String, turn_id: String, collaboration_mode: CollaborationMode, token_usage_at_turn_start: TokenUsage, saw_session_store: bool, saw_thread_store: bool, } struct TurnStartRecorder { records: Arc<std::sync::Mutex<Vec<RecordedTurnStart>>>, } impl codex_extension_api::TurnLifecycleContributor for TurnStartRecorder { fn on_turn_start<'a>( &'a self, input: codex_extension_api::TurnStartInput<'a>, ) -> codex_extension_api::ExtensionFuture<'a, ()> { Box::pin(async move { self.records .lock() .expect("turn start records lock") .push(RecordedTurnStart { session_level_id: input.session_store.level_id().to_string(), thread_level_id: input.thread_store.level_id().to_string(), turn_level_id: input.turn_store.level_id().to_string(), turn_id: input.turn_id.to_string(), collaboration_mode: input.collaboration_mode.clone(), token_usage_at_turn_start: input.token_usage_at_turn_start.clone(), saw_session_store: input .session_store .get::<SessionTurnStartMarker>() .is_some(), saw_thread_store: input .thread_store .get::<ThreadTurnStartMarker>() .is_some(), }); }) } } let (mut session, turn_context) = make_session_and_context().await; let records = Arc::new(std::sync::Mutex::new(Vec::new())); let mut builder = codex_extension_api::ExtensionRegistryBuilder::<crate::config::Config>::new(); builder.turn_lifecycle_contributor(Arc::new(TurnStartRecorder { records: Arc::clone(&records), })); session.services.extensions = Arc::new(builder.build()); session .services .session_extension_data .insert(SessionTurnStartMarker); session .services .thread_extension_data .insert(ThreadTurnStartMarker); let token_usage_at_turn_start = TokenUsage { input_tokens: 100, cached_input_tokens: 40, output_tokens: 25, reasoning_output_tokens: 5, total_tokens: 130, }; set_total_token_usage(&session, token_usage_at_turn_start.clone()).await; let expected = RecordedTurnStart { session_level_id: session.session_id().to_string(), thread_level_id: session.thread_id.to_string(), turn_level_id: turn_context.sub_id.clone(), turn_id: turn_context.sub_id.clone(), collaboration_mode: turn_context.collaboration_mode.clone(), token_usage_at_turn_start, saw_session_store: true, saw_thread_store: true, }; let sess = Arc::new(session); sess.spawn_task( Arc::new(turn_context), Vec::new(), NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; let actual = records .lock() .expect("turn start records lock") .drain(..) .collect::<Vec<_>>(); assert_eq!(vec![expected], actual);}#[tokio::test]async fn turn_error_lifecycle_exposes_error_and_stores() { struct SessionTurnErrorMarker; struct ThreadTurnErrorMarker; #[derive(Debug, PartialEq, Eq)] struct RecordedTurnError { session_level_id: String, thread_level_id: String, turn_level_id: String, turn_id: String, error: CodexErrorInfo, saw_session_store: bool, saw_thread_store: bool, } struct TurnErrorRecorder { records: Arc<std::sync::Mutex<Vec<RecordedTurnError>>>, } impl codex_extension_api::TurnLifecycleContributor for TurnErrorRecorder { fn on_turn_error<'a>( &'a self, input: codex_extension_api::TurnErrorInput<'a>, ) -> codex_extension_api::ExtensionFuture<'a, ()> { Box::pin(async move { self.records .lock() .expect("turn error records lock") .push(RecordedTurnError { session_level_id: input.session_store.level_id().to_string(), thread_level_id: input.thread_store.level_id().to_string(), turn_level_id: input.turn_store.level_id().to_string(), turn_id: input.turn_id.to_string(), error: input.error, saw_session_store: input .session_store .get::<SessionTurnErrorMarker>() .is_some(), saw_thread_store: input .thread_store .get::<ThreadTurnErrorMarker>() .is_some(), }); }) } } let (mut session, turn_context) = make_session_and_context().await; let records = Arc::new(std::sync::Mutex::new(Vec::new())); let mut builder = codex_extension_api::ExtensionRegistryBuilder::<crate::config::Config>::new(); builder.turn_lifecycle_contributor(Arc::new(TurnErrorRecorder { records: Arc::clone(&records), })); session.services.extensions = Arc::new(builder.build()); session .services .session_extension_data .insert(SessionTurnErrorMarker); session .services .thread_extension_data .insert(ThreadTurnErrorMarker); let expected = RecordedTurnError { session_level_id: session.session_id().to_string(), thread_level_id: session.thread_id.to_string(), turn_level_id: turn_context.sub_id.clone(), turn_id: turn_context.sub_id.clone(), error: CodexErrorInfo::UsageLimitExceeded, saw_session_store: true, saw_thread_store: true, }; session .emit_turn_error_lifecycle(&turn_context, CodexErrorInfo::UsageLimitExceeded) .await; let actual = records .lock() .expect("turn error records lock") .drain(..) .collect::<Vec<_>>(); assert_eq!(vec![expected], actual);}#[tokio::test]async fn config_change_contributor_observes_effective_config_changes() { struct SessionConfigMarker; struct ThreadConfigMarker; #[derive(Debug, PartialEq)] struct RecordedConfigChange { previous_model: Option<String>, new_model: Option<String>, previous_disabled_tools: Vec<ToolSuggestDisabledTool>, new_disabled_tools: Vec<ToolSuggestDisabledTool>, saw_session_store: bool, saw_thread_store: bool, } struct ConfigRecorder { records: Arc<std::sync::Mutex<Vec<RecordedConfigChange>>>, } impl codex_extension_api::ConfigContributor<crate::config::Config> for ConfigRecorder { fn on_config_changed( &self, session_store: &codex_extension_api::ExtensionData, thread_store: &codex_extension_api::ExtensionData, previous_config: &crate::config::Config, new_config: &crate::config::Config, ) { self.records .lock() .expect("config change records lock") .push(RecordedConfigChange { previous_model: previous_config.model.clone(), new_model: new_config.model.clone(), previous_disabled_tools: previous_config.tool_suggest.disabled_tools.clone(), new_disabled_tools: new_config.tool_suggest.disabled_tools.clone(), saw_session_store: session_store.get::<SessionConfigMarker>().is_some(), saw_thread_store: thread_store.get::<ThreadConfigMarker>().is_some(), }); } } let (mut session, _turn_context) = make_session_and_context().await; let records = Arc::new(std::sync::Mutex::new(Vec::new())); let mut builder = codex_extension_api::ExtensionRegistryBuilder::<crate::config::Config>::new(); builder.config_contributor(Arc::new(ConfigRecorder { records: Arc::clone(&records), })); session.services.extensions = Arc::new(builder.build()); session .services .session_extension_data .insert(SessionConfigMarker); session .services .thread_extension_data .insert(ThreadConfigMarker); let original_model = session.collaboration_mode().await.model().to_string(); let original_disabled_tools = session .get_config() .await .tool_suggest .disabled_tools .clone(); let next_model = if original_model == "gpt-5.4" { "gpt-5.2" } else { "gpt-5.4" }; let collaboration_mode = session.collaboration_mode().await.with_updates( Some(next_model.to_string()), /*effort*/ None, /*developer_instructions*/ None, ); session .update_settings(SessionSettingsUpdate { collaboration_mode: Some(collaboration_mode), ..Default::default() }) .await .expect("update settings"); let codex_home = session.codex_home().await; std::fs::create_dir_all(&codex_home).expect("create codex home"); std::fs::write( codex_home.join(CONFIG_TOML_FILE), r#"[tool_suggest]disabled_tools = [ { type = "connector", id = " calendar " }, { type = "plugin", id = "slack@openai-curated" },]"#, ) .expect("write user config"); let next_config = load_latest_config_for_session(&session).await; session.refresh_runtime_config(next_config).await; let expected_disabled_tools = vec![ ToolSuggestDisabledTool::connector("calendar"), ToolSuggestDisabledTool::plugin("slack@openai-curated"), ]; let expected = vec![ RecordedConfigChange { previous_model: Some(original_model), new_model: Some(next_model.to_string()), previous_disabled_tools: original_disabled_tools.clone(), new_disabled_tools: original_disabled_tools.clone(), saw_session_store: true, saw_thread_store: true, }, RecordedConfigChange { previous_model: Some(next_model.to_string()), new_model: Some(next_model.to_string()), previous_disabled_tools: original_disabled_tools, new_disabled_tools: expected_disabled_tools, saw_session_store: true, saw_thread_store: true, }, ]; let actual = records .lock() .expect("config change records lock") .drain(..) .collect::<Vec<_>>(); assert_eq!(expected, actual);}#[tokio::test]async fn record_initial_history_reconstructs_forked_transcript() { let (session, turn_context) = make_session_and_context().await; let (rollout_items, expected) = sample_rollout(&session, &turn_context).await; session .record_initial_history(InitialHistory::Forked(rollout_items)) .await; let history = session.state.lock().await.clone_history(); assert_eq!(expected, history.raw_items());}#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn session_configured_reports_permission_profile_for_external_sandbox() -> anyhow::Result<()>{ let server = start_mock_server().await; let sandbox_policy = SandboxPolicy::ExternalSandbox { network_access: codex_protocol::protocol::NetworkAccess::Restricted, }; let permission_profile = PermissionProfile::External { network: NetworkSandboxPolicy::Restricted, }; let expected_permission_profile = permission_profile.clone(); let mut builder = test_codex().with_config(move |config| { config .permissions .set_permission_profile(permission_profile.clone()) .expect("set permission profile"); config .set_legacy_sandbox_policy(sandbox_policy) .expect("set sandbox policy"); }); let test = builder.build(&server).await?; assert_eq!( test.session_configured.permission_profile, expected_permission_profile, "ExternalSandbox is represented explicitly instead of as a lossy root-write profile" ); Ok(())}#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn session_permission_profile_rebinds_runtime_workspace_roots() -> anyhow::Result<()> { let codex_home = tempfile::TempDir::new()?; let cwd = tempfile::TempDir::new()?; let old_root = test_path_buf("/workspace/old").abs(); let new_root = test_path_buf("/workspace/new").abs(); let config = ConfigBuilder::default() .codex_home(codex_home.path().to_path_buf()) .harness_overrides(crate::config::ConfigOverrides { cwd: Some(cwd.path().to_path_buf()), default_permissions: Some(BUILT_IN_PERMISSION_PROFILE_WORKSPACE.to_string()), additional_writable_roots: vec![old_root.to_path_buf()], ..Default::default() }) .build() .await?; let session_permission_profile_state = session_permission_profile_state_from_config(&config)?; let stored_file_system_policy = session_permission_profile_state .permission_profile() .file_system_sandbox_policy(); assert!( !stored_file_system_policy .can_write_path_with_cwd(old_root.as_path(), config.cwd.as_path()), "session permission profile state should keep runtime workspace roots symbolic" ); let mut session_configuration = make_session_configuration_for_tests().await; session_configuration.environments = TurnEnvironmentSelections::new(config.cwd.clone(), Vec::new()); session_configuration.workspace_roots = config.workspace_roots.clone(); session_configuration.permission_profile_state = session_permission_profile_state; let initial_policy = session_configuration.file_system_sandbox_policy(); assert!(initial_policy.can_write_path_with_cwd(old_root.as_path(), config.cwd.as_path())); let updated = session_configuration.apply(&SessionSettingsUpdate { workspace_roots: Some(vec![new_root.clone()]), ..Default::default() })?; let updated_policy = updated.file_system_sandbox_policy(); assert!(updated_policy.can_write_path_with_cwd(new_root.as_path(), updated.cwd().as_path())); assert!(!updated_policy.can_write_path_with_cwd(old_root.as_path(), updated.cwd().as_path())); Ok(())}#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn fork_startup_context_then_first_turn_diff_snapshot() -> anyhow::Result<()> { let server = start_mock_server().await; mount_sse_once( &server, sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), ) .await; let first_forked_request = mount_sse_once( &server, sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]), ) .await; let mut builder = test_codex().with_config(|config| { config.permissions.approval_policy = codex_config::Constrained::allow_any(AskForApproval::OnRequest); }); let initial = builder.build(&server).await?; let rollout_path = initial .session_configured .rollout_path .clone() .expect("rollout path"); initial .codex .submit(Op::UserInput { items: vec![UserInput::Text { text: "fork seed".into(), text_elements: Vec::new(), }], final_output_json_schema: None, responsesapi_client_metadata: None, additional_context: Default::default(), thread_settings: Default::default(), }) .await?; wait_for_event(&initial.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; // Forking reads the persisted rollout JSONL, so force the completed source turn to disk // before snapshotting from it. initial.codex.ensure_rollout_materialized().await; initial .codex .flush_rollout() .await .expect("source rollout should flush before fork"); let mut fork_config = initial.config.clone(); fork_config.permissions.approval_policy = codex_config::Constrained::allow_any(AskForApproval::UnlessTrusted); let forked = initial .thread_manager .fork_thread( usize::MAX, fork_config.clone(), rollout_path, /*thread_source*/ None, /*parent_trace*/ None, ) .await?; let collaboration_mode = CollaborationMode { mode: ModeKind::Plan, settings: Settings { model: forked.session_configured.model.clone(), reasoning_effort: None, developer_instructions: Some("Fork turn collaboration instructions.".to_string()), }, }; forked .thread .submit(Op::UserInput { items: vec![UserInput::Text { text: "after fork".into(), text_elements: Vec::new(), }], final_output_json_schema: None, responsesapi_client_metadata: None, additional_context: Default::default(), thread_settings: ThreadSettingsOverrides { approval_policy: Some(AskForApproval::Never), collaboration_mode: Some(collaboration_mode), ..Default::default() }, }) .await?; wait_for_event(&forked.thread, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; let request = first_forked_request.single_request(); let snapshot = context_snapshot::format_labeled_requests_snapshot( "First request after fork when startup preserves the parent baseline, the fork changes approval policy, and the first forked turn enters plan mode.", &[("First Forked Turn Request", &request)], &ContextSnapshotOptions::default() .render_mode(ContextSnapshotRenderMode::KindWithTextPrefix { max_chars: 96 }) .strip_capability_instructions() .strip_agents_md_user_context(), ); let mut settings = insta::Settings::clone_current(); settings.set_snapshot_path("snapshots"); settings.set_prepend_module_to_snapshot(false); settings.bind(|| { insta::assert_snapshot!( "codex_core__codex_tests__fork_startup_context_then_first_turn_diff", snapshot ); }); Ok(())}#[tokio::test]async fn record_initial_history_forked_hydrates_previous_turn_settings() { let (session, turn_context) = make_session_and_context().await; let previous_model = "forked-rollout-model"; let previous_context_item = TurnContextItem { turn_id: Some(turn_context.sub_id.clone()), #[allow(deprecated)] cwd: turn_context.cwd.to_path_buf(), workspace_roots: None, current_date: turn_context.current_date.clone(), timezone: turn_context.timezone.clone(), approval_policy: turn_context.approval_policy.value(), sandbox_policy: turn_context.sandbox_policy(), permission_profile: None, network: None, file_system_sandbox_policy: None, model: previous_model.to_string(), comp_hash: None, personality: turn_context.personality, collaboration_mode: Some(turn_context.collaboration_mode.clone()), multi_agent_version: None, realtime_active: Some(turn_context.realtime_active), effort: turn_context.reasoning_effort.clone(), summary: codex_protocol::config_types::ReasoningSummary::Auto, }; let turn_id = previous_context_item .turn_id .clone() .expect("thread settings should have turn_id"); let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: turn_id.clone(), trace_id: None, started_at: None, model_context_window: Some(128_000), collaboration_mode_kind: ModeKind::Default, }, )), RolloutItem::EventMsg(EventMsg::UserMessage( codex_protocol::protocol::UserMessageEvent { client_id: None, message: "forked seed".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() }, )), RolloutItem::TurnContext(previous_context_item.clone()), RolloutItem::EventMsg(EventMsg::TurnComplete( codex_protocol::protocol::TurnCompleteEvent { turn_id, last_agent_message: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, }, )), ]; session .record_initial_history(InitialHistory::Forked(rollout_items)) .await; let history = session.clone_history().await; assert_eq!( session.previous_turn_settings().await, Some(PreviousTurnSettings { model: previous_model.to_string(), comp_hash: None, realtime_active: Some(turn_context.realtime_active), }) ); assert_eq!(history.raw_items(), &[]); assert_eq!( serde_json::to_value(session.reference_context_item().await) .expect("serialize fork reference context item"), serde_json::to_value(Some(previous_context_item)) .expect("serialize expected reference context item") );}#[tokio::test]async fn thread_rollback_drops_last_turn_from_history() { let (mut sess, tc, rx) = make_session_and_context_with_rx().await; let rollout_path = attach_thread_persistence( Arc::get_mut(&mut sess).expect("session should not have additional references"), ) .await; let initial_context = sess.build_initial_context(tc.as_ref()).await; let turn_1 = vec![ user_message("turn 1 user"), assistant_message("turn 1 assistant"), ]; let turn_2 = vec![ user_message("turn 2 user"), assistant_message("turn 2 assistant"), ]; let mut full_history = Vec::new(); full_history.extend(initial_context.clone()); full_history.extend(turn_1.clone()); full_history.extend(turn_2); sess.replace_history(full_history.clone(), Some(tc.to_turn_context_item())) .await; let rollout_items: Vec<RolloutItem> = full_history .into_iter() .map(RolloutItem::ResponseItem) .collect(); sess.persist_rollout_items(&rollout_items).await; sess.set_previous_turn_settings(Some(PreviousTurnSettings { model: "stale-model".to_string(), comp_hash: None, realtime_active: Some(tc.realtime_active), })) .await; { let mut state = sess.state.lock().await; state.set_reference_context_item(Some(tc.to_turn_context_item())); } handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await; let rollback_event = wait_for_thread_rolled_back(&rx).await; assert_eq!(rollback_event.num_turns, 1); let mut expected = Vec::new(); expected.extend(initial_context); expected.extend(turn_1); let history = sess.clone_history().await; assert_eq!(expected, history.raw_items()); assert_eq!(sess.previous_turn_settings().await, None); assert!(sess.reference_context_item().await.is_none()); let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path) .await .expect("read rollout history") else { panic!("expected resumed rollout history"); }; assert!(resumed.history.iter().any(|item| { matches!( item, RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) if rollback.num_turns == 1 ) }));}#[tokio::test]async fn thread_rollback_clears_history_when_num_turns_exceeds_existing_turns() { let (mut sess, tc, rx) = make_session_and_context_with_rx().await; attach_thread_persistence( Arc::get_mut(&mut sess).expect("session should not have additional references"), ) .await; let initial_context = sess.build_initial_context(tc.as_ref()).await; let turn_1 = vec![user_message("turn 1 user")]; let mut full_history = Vec::new(); full_history.extend(initial_context.clone()); full_history.extend(turn_1); sess.replace_history(full_history.clone(), Some(tc.to_turn_context_item())) .await; let rollout_items: Vec<RolloutItem> = full_history .into_iter() .map(RolloutItem::ResponseItem) .collect(); sess.persist_rollout_items(&rollout_items).await; handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 99).await; let rollback_event = wait_for_thread_rolled_back(&rx).await; assert_eq!(rollback_event.num_turns, 99); let history = sess.clone_history().await; assert_eq!(initial_context, history.raw_items());}#[tokio::test]async fn thread_rollback_fails_without_persisted_thread_history() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let initial_context = sess.build_initial_context(tc.as_ref()).await; sess.record_conversation_items(tc.as_ref(), &initial_context) .await; handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await; let error_event = wait_for_thread_rollback_failed(&rx).await; assert_eq!( error_event.message, "thread rollback requires persisted thread history" ); assert_eq!( error_event.codex_error_info, Some(CodexErrorInfo::ThreadRollbackFailed) ); assert_eq!(sess.clone_history().await.raw_items(), initial_context);}#[tokio::test]async fn thread_rollback_recomputes_previous_turn_settings_and_reference_context_from_replay() { let (mut sess, tc, rx) = make_session_and_context_with_rx().await; attach_thread_persistence( Arc::get_mut(&mut sess).expect("session should not have additional references"), ) .await; let first_context_item = tc.to_turn_context_item(); let first_turn_id = first_context_item .turn_id .clone() .expect("thread settings should have turn_id"); let mut rolled_back_context_item = first_context_item.clone(); rolled_back_context_item.turn_id = Some("rolled-back-turn".to_string()); rolled_back_context_item.model = "rolled-back-model".to_string(); let rolled_back_turn_id = rolled_back_context_item .turn_id .clone() .expect("thread settings should have turn_id"); let turn_one_user = user_message("turn 1 user"); let turn_one_assistant = assistant_message("turn 1 assistant"); let turn_two_user = user_message("turn 2 user"); let turn_two_assistant = assistant_message("turn 2 assistant"); sess.persist_rollout_items(&[ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: first_turn_id.clone(), trace_id: None, started_at: None, model_context_window: Some(128_000), collaboration_mode_kind: ModeKind::Default, }, )), RolloutItem::EventMsg(EventMsg::UserMessage( codex_protocol::protocol::UserMessageEvent { client_id: None, message: "turn 1 user".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() }, )), RolloutItem::TurnContext(first_context_item.clone()), RolloutItem::ResponseItem(turn_one_user.clone()), RolloutItem::ResponseItem(turn_one_assistant.clone()), RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { turn_id: first_turn_id, last_agent_message: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, })), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: rolled_back_turn_id.clone(), trace_id: None, started_at: None, model_context_window: Some(128_000), collaboration_mode_kind: ModeKind::Default, }, )), RolloutItem::EventMsg(EventMsg::UserMessage( codex_protocol::protocol::UserMessageEvent { client_id: None, message: "turn 2 user".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() }, )), RolloutItem::TurnContext(rolled_back_context_item), RolloutItem::ResponseItem(turn_two_user), RolloutItem::ResponseItem(turn_two_assistant), RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { turn_id: rolled_back_turn_id, last_agent_message: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, })), ]) .await; sess.replace_history( vec![assistant_message("stale history")], Some(first_context_item.clone()), ) .await; sess.set_previous_turn_settings(Some(PreviousTurnSettings { model: "stale-model".to_string(), comp_hash: None, realtime_active: None, })) .await; handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await; let rollback_event = wait_for_thread_rolled_back(&rx).await; assert_eq!(rollback_event.num_turns, 1); assert_eq!( sess.clone_history().await.raw_items(), vec![turn_one_user, turn_one_assistant] ); assert_eq!( sess.previous_turn_settings().await, Some(PreviousTurnSettings { model: tc.model_info.slug.clone(), comp_hash: None, realtime_active: Some(tc.realtime_active), }) ); assert_eq!( serde_json::to_value(sess.reference_context_item().await) .expect("serialize replay reference context item"), serde_json::to_value(Some(first_context_item)) .expect("serialize expected reference context item") );}#[tokio::test]async fn thread_rollback_restores_cleared_reference_context_item_after_compaction() { let (mut sess, tc, rx) = make_session_and_context_with_rx().await; attach_thread_persistence( Arc::get_mut(&mut sess).expect("session should not have additional references"), ) .await; let first_context_item = tc.to_turn_context_item(); let first_turn_id = first_context_item .turn_id .clone() .expect("thread settings should have turn_id"); let compact_turn_id = "compact-turn".to_string(); let rolled_back_turn_id = "rolled-back-turn".to_string(); let compacted_history = vec![ user_message("turn 1 user"), user_message("summary after compaction"), ]; sess.persist_rollout_items(&[ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: first_turn_id.clone(), trace_id: None, started_at: None, model_context_window: Some(128_000), collaboration_mode_kind: ModeKind::Default, }, )), RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "turn 1 user".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() })), RolloutItem::TurnContext(first_context_item.clone()), RolloutItem::ResponseItem(user_message("turn 1 user")), RolloutItem::ResponseItem(assistant_message("turn 1 assistant")), RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { turn_id: first_turn_id, last_agent_message: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, })), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: compact_turn_id.clone(), trace_id: None, started_at: None, model_context_window: Some(128_000), collaboration_mode_kind: ModeKind::Default, }, )), RolloutItem::Compacted(CompactedItem { message: "summary after compaction".to_string(), replacement_history: Some(compacted_history.clone()), window_id: Some(7), }), RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { turn_id: compact_turn_id, last_agent_message: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, })), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: rolled_back_turn_id.clone(), trace_id: None, started_at: None, model_context_window: Some(128_000), collaboration_mode_kind: ModeKind::Default, }, )), RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "turn 2 user".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() })), RolloutItem::TurnContext(TurnContextItem { turn_id: Some(rolled_back_turn_id.clone()), model: "rolled-back-model".to_string(), comp_hash: None, ..first_context_item.clone() }), RolloutItem::ResponseItem(user_message("turn 2 user")), RolloutItem::ResponseItem(assistant_message("turn 2 assistant")), RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { turn_id: rolled_back_turn_id, last_agent_message: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, })), ]) .await; sess.replace_history( vec![assistant_message("stale history")], Some(first_context_item), ) .await; { let mut state = sess.state.lock().await; state.set_auto_compact_window_id(/*window_id*/ 99); } handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await; let rollback_event = wait_for_thread_rolled_back(&rx).await; assert_eq!(rollback_event.num_turns, 1); assert_eq!(sess.clone_history().await.raw_items(), compacted_history); assert!(sess.reference_context_item().await.is_none()); assert!(sess.current_window_id().await.ends_with(":7"));}#[tokio::test]async fn thread_rollback_persists_marker_and_replays_cumulatively() { let (mut sess, tc, rx) = make_session_and_context_with_rx().await; let rollout_path = attach_thread_persistence( Arc::get_mut(&mut sess).expect("session should not have additional references"), ) .await; let turn_context_item = tc.to_turn_context_item(); sess.persist_rollout_items(&[ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-1".to_string(), trace_id: None, started_at: None, model_context_window: Some(128_000), collaboration_mode_kind: ModeKind::Default, }, )), RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "turn 1 user".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() })), RolloutItem::TurnContext(turn_context_item.clone()), RolloutItem::ResponseItem(user_message("turn 1 user")), RolloutItem::ResponseItem(assistant_message("turn 1 assistant")), RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { turn_id: "turn-1".to_string(), last_agent_message: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, })), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-2".to_string(), trace_id: None, started_at: None, model_context_window: Some(128_000), collaboration_mode_kind: ModeKind::Default, }, )), RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "turn 2 user".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() })), RolloutItem::TurnContext(turn_context_item.clone()), RolloutItem::ResponseItem(user_message("turn 2 user")), RolloutItem::ResponseItem(assistant_message("turn 2 assistant")), RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { turn_id: "turn-2".to_string(), last_agent_message: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, })), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-3".to_string(), trace_id: None, started_at: None, model_context_window: Some(128_000), collaboration_mode_kind: ModeKind::Default, }, )), RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "turn 3 user".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() })), RolloutItem::TurnContext(turn_context_item), RolloutItem::ResponseItem(user_message("turn 3 user")), RolloutItem::ResponseItem(assistant_message("turn 3 assistant")), RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { turn_id: "turn-3".to_string(), last_agent_message: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, })), ]) .await; handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await; let first_rollback = wait_for_thread_rolled_back(&rx).await; assert_eq!(first_rollback.num_turns, 1); handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await; let second_rollback = wait_for_thread_rolled_back(&rx).await; assert_eq!(second_rollback.num_turns, 1); assert_eq!( sess.clone_history().await.raw_items(), vec![ user_message("turn 1 user"), assistant_message("turn 1 assistant") ] ); let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path) .await .expect("read rollout history") else { panic!("expected resumed rollout history"); }; let rollback_markers = resumed .history .iter() .filter(|item| matches!(item, RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_)))) .count(); assert_eq!(rollback_markers, 2);}#[tokio::test]async fn thread_rollback_fails_when_turn_in_progress() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let initial_context = sess.build_initial_context(tc.as_ref()).await; sess.record_conversation_items(tc.as_ref(), &initial_context) .await; *sess.active_turn.lock().await = Some(crate::state::ActiveTurn::default()); handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await; let error_event = wait_for_thread_rollback_failed(&rx).await; assert_eq!( error_event.codex_error_info, Some(CodexErrorInfo::ThreadRollbackFailed) ); let history = sess.clone_history().await; assert_eq!(initial_context, history.raw_items());}#[tokio::test]async fn thread_rollback_fails_when_num_turns_is_zero() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let initial_context = sess.build_initial_context(tc.as_ref()).await; sess.record_conversation_items(tc.as_ref(), &initial_context) .await; handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 0).await; let error_event = wait_for_thread_rollback_failed(&rx).await; assert_eq!(error_event.message, "num_turns must be >= 1"); assert_eq!( error_event.codex_error_info, Some(CodexErrorInfo::ThreadRollbackFailed) ); let history = sess.clone_history().await; assert_eq!(initial_context, history.raw_items());}#[tokio::test]async fn set_rate_limits_retains_previous_credits() { let codex_home = tempfile::tempdir().expect("create temp dir"); let config = build_test_config(codex_home.path()).await; let config = Arc::new(config); let model = get_model_offline_for_tests(config.model.as_deref()); let model_info = construct_model_info_offline_for_tests(model.as_str(), &config.to_models_manager_config()); let reasoning_effort = config.model_reasoning_effort.clone(); let collaboration_mode = CollaborationMode { mode: ModeKind::Default, settings: Settings { model, reasoning_effort, developer_instructions: None, }, }; let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config .base_instructions .clone() .unwrap_or_else(|| model_info.get_model_instructions(config.personality)), compact_prompt: config.compact_prompt.clone(), approval_policy: config.permissions.approval_policy.clone(), approvals_reviewer: config.approvals_reviewer, permission_profile_state: config.permissions.permission_profile_state().clone(), windows_sandbox_level: WindowsSandboxLevel::from_config(&config), environments: TurnEnvironmentSelections::new(config.cwd.clone(), Vec::new()), workspace_roots: config.workspace_roots.clone(), codex_home: config.codex_home.clone(), thread_name: None, original_config_do_not_use: Arc::clone(&config), metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, forked_from_thread_id: None, parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), user_shell_override: None, }; let mut state = SessionState::new(session_configuration); let initial = RateLimitSnapshot { limit_id: None, limit_name: None, primary: Some(RateLimitWindow { used_percent: 10.0, window_minutes: Some(15), resets_at: Some(1_700), }), secondary: None, credits: Some(CreditsSnapshot { has_credits: true, unlimited: false, balance: Some("10.00".to_string()), }), individual_limit: None, plan_type: Some(codex_protocol::account::PlanType::Plus), rate_limit_reached_type: None, }; state.set_rate_limits(initial.clone()); let update = RateLimitSnapshot { limit_id: Some("codex_other".to_string()), limit_name: Some("codex_other".to_string()), primary: Some(RateLimitWindow { used_percent: 40.0, window_minutes: Some(30), resets_at: Some(1_800), }), secondary: Some(RateLimitWindow { used_percent: 5.0, window_minutes: Some(60), resets_at: Some(1_900), }), credits: None, individual_limit: None, plan_type: None, rate_limit_reached_type: None, }; state.set_rate_limits(update.clone()); assert_eq!( state.latest_rate_limits, Some(RateLimitSnapshot { limit_id: Some("codex_other".to_string()), limit_name: Some("codex_other".to_string()), primary: update.primary.clone(), secondary: update.secondary, credits: initial.credits, individual_limit: initial.individual_limit, plan_type: initial.plan_type, rate_limit_reached_type: None, }) );}#[tokio::test]async fn set_rate_limits_updates_plan_type_when_present() { let codex_home = tempfile::tempdir().expect("create temp dir"); let config = build_test_config(codex_home.path()).await; let config = Arc::new(config); let model = get_model_offline_for_tests(config.model.as_deref()); let model_info = construct_model_info_offline_for_tests(model.as_str(), &config.to_models_manager_config()); let reasoning_effort = config.model_reasoning_effort.clone(); let collaboration_mode = CollaborationMode { mode: ModeKind::Default, settings: Settings { model, reasoning_effort, developer_instructions: None, }, }; let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config .base_instructions .clone() .unwrap_or_else(|| model_info.get_model_instructions(config.personality)), compact_prompt: config.compact_prompt.clone(), approval_policy: config.permissions.approval_policy.clone(), approvals_reviewer: config.approvals_reviewer, permission_profile_state: config.permissions.permission_profile_state().clone(), windows_sandbox_level: WindowsSandboxLevel::from_config(&config), environments: TurnEnvironmentSelections::new(config.cwd.clone(), Vec::new()), workspace_roots: config.workspace_roots.clone(), codex_home: config.codex_home.clone(), thread_name: None, original_config_do_not_use: Arc::clone(&config), metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, forked_from_thread_id: None, parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), user_shell_override: None, }; let mut state = SessionState::new(session_configuration); let initial = RateLimitSnapshot { limit_id: None, limit_name: None, primary: Some(RateLimitWindow { used_percent: 15.0, window_minutes: Some(20), resets_at: Some(1_600), }), secondary: Some(RateLimitWindow { used_percent: 5.0, window_minutes: Some(45), resets_at: Some(1_650), }), credits: Some(CreditsSnapshot { has_credits: true, unlimited: false, balance: Some("15.00".to_string()), }), individual_limit: None, plan_type: Some(codex_protocol::account::PlanType::Plus), rate_limit_reached_type: None, }; state.set_rate_limits(initial.clone()); let update = RateLimitSnapshot { limit_id: None, limit_name: None, primary: Some(RateLimitWindow { used_percent: 35.0, window_minutes: Some(25), resets_at: Some(1_700), }), secondary: None, credits: None, individual_limit: None, plan_type: Some(codex_protocol::account::PlanType::Pro), rate_limit_reached_type: None, }; state.set_rate_limits(update.clone()); assert_eq!( state.latest_rate_limits, Some(RateLimitSnapshot { limit_id: Some("codex".to_string()), limit_name: None, primary: update.primary, secondary: update.secondary, credits: initial.credits, individual_limit: initial.individual_limit, plan_type: update.plan_type, rate_limit_reached_type: None, }) );}#[test]fn prefers_structured_content_when_present() { let ctr = McpCallToolResult { // Content present but should be ignored because structured_content is set. content: vec![text_block("ignored")], is_error: None, structured_content: Some(json!({ "ok": true, "value": 42 })), meta: None, }; let got = ctr.into_function_call_output_payload(); let expected = FunctionCallOutputPayload { body: FunctionCallOutputBody::Text( serde_json::to_string(&json!({ "ok": true, "value": 42 })) .unwrap(), ), success: Some(true), }; assert_eq!(expected, got);}#[tokio::test]async fn includes_timed_out_message() { let exec = ExecToolCallOutput { exit_code: 0, stdout: StreamOutput::new(String::new()), stderr: StreamOutput::new(String::new()), aggregated_output: StreamOutput::new("Command output".to_string()), duration: StdDuration::from_secs(1), timed_out: true, }; let (_, turn_context) = make_session_and_context().await; let out = format_exec_output_str(&exec, turn_context.truncation_policy); assert_eq!( out, "command timed out after 1000 milliseconds\nCommand output" );}#[tokio::test]async fn turn_context_with_model_updates_model_fields() { let (session, mut turn_context) = make_session_and_context().await; turn_context.reasoning_effort = Some(ReasoningEffortConfig::Minimal); let updated = turn_context .with_model("gpt-5.4".to_string(), &session.services.models_manager) .await; let expected_model_info = session .services .models_manager .get_model_info( "gpt-5.4", &updated.config.as_ref().to_models_manager_config(), ) .await; assert_eq!(updated.config.model.as_deref(), Some("gpt-5.4")); assert_eq!(updated.collaboration_mode.model(), "gpt-5.4"); assert_eq!(updated.model_info, expected_model_info); assert_eq!( updated.reasoning_effort, Some(ReasoningEffortConfig::Medium) ); assert_eq!( updated.collaboration_mode.reasoning_effort(), Some(ReasoningEffortConfig::Medium) ); assert_eq!( updated.config.model_reasoning_effort, Some(ReasoningEffortConfig::Medium) ); assert_eq!( updated.truncation_policy, expected_model_info.truncation_policy.into() );}#[test]fn falls_back_to_content_when_structured_is_null() { let ctr = McpCallToolResult { content: vec![text_block("hello"), text_block("world")], is_error: None, structured_content: Some(serde_json::Value::Null), meta: None, }; let got = ctr.into_function_call_output_payload(); let expected = FunctionCallOutputPayload { body: FunctionCallOutputBody::Text( serde_json::to_string(&vec![text_block("hello"), text_block("world")]).unwrap(), ), success: Some(true), }; assert_eq!(expected, got);}#[test]fn success_flag_reflects_is_error_true() { let ctr = McpCallToolResult { content: vec![text_block("unused")], is_error: Some(true), structured_content: Some(json!({ "message": "bad" })), meta: None, }; let got = ctr.into_function_call_output_payload(); let expected = FunctionCallOutputPayload { body: FunctionCallOutputBody::Text( serde_json::to_string(&json!({ "message": "bad" })).unwrap(), ), success: Some(false), }; assert_eq!(expected, got);}#[test]fn success_flag_true_with_no_error_and_content_used() { let ctr = McpCallToolResult { content: vec![text_block("alpha")], is_error: Some(false), structured_content: None, meta: None, }; let got = ctr.into_function_call_output_payload(); let expected = FunctionCallOutputPayload { body: FunctionCallOutputBody::Text( serde_json::to_string(&vec![text_block("alpha")]).unwrap(), ), success: Some(true), }; assert_eq!(expected, got);}async fn wait_for_thread_rolled_back(rx: &async_channel::Receiver<Event>) -> ThreadRolledBackEvent { let deadline = StdDuration::from_secs(2); let start = std::time::Instant::now(); loop { let remaining = deadline.saturating_sub(start.elapsed()); let evt = tokio::time::timeout(remaining, rx.recv()) .await .expect("timeout waiting for event") .expect("event"); match evt.msg { EventMsg::ThreadRolledBack(payload) => return payload, _ => continue, } }}async fn wait_for_thread_rollback_failed(rx: &async_channel::Receiver<Event>) -> ErrorEvent { let deadline = StdDuration::from_secs(2); let start = std::time::Instant::now(); loop { let remaining = deadline.saturating_sub(start.elapsed()); let evt = tokio::time::timeout(remaining, rx.recv()) .await .expect("timeout waiting for event") .expect("event"); match evt.msg { EventMsg::Error(payload) if payload.codex_error_info == Some(CodexErrorInfo::ThreadRollbackFailed) => { return payload; } _ => continue, } }}async fn attach_thread_persistence(session: &mut Session) -> PathBuf { let config = session.get_config().await; let live_thread = LiveThread::create( Arc::clone(&session.services.thread_store), CreateThreadParams { thread_id: session.thread_id, extra_config: None, forked_from_id: None, parent_thread_id: None, source: SessionSource::Exec, thread_source: None, base_instructions: BaseInstructions::default(), dynamic_tools: Vec::new(), multi_agent_version: None, metadata: ThreadPersistenceMetadata { cwd: Some(config.cwd.to_path_buf()), model_provider: config.model_provider_id.clone(), memory_mode: if config.memories.generate_memories { ThreadMemoryMode::Enabled } else { ThreadMemoryMode::Disabled }, }, }, ) .await .expect("create thread persistence"); session.services.live_thread = Some(live_thread); session.ensure_rollout_materialized().await; session .flush_rollout() .await .expect("attached rollout should flush"); session .current_rollout_path() .await .expect("load rollout path") .expect("thread should have rollout path")}fn text_block(s: &str) -> serde_json::Value { json!({ "type": "text", "text": s, })}async fn build_test_config(codex_home: &Path) -> Config { ConfigBuilder::without_managed_config_for_tests() .codex_home(codex_home.to_path_buf()) .build() .await .expect("load default test config")}fn session_telemetry( conversation_id: ThreadId, config: &Config, model_info: &ModelInfo, session_source: SessionSource,) -> SessionTelemetry { SessionTelemetry::new( conversation_id, get_model_offline_for_tests(config.model.as_deref()).as_str(), model_info.slug.as_str(), /*account_id*/ None, Some("test@test.com".to_string()), Some(TelemetryAuthMode::Chatgpt), "test_originator".to_string(), /*log_user_prompts*/ false, "test".to_string(), session_source, )}fn model_with_default_service_tier(default_service_tier: Option<&str>) -> ModelInfo { let mut model_info = model_info::model_info_from_slug("gpt-5.4"); model_info.service_tiers = vec![ModelServiceTier { id: ServiceTier::Fast.request_value().to_string(), name: "Fast".to_string(), description: "Priority processing.".to_string(), }]; model_info.default_service_tier = default_service_tier.map(str::to_string); model_info}#[test]fn get_service_tier_does_not_use_model_default_when_absent_and_fast_mode_enabled() { let model_info = model_with_default_service_tier(Some(ServiceTier::Fast.request_value())); assert_eq!( get_service_tier( /*configured_service_tier*/ None, /*fast_mode_enabled*/ true, &model_info, ), None );}#[test]fn get_service_tier_does_not_use_model_default_when_fast_mode_disabled() { let model_info = model_with_default_service_tier(Some(ServiceTier::Fast.request_value())); assert_eq!( get_service_tier( /*configured_service_tier*/ None, /*fast_mode_enabled*/ false, &model_info, ), None );}#[test]fn get_service_tier_keeps_supported_explicit_tier() { let model_info = model_with_default_service_tier(Some(ServiceTier::Fast.request_value())); assert_eq!( get_service_tier( Some(ServiceTier::Fast.request_value().to_string()), /*fast_mode_enabled*/ true, &model_info, ), Some(ServiceTier::Fast.request_value().to_string()) );}#[test]fn get_service_tier_does_not_default_when_model_has_no_default() { let model_info = model_with_default_service_tier(/*default_service_tier*/ None); assert_eq!( get_service_tier( /*configured_service_tier*/ None, /*fast_mode_enabled*/ true, &model_info, ), None );}#[test]fn get_service_tier_drops_unsupported_configured_tier_when_fast_mode_enabled() { let model_info = model_with_default_service_tier(Some(ServiceTier::Fast.request_value())); assert_eq!( get_service_tier( Some("unsupported".to_string()), /*fast_mode_enabled*/ true, &model_info, ), None ); assert_eq!( get_service_tier( Some(ServiceTier::Flex.request_value().to_string()), /*fast_mode_enabled*/ true, &model_info, ), None ); assert_eq!( get_service_tier( Some(SERVICE_TIER_DEFAULT_REQUEST_VALUE.to_string()), /*fast_mode_enabled*/ true, &model_info, ), Some(SERVICE_TIER_DEFAULT_REQUEST_VALUE.to_string()) );}#[test]fn get_service_tier_ignores_configured_tier_when_fast_mode_disabled() { let model_info = model_with_default_service_tier(Some(ServiceTier::Fast.request_value())); assert_eq!( get_service_tier( Some(ServiceTier::Fast.request_value().to_string()), /*fast_mode_enabled*/ false, &model_info, ), None ); assert_eq!( get_service_tier( Some(SERVICE_TIER_DEFAULT_REQUEST_VALUE.to_string()), /*fast_mode_enabled*/ false, &model_info, ), None ); assert_eq!( get_service_tier( Some("unsupported".to_string()), /*fast_mode_enabled*/ false, &model_info, ), None ); assert_eq!( get_service_tier( /*configured_service_tier*/ None, /*fast_mode_enabled*/ false, &model_info, ), None );}#[tokio::test]async fn session_settings_null_service_tier_update_uses_default_service_tier() { let session_configuration = make_session_configuration_for_tests().await; let updated = session_configuration .apply(&SessionSettingsUpdate { service_tier: Some(None), ..Default::default() }) .expect("null service tier update should apply"); assert_eq!( updated.service_tier, Some(SERVICE_TIER_DEFAULT_REQUEST_VALUE.to_string()) );}#[tokio::test]async fn session_settings_legacy_fast_service_tier_update_uses_priority_request_value() { let session_configuration = make_session_configuration_for_tests().await; let updated = session_configuration .apply(&SessionSettingsUpdate { service_tier: Some(Some("fast".to_string())), ..Default::default() }) .expect("legacy fast service tier update should apply"); assert_eq!( updated.service_tier, Some(ServiceTier::Fast.request_value().to_string()) );}pub(crate) async fn make_session_configuration_for_tests() -> SessionConfiguration { let codex_home = tempfile::tempdir().expect("create temp dir"); let config = build_test_config(codex_home.path()).await; let config = Arc::new(config); let model = get_model_offline_for_tests(config.model.as_deref()); let model_info = construct_model_info_offline_for_tests(model.as_str(), &config.to_models_manager_config()); let reasoning_effort = config.model_reasoning_effort.clone(); let collaboration_mode = CollaborationMode { mode: ModeKind::Default, settings: Settings { model, reasoning_effort, developer_instructions: None, }, }; SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config .base_instructions .clone() .unwrap_or_else(|| model_info.get_model_instructions(config.personality)), compact_prompt: config.compact_prompt.clone(), approval_policy: config.permissions.approval_policy.clone(), approvals_reviewer: config.approvals_reviewer, permission_profile_state: config.permissions.permission_profile_state().clone(), windows_sandbox_level: WindowsSandboxLevel::from_config(&config), environments: TurnEnvironmentSelections::new(config.cwd.clone(), Vec::new()), workspace_roots: config.workspace_roots.clone(), codex_home: config.codex_home.clone(), thread_name: None, original_config_do_not_use: Arc::clone(&config), metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, forked_from_thread_id: None, parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), user_shell_override: None, }}#[tokio::test]async fn emit_subagent_session_started_includes_fork_lineage_from_session_configuration() { use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path; let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/codex/analytics-events/events")) .respond_with(ResponseTemplate::new(200)) .mount(&server) .await; let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let analytics_events_client = AnalyticsEventsClient::new( auth_manager, server.uri(), /*analytics_enabled*/ Some(true), ); let parent_thread_id = ThreadId::new(); let forked_from_thread_id = ThreadId::new(); let child_thread_id = ThreadId::new(); let mut session_configuration = make_session_configuration_for_tests().await; session_configuration.forked_from_thread_id = Some(forked_from_thread_id); emit_subagent_session_started( &analytics_events_client, AppServerClientMetadata { client_name: Some("codex-tui".to_string()), client_version: Some("1.0.0".to_string()), }, SessionId::from(child_thread_id), child_thread_id, Some(parent_thread_id), session_configuration.thread_config_snapshot(), SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }, ); let event = timeout(Duration::from_secs(1), async { 'wait_for_event: loop { if let Some(requests) = server.received_requests().await { for request in requests { let payload: serde_json::Value = serde_json::from_slice(&request.body).expect("valid analytics payload"); if let Some(event) = payload["events"].as_array().and_then(|events| { events .iter() .find(|event| event["event_type"] == "codex_thread_initialized") }) { break 'wait_for_event event.clone(); } } } tokio::time::sleep(Duration::from_millis(10)).await; } }) .await .expect("subagent initialization analytics should be emitted"); assert_eq!( event["event_params"]["parent_thread_id"], parent_thread_id.to_string() ); assert_eq!( event["event_params"]["forked_from_thread_id"], forked_from_thread_id.to_string() );}async fn resolved_environments_for_configuration( session_configuration: &SessionConfiguration,) -> (Arc<EnvironmentManager>, TurnEnvironmentSnapshot) { let environment_manager = Arc::new(EnvironmentManager::default_for_tests()); let turn_environments = ThreadEnvironments::new( Arc::clone(&environment_manager), default_user_shell(), ShellSnapshot::disabled(), TurnEnvironmentSnapshot::default(), ); turn_environments.update_selections(session_configuration.environment_selections()); (environment_manager, turn_environments.snapshot().await)}#[tokio::test]async fn session_configuration_apply_preserves_profile_file_system_policy_on_cwd_only_update() { let mut session_configuration = make_session_configuration_for_tests().await; let workspace = tempfile::tempdir().expect("create temp dir"); let project_root = workspace.path().join("project"); let original_cwd = project_root.join("subdir"); let docs_dir = original_cwd.join("docs"); std::fs::create_dir_all(&docs_dir).expect("create docs dir"); let project_root = project_root.abs(); let docs_dir = docs_dir.abs(); session_configuration.environments = TurnEnvironmentSelections::new(original_cwd.abs(), Vec::new()); let sandbox_policy = SandboxPolicy::WorkspaceWrite { writable_roots: Vec::new(), network_access: false, exclude_tmpdir_env_var: true, exclude_slash_tmp: true, }; let file_system_sandbox_policy = FileSystemSandboxPolicy::restricted(vec![ FileSystemSandboxEntry { path: FileSystemPath::Special { value: FileSystemSpecialPath::project_roots(/*subpath*/ None), }, access: FileSystemAccessMode::Write, }, FileSystemSandboxEntry { path: FileSystemPath::Path { path: docs_dir }, access: FileSystemAccessMode::Read, }, ]); let network_sandbox_policy = NetworkSandboxPolicy::from(&sandbox_policy); session_configuration .set_permission_profile_for_tests( PermissionProfile::from_runtime_permissions_with_enforcement( SandboxEnforcement::from_legacy_sandbox_policy(&sandbox_policy), &file_system_sandbox_policy, network_sandbox_policy, ), ) .expect("set permission profile"); let expected_file_system_sandbox_policy = file_system_sandbox_policy .materialize_project_roots_with_workspace_roots(&session_configuration.workspace_roots); let updated = session_configuration .apply(&SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new(project_root, Vec::new())), ..Default::default() }) .expect("cwd-only update should succeed"); assert_eq!( updated.file_system_sandbox_policy(), expected_file_system_sandbox_policy );}#[tokio::test]async fn session_configuration_apply_permission_profile_preserves_existing_deny_read_entries() { let mut session_configuration = make_session_configuration_for_tests().await; let cwd = tempfile::tempdir().expect("create temp dir"); session_configuration.environments = TurnEnvironmentSelections::new(cwd.path().abs(), Vec::new()); let workspace_policy = SandboxPolicy::new_workspace_write_policy(); let deny_entry = FileSystemSandboxEntry { path: FileSystemPath::GlobPattern { pattern: "**/*.env".to_string(), }, access: FileSystemAccessMode::Deny, }; let mut existing_file_system_policy = FileSystemSandboxPolicy::from_legacy_sandbox_policy_for_cwd( &workspace_policy, session_configuration.cwd().as_path(), ); existing_file_system_policy.glob_scan_max_depth = Some(2); existing_file_system_policy.entries.push(deny_entry.clone()); session_configuration .set_permission_profile_for_tests( PermissionProfile::from_runtime_permissions_with_enforcement( SandboxEnforcement::from_legacy_sandbox_policy(&workspace_policy), &existing_file_system_policy, NetworkSandboxPolicy::Restricted, ), ) .expect("set permission profile"); let requested_file_system_policy = FileSystemSandboxPolicy::from_legacy_sandbox_policy_for_cwd( &workspace_policy, session_configuration.cwd().as_path(), ); let permission_profile = codex_protocol::models::PermissionProfile::from_runtime_permissions( &requested_file_system_policy, NetworkSandboxPolicy::Restricted, ); let updated = session_configuration .apply(&SessionSettingsUpdate { permission_profile: Some(permission_profile), ..Default::default() }) .expect("permission profile update should succeed"); let mut expected_file_system_policy = requested_file_system_policy .materialize_project_roots_with_workspace_roots(&session_configuration.workspace_roots); expected_file_system_policy.glob_scan_max_depth = Some(2); expected_file_system_policy.entries.push(deny_entry); assert_eq!( updated.file_system_sandbox_policy(), expected_file_system_policy );}#[tokio::test]async fn session_configuration_apply_permission_profile_accepts_direct_write_roots() { let mut session_configuration = make_session_configuration_for_tests().await; let cwd = tempfile::tempdir().expect("create cwd"); session_configuration.environments = TurnEnvironmentSelections::new(cwd.path().abs(), Vec::new()); let external_write_dir = tempfile::tempdir().expect("create external write root"); let external_write_path = AbsolutePathBuf::from_absolute_path( codex_utils_absolute_path::canonicalize_preserving_symlinks(external_write_dir.path()) .expect("canonical temp dir"), ) .expect("canonical temp dir should be absolute"); let file_system_sandbox_policy = FileSystemSandboxPolicy::restricted(vec![FileSystemSandboxEntry { path: FileSystemPath::Path { path: external_write_path.clone(), }, access: FileSystemAccessMode::Write, }]); let permission_profile = PermissionProfile::from_runtime_permissions( &file_system_sandbox_policy, NetworkSandboxPolicy::Restricted, ); let updated = session_configuration .apply(&SessionSettingsUpdate { permission_profile: Some(permission_profile.clone()), ..Default::default() }) .expect("permission profile update should accept direct runtime permissions"); assert_eq!(updated.permission_profile(), permission_profile); assert_eq!( updated.file_system_sandbox_policy(), file_system_sandbox_policy ); assert_eq!( updated.sandbox_policy(), SandboxPolicy::WorkspaceWrite { writable_roots: vec![external_write_path], network_access: false, exclude_tmpdir_env_var: true, exclude_slash_tmp: true, } );}#[tokio::test]async fn session_configuration_apply_rebinds_symbolic_profile_to_updated_workspace_roots() { let mut session_configuration = make_session_configuration_for_tests().await; let old_root = tempfile::tempdir().expect("create old root"); let new_root = tempfile::tempdir().expect("create new root"); let profile_root = tempfile::tempdir().expect("create profile root"); let old_root = old_root.path().abs(); let new_root = new_root.path().abs(); let profile_root = profile_root.path().abs(); session_configuration.workspace_roots = vec![old_root.clone()]; let file_system_sandbox_policy = FileSystemSandboxPolicy::restricted(vec![FileSystemSandboxEntry { path: FileSystemPath::Special { value: FileSystemSpecialPath::project_roots(/*subpath*/ None), }, access: FileSystemAccessMode::Write, }]); let permission_profile = PermissionProfile::from_runtime_permissions( &file_system_sandbox_policy, NetworkSandboxPolicy::Restricted, ); let updated = session_configuration .apply(&SessionSettingsUpdate { workspace_roots: Some(vec![new_root.clone()]), permission_profile: Some(permission_profile), active_permission_profile: Some(ActivePermissionProfile::new("dev")), profile_workspace_roots: Some(vec![profile_root.clone()]), ..Default::default() }) .expect("permission profile update should succeed"); let updated_policy = updated.file_system_sandbox_policy(); assert!(updated_policy.can_write_path_with_cwd(new_root.as_path(), updated.cwd().as_path())); assert!(!updated_policy.can_write_path_with_cwd(old_root.as_path(), updated.cwd().as_path())); assert_eq!( updated.active_permission_profile(), Some(ActivePermissionProfile::new("dev")) ); assert_eq!(updated.profile_workspace_roots(), &[profile_root]);}#[tokio::test]async fn session_configuration_apply_retargets_implicit_workspace_root_on_cwd_update() { let mut session_configuration = make_session_configuration_for_tests().await; let old_root = tempfile::tempdir().expect("create old root"); let new_root = tempfile::tempdir().expect("create new root"); let extra_root = tempfile::tempdir().expect("create extra root"); let old_root = old_root.path().abs(); let new_root = new_root.path().abs(); let extra_root = extra_root.path().abs(); session_configuration.environments = TurnEnvironmentSelections::new(old_root.clone(), Vec::new()); session_configuration.workspace_roots = vec![old_root.clone(), extra_root.clone()]; let file_system_sandbox_policy = FileSystemSandboxPolicy::restricted(vec![FileSystemSandboxEntry { path: FileSystemPath::Special { value: FileSystemSpecialPath::project_roots(/*subpath*/ None), }, access: FileSystemAccessMode::Write, }]); let permission_profile = PermissionProfile::from_runtime_permissions( &file_system_sandbox_policy, NetworkSandboxPolicy::Restricted, ); session_configuration .set_permission_profile_for_tests(permission_profile) .expect("set permission profile"); let updated = session_configuration .apply(&SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new(new_root.clone(), Vec::new())), ..Default::default() }) .expect("cwd-only update should succeed"); assert_eq!( updated.workspace_roots, vec![new_root.clone(), extra_root.clone()] ); let updated_policy = updated.file_system_sandbox_policy(); assert!(updated_policy.can_write_path_with_cwd(new_root.as_path(), updated.cwd().as_path())); assert!(updated_policy.can_write_path_with_cwd(extra_root.as_path(), updated.cwd().as_path())); assert!(!updated_policy.can_write_path_with_cwd(old_root.as_path(), updated.cwd().as_path()));}#[tokio::test]async fn active_profile_update_rebuilds_network_proxy_config() -> std::io::Result<()> { let codex_home = tempfile::tempdir().expect("create codex home"); let cwd = tempfile::tempdir().expect("create cwd"); let permissions = PermissionsToml { entries: std::collections::BTreeMap::from([ ( "locked-down".to_string(), PermissionProfileToml { description: None, extends: None, workspace_roots: None, filesystem: Some(FilesystemPermissionsToml { glob_scan_max_depth: None, entries: std::collections::BTreeMap::from([( ":minimal".to_string(), FilesystemPermissionToml::Access(FileSystemAccessMode::Read), )]), }), network: None, }, ), ( "web-enabled".to_string(), PermissionProfileToml { description: None, extends: None, workspace_roots: None, filesystem: Some(FilesystemPermissionsToml { glob_scan_max_depth: None, entries: std::collections::BTreeMap::from([( ":minimal".to_string(), FilesystemPermissionToml::Access(FileSystemAccessMode::Read), )]), }), network: Some(NetworkToml { enabled: Some(true), proxy_url: Some("http://127.0.0.1:43128".to_string()), enable_socks5: Some(false), ..Default::default() }), }, ), ]), }; let base_config = ConfigToml { features: Some(toml::from_str("network_proxy = true").expect("valid features")), default_permissions: Some("locked-down".to_string()), permissions: Some(permissions), ..Default::default() }; std::fs::write( codex_home.path().join(codex_config::CONFIG_TOML_FILE), toml::to_string(&base_config).expect("serialize config"), )?; let locked_config = Arc::new( ConfigBuilder::default() .codex_home(codex_home.path().to_path_buf()) .harness_overrides(ConfigOverrides { cwd: Some(cwd.path().to_path_buf()), ..Default::default() }) .build() .await?, ); assert_ne!( locked_config .permissions .network .as_ref() .map(crate::config::NetworkProxySpec::proxy_host_and_port) .as_deref(), Some("127.0.0.1:43128") ); let selected_config = ConfigBuilder::default() .codex_home(codex_home.path().to_path_buf()) .harness_overrides(ConfigOverrides { cwd: Some(cwd.path().to_path_buf()), default_permissions: Some("web-enabled".to_string()), ..Default::default() }) .build() .await?; let mut session_configuration = make_session_configuration_for_tests().await; session_configuration.permission_profile_state = locked_config.permissions.permission_profile_state().clone(); session_configuration.original_config_do_not_use = Arc::clone(&locked_config); let updated = session_configuration .apply(&SessionSettingsUpdate { permission_profile: Some(selected_config.permissions.permission_profile().clone()), active_permission_profile: selected_config.permissions.active_permission_profile(), ..Default::default() }) .expect("active profile update should apply"); let network = updated .original_config_do_not_use .permissions .network .as_ref() .expect("selected profile proxy should become the session proxy config"); assert_eq!(network.proxy_host_and_port(), "127.0.0.1:43128"); assert!(!network.socks_enabled()); Ok(())}#[cfg_attr(windows, ignore)]#[tokio::test]async fn new_default_turn_uses_config_aware_skills_for_role_overrides() { let (session, _turn_context) = make_session_and_context().await; let parent_config = session.get_config().await; let codex_home = parent_config.codex_home.clone(); let skill_dir = codex_home.join("skills").join("demo"); std::fs::create_dir_all(&skill_dir).expect("create skill dir"); let skill_path = skill_dir.join("SKILL.md"); std::fs::write( &skill_path, "---\nname: demo-skill\ndescription: demo description\n---\n\n# Body\n", ) .expect("write skill"); let skill_fs = session .services .turn_environments .environment_manager() .default_environment() .map(|environment| environment.get_filesystem()) .unwrap_or_else(|| std::sync::Arc::clone(&codex_exec_server::LOCAL_FS)); let parent_outcome = session .services .skills_manager .skills_for_cwd( &crate::skills_load_input_from_config(&parent_config, Vec::new()), /*force_reload*/ true, Some(Arc::clone(&skill_fs)), ) .await; let parent_skill = parent_outcome .skills .iter() .find(|skill| skill.name == "demo-skill") .expect("demo skill should be discovered"); assert_eq!(parent_outcome.is_skill_enabled(parent_skill), true); let role_path = codex_home.join("skills-role.toml"); std::fs::write( &role_path, format!( r#"developer_instructions = "Stay focused"[[skills.config]]path = "{}"enabled = false"#, skill_path.display() ), ) .expect("write role config"); let mut child_config = (*parent_config).clone(); child_config.agent_roles.insert( "custom".to_string(), crate::config::AgentRoleConfig { description: None, config_file: Some(role_path.to_path_buf()), nickname_candidates: None, }, ); crate::agent::role::apply_role_to_config(&mut child_config, Some("custom")) .await .expect("custom role should apply"); { let mut state = session.state.lock().await; state.session_configuration.original_config_do_not_use = Arc::new(child_config); } let child_turn = session .new_default_turn_with_sub_id("role-skill-turn".to_string()) .await; let child_skill = child_turn .turn_skills .outcome .skills .iter() .find(|skill| skill.name == "demo-skill") .expect("demo skill should be discovered"); assert_eq!( child_turn.turn_skills.outcome.is_skill_enabled(child_skill), false );}#[tokio::test]async fn session_configuration_apply_retargets_legacy_workspace_root_on_cwd_update() { let mut session_configuration = make_session_configuration_for_tests().await; let workspace = tempfile::tempdir().expect("create temp dir"); let original_cwd = workspace.path().join("repo-a").abs(); let project_root = workspace.path().join("repo-b").abs(); session_configuration.environments = TurnEnvironmentSelections::new(original_cwd.clone(), Vec::new()); session_configuration.workspace_roots = vec![session_configuration.cwd().clone()]; let sandbox_policy = SandboxPolicy::WorkspaceWrite { writable_roots: Vec::new(), network_access: false, exclude_tmpdir_env_var: true, exclude_slash_tmp: true, }; let file_system_sandbox_policy = FileSystemSandboxPolicy::from_legacy_sandbox_policy_for_cwd( &sandbox_policy, session_configuration.cwd(), ); session_configuration .set_permission_profile_for_tests( PermissionProfile::from_runtime_permissions_with_enforcement( SandboxEnforcement::from_legacy_sandbox_policy(&sandbox_policy), &file_system_sandbox_policy, NetworkSandboxPolicy::from(&sandbox_policy), ), ) .expect("set permission profile"); let updated = session_configuration .apply(&SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new( project_root.clone(), Vec::new(), )), ..Default::default() }) .expect("cwd-only update should succeed"); assert_eq!(updated.workspace_roots, vec![project_root.clone()]); assert!( updated .file_system_sandbox_policy() .can_write_path_with_cwd(project_root.as_path(), updated.cwd().as_path()), "cwd-only update should keep the new cwd writable" ); assert!( !updated .file_system_sandbox_policy() .can_write_path_with_cwd(original_cwd.as_path(), updated.cwd().as_path()), "cwd-only update should not keep the old implicit cwd writable" );}#[tokio::test]async fn session_configuration_apply_preserves_absolute_cwd_write_root_on_cwd_update() { let mut session_configuration = make_session_configuration_for_tests().await; let workspace = tempfile::tempdir().expect("create temp dir"); let original_cwd = workspace.path().join("repo-a"); let next_cwd = workspace.path().join("repo-b"); std::fs::create_dir_all(&original_cwd).expect("create original cwd"); std::fs::create_dir_all(&next_cwd).expect("create next cwd"); let original_cwd = original_cwd.abs(); let next_cwd = next_cwd.abs(); session_configuration.environments = TurnEnvironmentSelections::new(original_cwd.clone(), Vec::new()); let file_system_sandbox_policy = FileSystemSandboxPolicy::restricted(vec![ FileSystemSandboxEntry { path: FileSystemPath::Special { value: FileSystemSpecialPath::Root, }, access: FileSystemAccessMode::Read, }, FileSystemSandboxEntry { path: FileSystemPath::Path { path: original_cwd.clone(), }, access: FileSystemAccessMode::Write, }, ]); session_configuration .set_permission_profile_for_tests( PermissionProfile::from_runtime_permissions_with_enforcement( SandboxEnforcement::Managed, &file_system_sandbox_policy, NetworkSandboxPolicy::Restricted, ), ) .expect("set permission profile"); let updated = session_configuration .apply(&SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new(next_cwd.clone(), Vec::new())), ..Default::default() }) .expect("cwd-only update should succeed"); assert_eq!( updated.file_system_sandbox_policy(), file_system_sandbox_policy ); assert!( updated .file_system_sandbox_policy() .can_write_path_with_cwd(original_cwd.as_path(), updated.cwd().as_path()), "absolute grant to the old cwd must remain writable" ); assert!( !updated .file_system_sandbox_policy() .can_write_path_with_cwd(next_cwd.as_path(), updated.cwd().as_path()), "cwd-only update must not reinterpret an absolute old-cwd grant as :workspace_roots" );}#[tokio::test]async fn session_update_settings_does_not_rewrite_sticky_environment_cwds() { let (session, turn_context) = make_session_and_context().await; #[allow(deprecated)] let updated_cwd = turn_context.cwd.join("project"); let current_environments = { let state = session.state.lock().await; state .session_configuration .environment_selections() .to_vec() }; let expected_environments = current_environments.clone(); std::fs::create_dir_all(updated_cwd.as_path()).expect("create project dir"); session .update_settings(SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new( updated_cwd.clone(), current_environments, )), ..Default::default() }) .await .expect("cwd update should succeed"); let (session_cwd, stored_environments) = { let state = session.state.lock().await; ( state.session_configuration.cwd().clone(), state .session_configuration .environment_selections() .to_vec(), ) }; let config = session.get_config().await; let next_turn = session.new_default_turn().await; assert_eq!(session_cwd, updated_cwd); assert_eq!(stored_environments, expected_environments); #[allow(deprecated)] let turn_cwd = turn_context.cwd.clone(); #[allow(deprecated)] let next_turn_cwd = next_turn.cwd.clone(); assert_eq!(config.cwd, turn_cwd); assert_eq!(next_turn_cwd, turn_cwd); assert_eq!(next_turn.config.cwd, turn_cwd);}#[tokio::test]async fn relative_cwd_update_without_environments_resolves_under_session_cwd() { let (session, _turn_context) = make_session_and_context().await; let original_cwd = { let mut state = session.state.lock().await; state.session_configuration.environments.environments = Vec::new(); state.session_configuration.cwd().clone() }; let updated_cwd = original_cwd.join("project"); std::fs::create_dir_all(updated_cwd.as_path()).expect("create project dir"); session .update_settings(SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new( updated_cwd.clone(), Vec::new(), )), ..Default::default() }) .await .expect("cwd update should succeed"); let state = session.state.lock().await; assert_eq!(state.session_configuration.cwd(), &updated_cwd); assert!( state .session_configuration .environment_selections() .is_empty() );}#[tokio::test]async fn environment_settings_preserve_explicit_primary_cwd() { let (session, _turn_context) = make_session_and_context().await; let (original_cwd, environment_cwd, environments) = { let mut state = session.state.lock().await; let original_cwd = state.session_configuration.cwd().clone(); let environment_cwd = original_cwd.join("environment"); let environments = vec![local(environment_cwd.clone())]; state.session_configuration.environments.environments = environments.clone(); (original_cwd, environment_cwd, environments) }; let updated_cwd = original_cwd.join("project"); std::fs::create_dir_all(updated_cwd.as_path()).expect("create project dir"); session .update_settings(SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new( updated_cwd.clone(), environments, )), ..Default::default() }) .await .expect("cwd update should succeed"); let state = session.state.lock().await; assert_eq!(state.session_configuration.cwd(), &updated_cwd); assert_eq!( state.session_configuration.environment_selections()[0].cwd, PathUri::from_abs_path(&environment_cwd) );}#[tokio::test]async fn absolute_cwd_update_with_turn_environment_is_allowed() { let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; let absolute_cwd = { let state = session.state.lock().await; state.session_configuration.cwd().join("absolute-turn") }; std::fs::create_dir_all(absolute_cwd.as_path()).expect("create absolute turn dir"); let turn_context = session .new_turn_with_sub_id( "sub-1".to_string(), SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new( absolute_cwd.clone(), vec![local(absolute_cwd.clone())], )), ..Default::default() }, ) .await .expect("absolute cwd with explicit environments should succeed"); #[allow(deprecated)] let turn_cwd = turn_context.cwd.clone(); assert_eq!(turn_cwd, absolute_cwd); assert_eq!(turn_context.config.cwd, absolute_cwd); assert_eq!(turn_context.environments.turn_environments.len(), 1);}#[tokio::test]async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { let codex_home = tempfile::tempdir().expect("create temp dir"); let mut config = build_test_config(codex_home.path()).await; config .features .enable(Feature::ShellZshFork) .expect("test config should allow shell_zsh_fork"); config.zsh_path = None; let config = Arc::new(config); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key")); let models_manager = models_manager_with_provider( config.codex_home.to_path_buf(), auth_manager.clone(), config.model_provider.clone(), ); let model = get_model_offline_for_tests(config.model.as_deref()); let model_info = construct_model_info_offline_for_tests(model.as_str(), &config.to_models_manager_config()); let collaboration_mode = CollaborationMode { mode: ModeKind::Default, settings: Settings { model, reasoning_effort: config.model_reasoning_effort.clone(), developer_instructions: None, }, }; let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config .base_instructions .clone() .unwrap_or_else(|| model_info.get_model_instructions(config.personality)), compact_prompt: config.compact_prompt.clone(), approval_policy: config.permissions.approval_policy.clone(), approvals_reviewer: config.approvals_reviewer, permission_profile_state: config.permissions.permission_profile_state().clone(), windows_sandbox_level: WindowsSandboxLevel::from_config(&config), environments: TurnEnvironmentSelections::new(config.cwd.clone(), Vec::new()), workspace_roots: config.workspace_roots.clone(), codex_home: config.codex_home.clone(), thread_name: None, original_config_do_not_use: Arc::clone(&config), metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, forked_from_thread_id: None, parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), user_shell_override: None, }; let (tx_event, _rx_event) = async_channel::unbounded(); let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let plugins_manager = Arc::new(PluginsManager::new(config.codex_home.to_path_buf())); let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager))); let skills_manager = Arc::new(SkillsManager::new( config.codex_home.clone(), /*bundled_skills_enabled*/ true, )); let environment_manager = Arc::new(EnvironmentManager::default_for_tests()); let result = Session::new( session_configuration, Arc::clone(&config), /*user_instructions*/ None, "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, Arc::new(ExecPolicyManager::default()), tx_event, agent_status_tx, InitialHistory::New, SessionSource::Exec, skills_manager, plugins_manager, mcp_manager, Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), codex_extension_api::ExtensionDataInit::default(), AgentControl::default(), environment_manager, /*inherited_environments*/ None, /*analytics_events_client*/ None, Arc::new(codex_thread_store::LocalThreadStore::new( codex_thread_store::LocalThreadStoreConfig::from_config(config.as_ref()), /*state_db*/ None, )), codex_rollout_trace::ThreadTraceContext::disabled(), /*attestation_provider*/ None, Some(config.multi_agent_version_from_features()), ) .await; let err = match result { Ok(_) => panic!("expected startup to fail"), Err(err) => err, }; let msg = format!("{err:#}"); assert!(msg.contains("zsh fork feature enabled, but no packaged zsh fork is available"));}// todo: use online model infopub(crate) async fn make_session_and_context() -> (Session, TurnContext) { let (tx_event, _rx_event) = async_channel::unbounded(); let codex_home = tempfile::tempdir().expect("create temp dir"); let config = build_test_config(codex_home.path()).await; let config = Arc::new(config); let thread_id = ThreadId::default(); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key")); let models_manager = models_manager_with_provider( config.codex_home.to_path_buf(), auth_manager.clone(), config.model_provider.clone(), ); let agent_control = AgentControl::default(); let exec_policy = Arc::new(ExecPolicyManager::default()); let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let model = get_model_offline_for_tests(config.model.as_deref()); let model_info = construct_model_info_offline_for_tests(model.as_str(), &config.to_models_manager_config()); let reasoning_effort = config.model_reasoning_effort.clone(); let collaboration_mode = CollaborationMode { mode: ModeKind::Default, settings: Settings { model, reasoning_effort, developer_instructions: None, }, }; let default_environments = vec![local(config.cwd.clone())]; let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config .base_instructions .clone() .unwrap_or_else(|| model_info.get_model_instructions(config.personality)), compact_prompt: config.compact_prompt.clone(), approval_policy: config.permissions.approval_policy.clone(), approvals_reviewer: config.approvals_reviewer, permission_profile_state: config.permissions.permission_profile_state().clone(), windows_sandbox_level: WindowsSandboxLevel::from_config(&config), environments: TurnEnvironmentSelections::new(config.cwd.clone(), default_environments), workspace_roots: config.workspace_roots.clone(), codex_home: config.codex_home.clone(), thread_name: None, original_config_do_not_use: Arc::clone(&config), metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, forked_from_thread_id: None, parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), user_shell_override: None, }; let per_turn_config = Session::build_per_turn_config(&session_configuration, session_configuration.cwd().clone()); let model_info = construct_model_info_offline_for_tests( session_configuration.collaboration_mode.model(), &per_turn_config.to_models_manager_config(), ); let session_telemetry = session_telemetry( thread_id, config.as_ref(), &model_info, session_configuration.session_source.clone(), ); let state = SessionState::new(session_configuration.clone()); let (environment_manager, resolved_environments) = resolved_environments_for_configuration(&session_configuration).await; let resolved_turn_environments = resolved_environments.clone(); let turn_environments = Arc::new(ThreadEnvironments::new( environment_manager, default_user_shell(), ShellSnapshot::disabled(), resolved_environments, )); let environment = Arc::clone( &resolved_turn_environments .primary() .expect("primary environment") .environment, ); let plugins_manager = Arc::new(PluginsManager::new(config.codex_home.to_path_buf())); let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager))); let skills_manager = Arc::new(SkillsManager::new( config.codex_home.clone(), /*bundled_skills_enabled*/ true, )); let network_approval = Arc::new(NetworkApprovalService::default()); let services = SessionServices { mcp_connection_manager: Arc::new(arc_swap::ArcSwap::from_pointee( McpConnectionManager::new_uninitialized_with_permission_profile( &config.permissions.approval_policy, config.permissions.permission_profile(), config.prefix_mcp_tool_names(), ), )), mcp_startup_cancellation_token: Mutex::new(CancellationToken::new()), unified_exec_manager: UnifiedExecProcessManager::new( config.background_terminal_max_timeout, ), shell_zsh_path: None, main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(), analytics_events_client: AnalyticsEventsClient::new( Arc::clone(&auth_manager), config.chatgpt_base_url.trim_end_matches('/').to_string(), config.analytics_enabled, ), hooks: arc_swap::ArcSwap::from_pointee(Hooks::new(HooksConfig { legacy_notify_argv: config.notify.clone(), ..HooksConfig::default() })), rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), user_shell: Arc::new(default_user_shell()), show_raw_agent_reasoning: config.show_raw_agent_reasoning, exec_policy, auth_manager: auth_manager.clone(), session_telemetry: session_telemetry.clone(), models_manager: Arc::clone(&models_manager), tool_approvals: Mutex::new(ApprovalStore::default()), guardian_rejections: Mutex::new(std::collections::HashMap::new()), guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), skills_manager, plugins_manager, mcp_manager, extensions: Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), session_extension_data: codex_extension_api::ExtensionData::new( agent_control.session_id().to_string(), ), thread_extension_data: codex_extension_api::ExtensionData::new(thread_id.to_string()), mcp_thread_init: codex_extension_api::ExtensionDataInit::default(), agent_control, network_proxy: arc_swap::ArcSwapOption::from(None), network_proxy_audit_metadata: crate::config::NetworkProxyAuditMetadata::default(), managed_network_requirements_configured: false, network_approval: Arc::clone(&network_approval), state_db: None, live_thread: None, thread_store: Arc::new(codex_thread_store::LocalThreadStore::new( codex_thread_store::LocalThreadStoreConfig::from_config(config.as_ref()), /*state_db*/ None, )), attestation_provider: None, model_client: ModelClient::new( Some(auth_manager.clone()), thread_id, session_configuration.provider.clone(), session_configuration.session_source.clone(), config.model_verbosity, config.features.enabled(Feature::EnableRequestCompression), config.features.enabled(Feature::RuntimeMetrics), Session::build_model_client_beta_features_header(config.as_ref()), /*attestation_provider*/ None, ), code_mode_service: crate::tools::code_mode::CodeModeService::new(), tool_search_handler_cache: Default::default(), turn_environments: Arc::clone(&turn_environments), }; let plugin_outcome = services .plugins_manager .plugins_for_config(&per_turn_config.plugins_config_input()) .await; let effective_skill_roots = plugin_outcome.effective_plugin_skill_roots(); let skills_input = crate::skills_load_input_from_config(&per_turn_config, effective_skill_roots); let skill_fs = environment.get_filesystem(); let skills_outcome = Arc::new( services .skills_manager .skills_for_config(&skills_input, Some(Arc::clone(&skill_fs))) .await, ); let turn_context = Session::make_turn_context( thread_id, SessionId::from(thread_id), Some(Arc::clone(&auth_manager)), &session_telemetry, session_configuration.provider.clone(), &session_configuration, config.multi_agent_version_from_features(), services.user_shell.as_ref(), services.shell_zsh_path.as_ref(), services.main_execve_wrapper_exe.as_ref(), per_turn_config, model_info, &models_manager, /*network*/ None, resolved_turn_environments, session_configuration.cwd().clone(), "turn_id".to_string(), skills_outcome, ); let session = Session { thread_id, installation_id: "11111111-1111-4111-8111-111111111111".to_string(), tx_event, agent_status: agent_status_tx, out_of_band_elicitation_paused: watch::channel(false).0, state: Mutex::new(state), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), multi_agent_version: OnceLock::from(config.multi_agent_version_from_features()), pending_mcp_server_refresh_config: Mutex::new(None), conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), input_queue: super::input_queue::InputQueue::new(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), services, next_internal_sub_id: AtomicU64::new(0), }; (session, turn_context)}async fn make_session_with_config( mutator: impl FnOnce(&mut Config),) -> anyhow::Result<Arc<Session>> { let (session, _rx_event) = make_session_with_config_and_rx(mutator).await?; Ok(session)}async fn load_latest_config_for_session(session: &Session) -> Config { let config = session.get_config().await; ConfigBuilder::default() .codex_home(config.codex_home.to_path_buf()) .fallback_cwd(Some(config.cwd.to_path_buf())) .build() .await .expect("load latest config for session")}async fn make_session_with_config_and_rx( mutator: impl FnOnce(&mut Config),) -> anyhow::Result<(Arc<Session>, async_channel::Receiver<Event>)> { let codex_home = tempfile::tempdir().expect("create temp dir"); let mut config = build_test_config(codex_home.path()).await; mutator(&mut config); let config = Arc::new(config); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key")); let models_manager = models_manager_with_provider( config.codex_home.to_path_buf(), auth_manager.clone(), config.model_provider.clone(), ); let model = get_model_offline_for_tests(config.model.as_deref()); let model_info = construct_model_info_offline_for_tests(model.as_str(), &config.to_models_manager_config()); let collaboration_mode = CollaborationMode { mode: ModeKind::Default, settings: Settings { model, reasoning_effort: config.model_reasoning_effort.clone(), developer_instructions: None, }, }; let default_environments = vec![local(config.cwd.clone())]; let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config .base_instructions .clone() .unwrap_or_else(|| model_info.get_model_instructions(config.personality)), compact_prompt: config.compact_prompt.clone(), approval_policy: config.permissions.approval_policy.clone(), approvals_reviewer: config.approvals_reviewer, permission_profile_state: config.permissions.permission_profile_state().clone(), windows_sandbox_level: WindowsSandboxLevel::from_config(&config), environments: TurnEnvironmentSelections::new(config.cwd.clone(), default_environments), workspace_roots: config.workspace_roots.clone(), codex_home: config.codex_home.clone(), thread_name: None, original_config_do_not_use: Arc::clone(&config), metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, forked_from_thread_id: None, parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), user_shell_override: None, }; let (tx_event, rx_event) = async_channel::unbounded(); let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let plugins_manager = Arc::new(PluginsManager::new(config.codex_home.to_path_buf())); let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager))); let skills_manager = Arc::new(SkillsManager::new( config.codex_home.clone(), /*bundled_skills_enabled*/ true, )); let environment_manager = Arc::new(EnvironmentManager::default_for_tests()); let session = Session::new( session_configuration, Arc::clone(&config), /*user_instructions*/ None, "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, Arc::new(ExecPolicyManager::default()), tx_event, agent_status_tx, InitialHistory::New, SessionSource::Exec, skills_manager, plugins_manager, mcp_manager, Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), codex_extension_api::ExtensionDataInit::default(), AgentControl::default(), environment_manager, /*inherited_environments*/ None, /*analytics_events_client*/ None, Arc::new(codex_thread_store::LocalThreadStore::new( codex_thread_store::LocalThreadStoreConfig::from_config(config.as_ref()), /*state_db*/ None, )), codex_rollout_trace::ThreadTraceContext::disabled(), /*attestation_provider*/ None, Some(config.multi_agent_version_from_features()), ) .await?; Ok((session, rx_event))}async fn make_session_with_history_source_and_agent_control_and_rx( initial_history: InitialHistory, session_source: SessionSource, agent_control: AgentControl,) -> anyhow::Result<(Arc<Session>, async_channel::Receiver<Event>)> { let codex_home = tempfile::tempdir().expect("create temp dir"); let mut config = build_test_config(codex_home.path()).await; config.ephemeral = true; let config = Arc::new(config); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key")); let models_manager = models_manager_with_provider( config.codex_home.to_path_buf(), auth_manager.clone(), config.model_provider.clone(), ); let model = get_model_offline_for_tests(config.model.as_deref()); let model_info = construct_model_info_offline_for_tests(model.as_str(), &config.to_models_manager_config()); let collaboration_mode = CollaborationMode { mode: ModeKind::Default, settings: Settings { model, reasoning_effort: config.model_reasoning_effort.clone(), developer_instructions: None, }, }; let default_environments = vec![local(config.cwd.clone())]; let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config .base_instructions .clone() .unwrap_or_else(|| model_info.get_model_instructions(config.personality)), compact_prompt: config.compact_prompt.clone(), approval_policy: config.permissions.approval_policy.clone(), approvals_reviewer: config.approvals_reviewer, permission_profile_state: config.permissions.permission_profile_state().clone(), windows_sandbox_level: WindowsSandboxLevel::from_config(&config), environments: TurnEnvironmentSelections::new(config.cwd.clone(), default_environments), workspace_roots: config.workspace_roots.clone(), codex_home: config.codex_home.clone(), thread_name: None, original_config_do_not_use: Arc::clone(&config), metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, session_source: session_source.clone(), forked_from_thread_id: None, parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), user_shell_override: None, }; let (tx_event, rx_event) = async_channel::unbounded(); let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let plugins_manager = Arc::new(PluginsManager::new(config.codex_home.to_path_buf())); let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager))); let skills_manager = Arc::new(SkillsManager::new( config.codex_home.clone(), /*bundled_skills_enabled*/ true, )); let environment_manager = Arc::new(EnvironmentManager::default_for_tests()); let session = Session::new( session_configuration, Arc::clone(&config), /*user_instructions*/ None, "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, Arc::new(ExecPolicyManager::default()), tx_event, agent_status_tx, initial_history, session_source, skills_manager, plugins_manager, mcp_manager, Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), codex_extension_api::ExtensionDataInit::default(), agent_control, environment_manager, /*inherited_environments*/ None, /*analytics_events_client*/ None, Arc::new(codex_thread_store::LocalThreadStore::new( codex_thread_store::LocalThreadStoreConfig::from_config(config.as_ref()), Some( codex_state::StateRuntime::init( config.sqlite_home.clone(), config.model_provider_id.clone(), ) .await .expect("state db should initialize"), ), )), codex_rollout_trace::ThreadTraceContext::disabled(), /*attestation_provider*/ None, Some(config.multi_agent_version_from_features()), ) .await?; Ok((session, rx_event))}#[tokio::test]async fn resumed_root_session_uses_thread_id_as_session_id() { let thread_id = ThreadId::new(); let (session, rx_event) = make_session_with_history_source_and_agent_control_and_rx( InitialHistory::Resumed(ResumedHistory { conversation_id: thread_id, history: Vec::new(), rollout_path: None, }), SessionSource::Exec, AgentControl::default(), ) .await .expect("resume should succeed"); assert_eq!(session.thread_id(), thread_id); assert_eq!(session.session_id(), SessionId::from(thread_id)); let event = rx_event.recv().await.expect("session configured event"); let EventMsg::SessionConfigured(event) = event.msg else { panic!("expected session configured event"); }; assert_eq!(event.session_id, SessionId::from(thread_id)); assert_eq!(event.thread_id, thread_id);}#[tokio::test]async fn resumed_subagent_session_keeps_inherited_session_id() { let parent_thread_id = ThreadId::new(); let parent_session_id = SessionId::from(parent_thread_id); let thread_id = ThreadId::new(); let session_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }); let (session, rx_event) = make_session_with_history_source_and_agent_control_and_rx( InitialHistory::Resumed(ResumedHistory { conversation_id: thread_id, history: Vec::new(), rollout_path: None, }), session_source, AgentControl::default().with_session_id(parent_session_id, /*max_threads*/ usize::MAX), ) .await .expect("resume should succeed"); assert_eq!(session.thread_id(), thread_id); assert_eq!(session.session_id(), parent_session_id); let event = rx_event.recv().await.expect("session configured event"); let EventMsg::SessionConfigured(event) = event.msg else { panic!("expected session configured event"); }; assert_eq!(event.session_id, parent_session_id); assert_eq!(event.thread_id, thread_id);}#[tokio::test]async fn notify_request_permissions_response_ignores_unmatched_call_id() { let (session, _turn_context) = make_session_and_context().await; *session.active_turn.lock().await = Some(ActiveTurn::default()); session .notify_request_permissions_response( "missing", codex_protocol::request_permissions::RequestPermissionsResponse { permissions: RequestPermissionProfile { network: Some(codex_protocol::models::NetworkPermissions { enabled: Some(true), }), ..RequestPermissionProfile::default() }, scope: PermissionGrantScope::Turn, strict_auto_review: false, }, ) .await; assert_eq!( session .granted_turn_permissions(codex_exec_server::LOCAL_ENVIRONMENT_ID) .await, None );}#[tokio::test]async fn record_granted_request_permissions_for_turn_uses_originating_turn() { let (session, _turn_context) = make_session_and_context().await; let originating_active_turn = ActiveTurn::default(); let originating_turn_state = Arc::clone(&originating_active_turn.turn_state); *session.active_turn.lock().await = Some(originating_active_turn); let current_active_turn = ActiveTurn::default(); let current_turn_state = Arc::clone(¤t_active_turn.turn_state); *session.active_turn.lock().await = Some(current_active_turn); let requested_permissions = RequestPermissionProfile { network: Some(codex_protocol::models::NetworkPermissions { enabled: Some(true), }), ..RequestPermissionProfile::default() }; session .record_granted_request_permissions_for_turn( &codex_protocol::request_permissions::RequestPermissionsResponse { permissions: requested_permissions.clone(), scope: PermissionGrantScope::Turn, strict_auto_review: false, }, codex_exec_server::LOCAL_ENVIRONMENT_ID, Some(&originating_turn_state), ) .await; assert_eq!( originating_turn_state .lock() .await .granted_permissions(codex_exec_server::LOCAL_ENVIRONMENT_ID), Some(requested_permissions.into()) ); assert_eq!( current_turn_state .lock() .await .granted_permissions(codex_exec_server::LOCAL_ENVIRONMENT_ID), None ); assert_eq!( session .granted_turn_permissions(codex_exec_server::LOCAL_ENVIRONMENT_ID) .await, None );}#[tokio::test]async fn request_permission_grants_are_environment_keyed() { let (session, _turn_context) = make_session_and_context().await; let originating_active_turn = ActiveTurn::default(); let originating_turn_state = Arc::clone(&originating_active_turn.turn_state); *session.active_turn.lock().await = Some(originating_active_turn); let requested_permissions = RequestPermissionProfile { network: Some(codex_protocol::models::NetworkPermissions { enabled: Some(true), }), ..RequestPermissionProfile::default() }; session .record_granted_request_permissions_for_turn( &codex_protocol::request_permissions::RequestPermissionsResponse { permissions: requested_permissions.clone(), scope: PermissionGrantScope::Turn, strict_auto_review: false, }, "remote", Some(&originating_turn_state), ) .await; { let turn_state = originating_turn_state.lock().await; assert_eq!( turn_state.granted_permissions("remote"), Some(requested_permissions.clone().into()) ); assert_eq!(turn_state.granted_permissions("local"), None); } session .record_granted_request_permissions_for_turn( &codex_protocol::request_permissions::RequestPermissionsResponse { permissions: requested_permissions.clone(), scope: PermissionGrantScope::Session, strict_auto_review: false, }, "remote", /*originating_turn_state*/ None, ) .await; assert_eq!( session.granted_session_permissions("remote").await, Some(requested_permissions.into()) ); assert_eq!(session.granted_session_permissions("local").await, None);}#[tokio::test]async fn enable_strict_auto_review_for_turn_uses_originating_turn() { let (session, _turn_context) = make_session_and_context().await; let originating_active_turn = ActiveTurn::default(); let originating_turn_state = Arc::clone(&originating_active_turn.turn_state); *session.active_turn.lock().await = Some(originating_active_turn); let requested_permissions = RequestPermissionProfile { network: Some(codex_protocol::models::NetworkPermissions { enabled: Some(true), }), ..RequestPermissionProfile::default() }; session .record_granted_request_permissions_for_turn( &codex_protocol::request_permissions::RequestPermissionsResponse { permissions: requested_permissions.clone(), scope: PermissionGrantScope::Turn, strict_auto_review: true, }, codex_exec_server::LOCAL_ENVIRONMENT_ID, Some(&originating_turn_state), ) .await; assert!( originating_turn_state .lock() .await .strict_auto_review_enabled() );}#[test]fn strict_auto_review_session_scope_grants_no_permissions() { let requested_permissions = RequestPermissionProfile { network: Some(codex_protocol::models::NetworkPermissions { enabled: Some(true), }), ..RequestPermissionProfile::default() }; let response = Session::normalize_request_permissions_response( requested_permissions.clone(), codex_protocol::request_permissions::RequestPermissionsResponse { permissions: requested_permissions, scope: PermissionGrantScope::Session, strict_auto_review: true, }, std::path::Path::new("/tmp"), ); assert_eq!( response, codex_protocol::request_permissions::RequestPermissionsResponse { permissions: RequestPermissionProfile::default(), scope: PermissionGrantScope::Turn, strict_auto_review: false, } );}#[tokio::test]async fn request_permissions_emits_event_when_granular_policy_allows_requests() { let (session, mut turn_context, rx) = make_session_and_context_with_rx().await; *session.active_turn.lock().await = Some(ActiveTurn::default()); Arc::get_mut(&mut turn_context) .expect("single thread settings ref") .approval_policy .set(AskForApproval::Granular(GranularApprovalConfig { sandbox_approval: true, rules: true, skill_approval: true, request_permissions: true, mcp_elicitations: true, })) .expect("test setup should allow updating approval policy"); let session = Arc::new(session); let turn_context = Arc::new(turn_context); let call_id = "call-1".to_string(); let expected_response = codex_protocol::request_permissions::RequestPermissionsResponse { permissions: RequestPermissionProfile { network: Some(codex_protocol::models::NetworkPermissions { enabled: Some(true), }), ..RequestPermissionProfile::default() }, scope: PermissionGrantScope::Turn, strict_auto_review: false, }; let handle = tokio::spawn({ let session = Arc::clone(&session); let turn_context = Arc::clone(&turn_context); let call_id = call_id.clone(); async move { let environment = turn_context .environments .primary() .expect("primary environment") .selection(); session .request_permissions_for_environment( &turn_context, call_id, codex_protocol::request_permissions::RequestPermissionsArgs { environment_id: None, reason: Some("need network".to_string()), permissions: RequestPermissionProfile { network: Some(codex_protocol::models::NetworkPermissions { enabled: Some(true), }), ..RequestPermissionProfile::default() }, }, environment, CancellationToken::new(), ) .await } }); let request_event = tokio::time::timeout(StdDuration::from_secs(1), rx.recv()) .await .expect("request_permissions event timed out") .expect("request_permissions event missing"); let EventMsg::RequestPermissions(request) = request_event.msg else { panic!("expected request_permissions event"); }; assert_eq!(request.call_id, call_id); assert_eq!( request.environment_id.as_deref(), Some(codex_exec_server::LOCAL_ENVIRONMENT_ID) ); #[allow(deprecated)] let turn_cwd = turn_context.cwd.clone(); assert_eq!(request.cwd, Some(turn_cwd)); session .notify_request_permissions_response(&request.call_id, expected_response.clone()) .await; let response = tokio::time::timeout(StdDuration::from_secs(1), handle) .await .expect("request_permissions future timed out") .expect("request_permissions join error"); assert_eq!(response, Some(expected_response));}#[tokio::test]async fn request_permissions_tool_resolves_relative_paths_against_selected_environment() { let (session, mut turn_context, rx) = make_session_and_context_with_rx().await; *session.active_turn.lock().await = Some(ActiveTurn::default()); let environment_cwd = { #[allow(deprecated)] let legacy_cwd = turn_context.cwd.clone(); legacy_cwd.join("request-permissions-environment") }; std::fs::create_dir_all(environment_cwd.as_path()).expect("create environment cwd"); let turn_context_mut = Arc::get_mut(&mut turn_context).expect("single thread settings ref"); turn_context_mut .approval_policy .set(AskForApproval::Granular(GranularApprovalConfig { sandbox_approval: true, rules: true, skill_approval: true, request_permissions: true, mcp_elicitations: true, })) .expect("test setup should allow updating approval policy"); let current_environment = turn_context_mut.environments.turn_environments[0].clone(); turn_context_mut.environments.turn_environments[0] = TurnEnvironment::new( "remote".to_string(), current_environment.environment, PathUri::from_abs_path(&environment_cwd), current_environment.shell, ); let call_id = "call-1".to_string(); let handler = RequestPermissionsHandler; let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); let handle = tokio::spawn({ let session = Arc::clone(&session); let turn_context = Arc::clone(&turn_context); let tracker = Arc::clone(&tracker); let call_id = call_id.clone(); async move { handler .handle(ToolInvocation { session, turn: turn_context, cancellation_token: CancellationToken::new(), tracker, call_id, tool_name: codex_tools::ToolName::plain("request_permissions"), source: ToolCallSource::Direct, payload: ToolPayload::Function { arguments: json!({ "environment_id": "remote", "reason": "need write", "permissions": { "file_system": { "entries": [{ "path": { "type": "path", "path": "relative.txt", }, "access": "write", }], }, }, }) .to_string(), }, }) .await } }); let request_event = tokio::time::timeout(StdDuration::from_secs(1), rx.recv()) .await .expect("request_permissions event timed out") .expect("request_permissions event missing"); let EventMsg::RequestPermissions(request) = request_event.msg else { panic!("expected request_permissions event"); }; let expected_permissions = RequestPermissionProfile { file_system: Some(FileSystemPermissions { entries: vec![FileSystemSandboxEntry { path: FileSystemPath::Path { path: environment_cwd.join("relative.txt"), }, access: FileSystemAccessMode::Write, }], glob_scan_max_depth: None, }), ..Default::default() }; assert_eq!(request.environment_id.as_deref(), Some("remote")); assert_eq!(request.permissions, expected_permissions); session .notify_request_permissions_response( &request.call_id, codex_protocol::request_permissions::RequestPermissionsResponse { permissions: request.permissions, scope: PermissionGrantScope::Turn, strict_auto_review: false, }, ) .await; tokio::time::timeout(StdDuration::from_secs(1), handle) .await .expect("request_permissions handler timed out") .expect("request_permissions handler join error") .expect("request_permissions handler should succeed");}#[tokio::test]async fn request_permissions_tool_rejects_unknown_environment_id() { let (session, turn_context) = make_session_and_context().await; let result = RequestPermissionsHandler .handle(ToolInvocation { session: Arc::new(session), turn: Arc::new(turn_context), cancellation_token: CancellationToken::new(), tracker: Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())), call_id: "call-1".to_string(), tool_name: codex_tools::ToolName::plain("request_permissions"), source: ToolCallSource::Direct, payload: ToolPayload::Function { arguments: json!({ "environment_id": "missing", "permissions": { "network": { "enabled": true, }, }, }) .to_string(), }, }) .await; let Err(FunctionCallError::RespondToModel(output)) = result else { panic!("expected unknown environment id to be rejected"); }; assert_eq!(output, "unknown turn environment id `missing`");}#[tokio::test]async fn request_permissions_response_materializes_session_cwd_grants_before_recording() { let (session, mut turn_context, rx) = make_session_and_context_with_rx().await; *session.active_turn.lock().await = Some(ActiveTurn::default()); Arc::get_mut(&mut turn_context) .expect("single thread settings ref") .approval_policy .set(AskForApproval::Granular(GranularApprovalConfig { sandbox_approval: true, rules: true, skill_approval: true, request_permissions: true, mcp_elicitations: true, })) .expect("test setup should allow updating approval policy"); let session = Arc::new(session); let turn_context = Arc::new(turn_context); let call_id = "call-1".to_string(); let requested_permissions = RequestPermissionProfile { file_system: Some(FileSystemPermissions { entries: vec![FileSystemSandboxEntry { path: FileSystemPath::Special { value: FileSystemSpecialPath::project_roots(/*subpath*/ None), }, access: FileSystemAccessMode::Write, }], glob_scan_max_depth: None, }), ..Default::default() }; let handle = tokio::spawn({ let session = Arc::clone(&session); let turn_context = Arc::clone(&turn_context); let call_id = call_id.clone(); let requested_permissions = requested_permissions.clone(); async move { let environment = turn_context .environments .primary() .expect("primary environment") .selection(); session .request_permissions_for_environment( &turn_context, call_id, codex_protocol::request_permissions::RequestPermissionsArgs { environment_id: None, reason: Some("need cwd write".to_string()), permissions: requested_permissions, }, environment, CancellationToken::new(), ) .await } }); let request_event = tokio::time::timeout(StdDuration::from_secs(1), rx.recv()) .await .expect("request_permissions event timed out") .expect("request_permissions event missing"); let EventMsg::RequestPermissions(request) = request_event.msg else { panic!("expected request_permissions event"); }; assert_eq!( request.environment_id.as_deref(), Some(codex_exec_server::LOCAL_ENVIRONMENT_ID) ); let request_cwd = request.cwd.clone().expect("request cwd"); session .notify_request_permissions_response( &request.call_id, codex_protocol::request_permissions::RequestPermissionsResponse { permissions: request.permissions, scope: PermissionGrantScope::Session, strict_auto_review: false, }, ) .await; let expected_permissions = RequestPermissionProfile { file_system: Some(FileSystemPermissions::from_read_write_roots( /*read*/ None, Some(vec![request_cwd]), )), ..Default::default() }; let expected_response = codex_protocol::request_permissions::RequestPermissionsResponse { permissions: expected_permissions.clone(), scope: PermissionGrantScope::Session, strict_auto_review: false, }; let response = tokio::time::timeout(StdDuration::from_secs(1), handle) .await .expect("request_permissions future timed out") .expect("request_permissions join error"); assert_eq!(response, Some(expected_response)); assert_eq!( session .granted_session_permissions(codex_exec_server::LOCAL_ENVIRONMENT_ID) .await, Some(expected_permissions.into()) );}#[tokio::test]async fn request_permissions_is_auto_denied_when_granular_policy_blocks_tool_requests() { let (session, mut turn_context, rx) = make_session_and_context_with_rx().await; *session.active_turn.lock().await = Some(ActiveTurn::default()); Arc::get_mut(&mut turn_context) .expect("single thread settings ref") .approval_policy .set(AskForApproval::Granular(GranularApprovalConfig { sandbox_approval: true, rules: true, skill_approval: true, request_permissions: false, mcp_elicitations: true, })) .expect("test setup should allow updating approval policy"); let session = Arc::new(session); let turn_context = Arc::new(turn_context); let call_id = "call-1".to_string(); let environment = turn_context .environments .primary() .expect("primary environment") .selection(); let response = session .request_permissions_for_environment( &turn_context, call_id, codex_protocol::request_permissions::RequestPermissionsArgs { environment_id: None, reason: Some("need network".to_string()), permissions: RequestPermissionProfile { network: Some(codex_protocol::models::NetworkPermissions { enabled: Some(true), }), ..RequestPermissionProfile::default() }, }, environment, CancellationToken::new(), ) .await; assert_eq!( response, Some( codex_protocol::request_permissions::RequestPermissionsResponse { permissions: RequestPermissionProfile::default(), scope: PermissionGrantScope::Turn, strict_auto_review: false, } ) ); assert!( tokio::time::timeout(StdDuration::from_millis(100), rx.recv()) .await .is_err(), "request_permissions should not emit an event when granular.request_permissions is false" );}#[tokio::test]async fn submit_with_id_captures_current_span_trace_context() { let (session, _turn_context) = make_session_and_context().await; let (tx_sub, rx_sub) = async_channel::bounded(1); let (_tx_event, rx_event) = async_channel::unbounded(); let (_agent_status_tx, agent_status) = watch::channel(AgentStatus::PendingInit); let codex = Codex { tx_sub, rx_event, agent_status, session: Arc::new(session), session_loop_termination: completed_session_loop_termination(), }; let _trace_test_context = install_test_tracing("codex-core-tests"); let request_parent = W3cTraceContext { traceparent: Some("00-00000000000000000000000000000011-0000000000000022-01".into()), tracestate: Some("vendor=value".into()), }; let request_span = info_span!("app_server.request"); assert!(set_parent_from_w3c_trace_context( &request_span, &request_parent )); let expected_trace = async { let expected_trace = current_span_w3c_trace_context().expect("current span should have trace context"); codex .submit_with_id(Submission { id: "sub-1".into(), op: Op::Interrupt, client_user_message_id: None, trace: None, }) .await .expect("submit should succeed"); expected_trace } .instrument(request_span) .await; let submitted = rx_sub.recv().await.expect("submission"); assert_eq!(submitted.trace, Some(expected_trace));}#[tokio::test]async fn new_default_turn_captures_current_span_trace_id() { let (session, _turn_context) = make_session_and_context().await; let _trace_test_context = install_test_tracing("codex-core-tests"); let request_parent = W3cTraceContext { traceparent: Some("00-00000000000000000000000000000011-0000000000000022-01".into()), tracestate: Some("vendor=value".into()), }; let request_span = info_span!("app_server.request"); assert!(set_parent_from_w3c_trace_context( &request_span, &request_parent )); let turn_trace_id = async { let expected_trace_id = Span::current() .context() .span() .span_context() .trace_id() .to_string(); let turn_context = session.new_default_turn().await; assert_eq!(turn_context.trace_id, Some(expected_trace_id)); turn_context.trace_id.clone() } .instrument(request_span) .await; assert_eq!( turn_trace_id.as_deref(), Some("00000000000000000000000000000011") );}#[test]fn submission_dispatch_span_prefers_submission_trace_context() { let _trace_test_context = install_test_tracing("codex-core-tests"); let ambient_parent = W3cTraceContext { traceparent: Some("00-00000000000000000000000000000033-0000000000000044-01".into()), tracestate: None, }; let ambient_span = info_span!("ambient"); assert!(set_parent_from_w3c_trace_context( &ambient_span, &ambient_parent )); let submission_trace = W3cTraceContext { traceparent: Some("00-00000000000000000000000000000055-0000000000000066-01".into()), tracestate: Some("vendor=value".into()), }; let dispatch_span = ambient_span.in_scope(|| { submission_dispatch_span(&Submission { id: "sub-1".into(), op: Op::Interrupt, client_user_message_id: None, trace: Some(submission_trace), }) }); let trace_id = dispatch_span.context().span().span_context().trace_id(); assert_eq!( trace_id, TraceId::from_hex("00000000000000000000000000000055").expect("trace id") );}#[test]fn submission_dispatch_span_uses_debug_for_realtime_audio() { let _trace_test_context = install_test_tracing("codex-core-tests"); let dispatch_span = submission_dispatch_span(&Submission { id: "sub-1".into(), op: Op::RealtimeConversationAudio(ConversationAudioParams { frame: RealtimeAudioFrame { data: "ZmFrZQ==".into(), sample_rate: 16_000, num_channels: 1, samples_per_channel: Some(160), item_id: None, }, }), client_user_message_id: None, trace: None, }); assert_eq!( dispatch_span.metadata().expect("span metadata").level(), &tracing::Level::DEBUG );}#[test]fn op_kind_for_input_and_context_ops() { assert_eq!( Op::UserInput { items: vec![], final_output_json_schema: None, responsesapi_client_metadata: None, additional_context: Default::default(), thread_settings: Default::default(), } .kind(), "user_input" ); assert_eq!( Op::ThreadSettings { thread_settings: ThreadSettingsOverrides::default(), } .kind(), "thread_settings" );}#[tokio::test]async fn user_turn_updates_approvals_reviewer() { let (session, turn_context, _rx) = make_session_and_context_with_rx().await; let config = session.get_config().await; handlers::user_input_or_turn( &session, "sub-1".to_string(), Op::UserInput { items: vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], final_output_json_schema: None, responsesapi_client_metadata: None, additional_context: Default::default(), thread_settings: codex_protocol::protocol::ThreadSettingsOverrides { environments: Some(local_selections(config.cwd.clone())), approval_policy: Some(config.permissions.approval_policy.value()), approvals_reviewer: Some(codex_config::types::ApprovalsReviewer::AutoReview), sandbox_policy: Some(config.legacy_sandbox_policy()), summary: config.model_reasoning_summary, personality: config.personality, collaboration_mode: Some(codex_protocol::config_types::CollaborationMode { mode: codex_protocol::config_types::ModeKind::Default, settings: codex_protocol::config_types::Settings { model: turn_context.model_info.slug.clone(), reasoning_effort: config.model_reasoning_effort.clone(), developer_instructions: None, }, }), ..Default::default() }, }, /*client_user_message_id*/ None, ) .await; let state = session.state.lock().await; assert_eq!( state.session_configuration.approvals_reviewer, codex_config::types::ApprovalsReviewer::AutoReview );}#[tokio::test]async fn turn_environments_set_primary_environment() { let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; let selected_cwd = AbsolutePathBuf::try_from(session.get_config().await.cwd.as_path().join("selected")) .expect("absolute path"); let turn_context = session .new_turn_with_sub_id( "sub-1".to_string(), SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new( selected_cwd.clone(), vec![local(selected_cwd.clone())], )), ..Default::default() }, ) .await .expect("turn should start"); let turn_environments = &turn_context.environments; assert_eq!(turn_environments.turn_environments.len(), 1); let turn_environment = turn_context .environments .primary() .expect("primary environment should be set"); assert!(std::sync::Arc::ptr_eq( &turn_environment.environment, &turn_environments.turn_environments[0].environment )); assert!(!turn_context.environments.turn_environments.is_empty()); #[allow(deprecated)] let turn_cwd = turn_context.cwd.clone(); assert_eq!(turn_cwd.as_path(), selected_cwd.as_path()); assert_eq!(turn_context.config.cwd.as_path(), selected_cwd.as_path()); let stored_environment = { session .services .turn_environments .snapshot() .await .primary_environment() .expect("stored primary environment") }; assert!(Arc::ptr_eq( &stored_environment, &turn_environment.environment )); let default_turn = session.new_default_turn().await; assert!(Arc::ptr_eq( &stored_environment, &default_turn .environments .primary() .expect("default turn primary environment") .environment ));}#[tokio::test]async fn default_turn_does_not_overlay_legacy_fallback_cwd_onto_stored_thread_environments() { let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; let session_cwd = session.get_config().await.cwd.clone(); let selected_cwd = AbsolutePathBuf::try_from(session_cwd.as_path().join("selected")).expect("absolute path"); session .services .turn_environments .update_selections(&[local(selected_cwd.clone())]); { let mut state = session.state.lock().await; state.session_configuration.environments.environments = vec![local(selected_cwd.clone())]; } let turn_context = session.new_default_turn().await; let turn_environments = &turn_context.environments; assert_eq!(turn_environments.turn_environments.len(), 1); let turn_environment = turn_context .environments .primary() .expect("primary environment should be set"); assert!(std::sync::Arc::ptr_eq( &turn_environment.environment, &turn_environments.turn_environments[0].environment )); #[allow(deprecated)] let turn_cwd = turn_context.cwd.clone(); assert_eq!(turn_cwd, selected_cwd); assert_eq!(turn_context.config.cwd, selected_cwd);}#[tokio::test]async fn default_turn_honors_empty_stored_thread_environments() { let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; let session_cwd = session.get_config().await.cwd.clone(); session.services.turn_environments.update_selections(&[]); { let mut state = session.state.lock().await; state.session_configuration.environments.environments = Vec::new(); } let turn_context = session.new_default_turn().await; assert!(turn_context.environments.primary().is_none()); assert!(turn_context.environments.turn_environments.is_empty()); #[allow(deprecated)] let turn_cwd = turn_context.cwd.clone(); assert_eq!(turn_cwd, session_cwd); assert_eq!(turn_context.config.cwd, session_cwd); assert_eq!(turn_context.environments.turn_environments.len(), 0);}#[tokio::test]async fn primary_environment_uses_first_turn_environment() { let (_session, mut turn_context) = make_session_and_context().await; let first_environment = turn_context.environments.turn_environments[0].clone(); #[allow(deprecated)] let second_cwd = turn_context.cwd.join("second"); let second_cwd_uri = codex_utils_path_uri::PathUri::from_abs_path(&second_cwd); turn_context .environments .turn_environments .push(TurnEnvironment::new( "second".to_string(), Arc::clone(&first_environment.environment), second_cwd_uri.clone(), /*shell*/ None, )); assert_eq!( turn_context .environments .primary() .expect("primary environment") .environment_id, first_environment.environment_id ); assert_eq!( turn_context .environments .turn_environments .iter() .find(|environment| environment.environment_id == "second") .expect("second environment") .cwd(), &second_cwd_uri ); assert_eq!(turn_context.environments.turn_environments.len(), 2); assert_eq!( turn_context.environments.turn_environments[1].cwd(), &second_cwd_uri );}#[tokio::test]async fn empty_turn_environments_clear_primary_environment() { let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; let turn_context = session .new_turn_with_sub_id( "sub-1".to_string(), SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new( session.get_config().await.cwd.clone(), vec![], )), ..Default::default() }, ) .await .expect("turn should start"); assert!(turn_context.environments.primary().is_none()); assert!(turn_context.environments.turn_environments.is_empty()); #[allow(deprecated)] let turn_cwd = turn_context.cwd.clone(); assert_eq!(turn_cwd, session.get_config().await.cwd); assert_eq!(turn_context.config.cwd, session.get_config().await.cwd);}#[tokio::test]async fn spawn_task_turn_span_inherits_dispatch_trace_context() { struct TraceCaptureTask { captured_trace: Arc<std::sync::Mutex<Option<W3cTraceContext>>>, } impl SessionTask for TraceCaptureTask { fn kind(&self) -> TaskKind { TaskKind::Regular } fn span_name(&self) -> &'static str { "session_task.trace_capture" } async fn run( self: Arc<Self>, _session: Arc<SessionTaskContext>, _ctx: Arc<TurnContext>, _input: Vec<TurnInput>, _cancellation_token: CancellationToken, ) -> Option<String> { let mut trace = self .captured_trace .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); *trace = current_span_w3c_trace_context(); None } } let _trace_test_context = install_test_tracing("codex-core-tests"); let request_parent = W3cTraceContext { traceparent: Some("00-00000000000000000000000000000011-0000000000000022-01".into()), tracestate: Some("vendor=value".into()), }; let request_span = tracing::info_span!("app_server.request"); assert!(set_parent_from_w3c_trace_context( &request_span, &request_parent )); let submission_trace = async { current_span_w3c_trace_context().expect("request span should have trace context") } .instrument(request_span) .await; let dispatch_span = submission_dispatch_span(&Submission { id: "sub-1".into(), op: Op::Interrupt, client_user_message_id: None, trace: Some(submission_trace.clone()), }); let dispatch_span_id = dispatch_span.context().span().span_context().span_id(); let (sess, tc, rx) = make_session_and_context_with_rx().await; let captured_trace = Arc::new(std::sync::Mutex::new(None)); async { sess.spawn_task( Arc::clone(&tc), vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], client_id: None, }], TraceCaptureTask { captured_trace: Arc::clone(&captured_trace), }, ) .await; } .instrument(dispatch_span) .await; let evt = tokio::time::timeout(StdDuration::from_secs(2), rx.recv()) .await .expect("timeout waiting for turn completion") .expect("event"); assert!(matches!(evt.msg, EventMsg::TurnComplete(_))); let task_trace = captured_trace .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .clone() .expect("turn task should capture the current span trace context"); let submission_context = codex_otel::context_from_w3c_trace_context(&submission_trace).expect("submission"); let task_context = codex_otel::context_from_w3c_trace_context(&task_trace).expect("task trace"); assert_eq!( task_context.span().span_context().trace_id(), submission_context.span().span_context().trace_id() ); assert_ne!( task_context.span().span_context().span_id(), dispatch_span_id );}#[cfg(debug_assertions)]#[tokio::test]async fn shutdown_complete_does_not_append_to_thread_store_after_shutdown() { let (mut session, _turn_context) = make_session_and_context().await; let store = Arc::new(codex_thread_store::InMemoryThreadStore::default()); let thread_store: Arc<dyn codex_thread_store::ThreadStore> = store.clone(); let config = session.get_config().await; let live_thread = LiveThread::create( Arc::clone(&thread_store), CreateThreadParams { thread_id: session.thread_id, extra_config: None, forked_from_id: None, parent_thread_id: None, source: SessionSource::Exec, thread_source: None, base_instructions: BaseInstructions::default(), dynamic_tools: Vec::new(), multi_agent_version: None, metadata: ThreadPersistenceMetadata { cwd: Some(config.cwd.to_path_buf()), model_provider: config.model_provider_id.clone(), memory_mode: if config.memories.generate_memories { ThreadMemoryMode::Enabled } else { ThreadMemoryMode::Disabled }, }, }, ) .await .expect("create thread persistence"); session.services.thread_store = thread_store; session.services.live_thread = Some(live_thread); let session = Arc::new(session); assert!(handlers::shutdown(&session, "sub-1".to_string()).await); assert_eq!( codex_thread_store::InMemoryThreadStoreCalls { create_thread: 1, shutdown_thread: 1, ..Default::default() }, store.calls().await );}#[tokio::test]async fn submission_loop_channel_close_emits_thread_stop_lifecycle() { struct SessionStopMarker; struct ThreadStopMarker; struct ThreadStopRecorder { calls: Arc<std::sync::atomic::AtomicUsize>, expected_thread_id: ThreadId, } impl codex_extension_api::ThreadLifecycleContributor<crate::config::Config> for ThreadStopRecorder { fn on_thread_stop<'a>( &'a self, input: codex_extension_api::ThreadStopInput<'a>, ) -> codex_extension_api::ExtensionFuture<'a, ()> { Box::pin(async move { assert_eq!( self.expected_thread_id.to_string(), input.thread_store.level_id() ); assert!(input.session_store.get::<SessionStopMarker>().is_some()); assert!(input.thread_store.get::<ThreadStopMarker>().is_some()); self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); }) } } let (mut session, turn_context) = make_session_and_context().await; let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let mut builder = codex_extension_api::ExtensionRegistryBuilder::<crate::config::Config>::new(); builder.thread_lifecycle_contributor(Arc::new(ThreadStopRecorder { calls: Arc::clone(&calls), expected_thread_id: session.thread_id, })); session.services.extensions = Arc::new(builder.build()); session .services .session_extension_data .insert(SessionStopMarker); session .services .thread_extension_data .insert(ThreadStopMarker); let (tx_sub, rx_sub) = async_channel::bounded(1); drop(tx_sub); let session = Arc::new(session); submission_loop(session, Arc::clone(&turn_context.config), rx_sub).await; assert_eq!(1, calls.load(std::sync::atomic::Ordering::SeqCst));}#[tokio::test]async fn submission_loop_channel_close_aborts_active_turn_before_thread_stop_lifecycle() { struct LifecycleRecorder { calls: Arc<std::sync::Mutex<Vec<&'static str>>>, expected_thread_id: ThreadId, expected_turn_id: String, } impl codex_extension_api::ThreadLifecycleContributor<crate::config::Config> for LifecycleRecorder { fn on_thread_stop<'a>( &'a self, input: codex_extension_api::ThreadStopInput<'a>, ) -> codex_extension_api::ExtensionFuture<'a, ()> { Box::pin(async move { assert_eq!( self.expected_thread_id.to_string(), input.thread_store.level_id() ); self.calls .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .push("thread_stop"); }) } } impl codex_extension_api::TurnLifecycleContributor for LifecycleRecorder { fn on_turn_abort<'a>( &'a self, input: codex_extension_api::TurnAbortInput<'a>, ) -> codex_extension_api::ExtensionFuture<'a, ()> { Box::pin(async move { assert_eq!( self.expected_thread_id.to_string(), input.thread_store.level_id() ); assert_eq!(self.expected_turn_id, input.turn_store.level_id()); assert_eq!(TurnAbortReason::Interrupted, input.reason); self.calls .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .push("turn_abort"); }) } } let (mut session, turn_context) = make_session_and_context().await; let calls = Arc::new(std::sync::Mutex::new(Vec::new())); let recorder = Arc::new(LifecycleRecorder { calls: Arc::clone(&calls), expected_thread_id: session.thread_id, expected_turn_id: turn_context.sub_id.clone(), }); let mut builder = codex_extension_api::ExtensionRegistryBuilder::<crate::config::Config>::new(); builder.thread_lifecycle_contributor(recorder.clone()); builder.turn_lifecycle_contributor(recorder); session.services.extensions = Arc::new(builder.build()); let session = Arc::new(session); session .spawn_task( Arc::new(turn_context), Vec::new(), NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; let (tx_sub, rx_sub) = async_channel::bounded(1); drop(tx_sub); submission_loop(Arc::clone(&session), session.get_config().await, rx_sub).await; assert_eq!( vec!["turn_abort", "thread_stop"], *calls .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) );}#[tokio::test]async fn shutdown_and_wait_allows_multiple_waiters() { let (session, _turn_context) = make_session_and_context().await; let (tx_sub, rx_sub) = async_channel::bounded(4); let (_tx_event, rx_event) = async_channel::unbounded(); let (_agent_status_tx, agent_status) = watch::channel(AgentStatus::PendingInit); let session_loop_handle = tokio::spawn(async move { let shutdown: Submission = rx_sub.recv().await.expect("shutdown submission"); assert_eq!(shutdown.op, Op::Shutdown); tokio::time::sleep(StdDuration::from_millis(50)).await; }); let codex = Arc::new(Codex { tx_sub, rx_event, agent_status, session: Arc::new(session), session_loop_termination: session_loop_termination_from_handle(session_loop_handle), }); let waiter_1 = { let codex = Arc::clone(&codex); tokio::spawn(async move { codex.shutdown_and_wait().await }) }; let waiter_2 = { let codex = Arc::clone(&codex); tokio::spawn(async move { codex.shutdown_and_wait().await }) }; waiter_1 .await .expect("first shutdown waiter join") .expect("first shutdown waiter"); waiter_2 .await .expect("second shutdown waiter join") .expect("second shutdown waiter");}#[tokio::test]async fn shutdown_and_wait_waits_when_shutdown_is_already_in_progress() { let (session, _turn_context) = make_session_and_context().await; let (tx_sub, rx_sub) = async_channel::bounded(4); drop(rx_sub); let (_tx_event, rx_event) = async_channel::unbounded(); let (_agent_status_tx, agent_status) = watch::channel(AgentStatus::PendingInit); let (shutdown_complete_tx, shutdown_complete_rx) = tokio::sync::oneshot::channel(); let session_loop_handle = tokio::spawn(async move { let _ = shutdown_complete_rx.await; }); let codex = Arc::new(Codex { tx_sub, rx_event, agent_status, session: Arc::new(session), session_loop_termination: session_loop_termination_from_handle(session_loop_handle), }); let waiter = { let codex = Arc::clone(&codex); tokio::spawn(async move { codex.shutdown_and_wait().await }) }; tokio::time::sleep(StdDuration::from_millis(10)).await; assert!(!waiter.is_finished()); shutdown_complete_tx .send(()) .expect("session loop should still be waiting to terminate"); waiter .await .expect("shutdown waiter join") .expect("shutdown waiter");}#[tokio::test]async fn shutdown_and_wait_shuts_down_cached_guardian_subagent() { let (parent_session, parent_turn_context) = make_session_and_context().await; let parent_session = Arc::new(parent_session); let parent_config = Arc::clone(&parent_turn_context.config); let (parent_tx_sub, parent_rx_sub) = async_channel::bounded(4); let (_parent_tx_event, parent_rx_event) = async_channel::unbounded(); let (_parent_status_tx, parent_agent_status) = watch::channel(AgentStatus::PendingInit); let parent_session_for_loop = Arc::clone(&parent_session); let parent_session_loop_handle = tokio::spawn(async move { submission_loop(parent_session_for_loop, parent_config, parent_rx_sub).await; }); let parent_codex = Codex { tx_sub: parent_tx_sub, rx_event: parent_rx_event, agent_status: parent_agent_status, session: Arc::clone(&parent_session), session_loop_termination: session_loop_termination_from_handle(parent_session_loop_handle), }; let (child_session, _child_turn_context) = make_session_and_context().await; let (child_tx_sub, child_rx_sub) = async_channel::bounded(4); let (_child_tx_event, child_rx_event) = async_channel::unbounded(); let (_child_status_tx, child_agent_status) = watch::channel(AgentStatus::PendingInit); let (child_shutdown_tx, child_shutdown_rx) = tokio::sync::oneshot::channel(); let child_session_loop_handle = tokio::spawn(async move { let shutdown: Submission = child_rx_sub .recv() .await .expect("child shutdown submission"); assert_eq!(shutdown.op, Op::Shutdown); child_shutdown_tx .send(()) .expect("child shutdown signal should be delivered"); }); let child_codex = Codex { tx_sub: child_tx_sub, rx_event: child_rx_event, agent_status: child_agent_status, session: Arc::new(child_session), session_loop_termination: session_loop_termination_from_handle(child_session_loop_handle), }; parent_session .guardian_review_session .cache_for_test(child_codex) .await; parent_codex .shutdown_and_wait() .await .expect("parent shutdown should succeed"); child_shutdown_rx .await .expect("guardian subagent should receive a shutdown op");}#[tokio::test]async fn cached_guardian_subagent_exposes_its_rollout_path() { let (parent_session, _parent_turn_context) = make_session_and_context().await; let parent_session = Arc::new(parent_session); let (mut child_session, _child_turn_context) = make_session_and_context().await; let child_rollout_path = attach_thread_persistence(&mut child_session).await; let (child_tx_sub, _child_rx_sub) = async_channel::bounded(4); let (_child_tx_event, child_rx_event) = async_channel::unbounded(); let (_child_status_tx, child_agent_status) = watch::channel(AgentStatus::PendingInit); let child_session_loop_handle = tokio::spawn(async {}); let child_codex = Codex { tx_sub: child_tx_sub, rx_event: child_rx_event, agent_status: child_agent_status, session: Arc::new(child_session), session_loop_termination: session_loop_termination_from_handle(child_session_loop_handle), }; parent_session .guardian_review_session .cache_for_test(child_codex) .await; assert_eq!( parent_session .guardian_review_session .trunk_rollout_path() .await, Some(child_rollout_path) );}#[tokio::test]async fn shutdown_and_wait_shuts_down_tracked_ephemeral_guardian_review() { let (parent_session, parent_turn_context) = make_session_and_context().await; let parent_session = Arc::new(parent_session); let parent_config = Arc::clone(&parent_turn_context.config); let (parent_tx_sub, parent_rx_sub) = async_channel::bounded(4); let (_parent_tx_event, parent_rx_event) = async_channel::unbounded(); let (_parent_status_tx, parent_agent_status) = watch::channel(AgentStatus::PendingInit); let parent_session_for_loop = Arc::clone(&parent_session); let parent_session_loop_handle = tokio::spawn(async move { submission_loop(parent_session_for_loop, parent_config, parent_rx_sub).await; }); let parent_codex = Codex { tx_sub: parent_tx_sub, rx_event: parent_rx_event, agent_status: parent_agent_status, session: Arc::clone(&parent_session), session_loop_termination: session_loop_termination_from_handle(parent_session_loop_handle), }; let (child_session, _child_turn_context) = make_session_and_context().await; let (child_tx_sub, child_rx_sub) = async_channel::bounded(4); let (_child_tx_event, child_rx_event) = async_channel::unbounded(); let (_child_status_tx, child_agent_status) = watch::channel(AgentStatus::PendingInit); let (child_shutdown_tx, child_shutdown_rx) = tokio::sync::oneshot::channel(); let child_session_loop_handle = tokio::spawn(async move { let shutdown: Submission = child_rx_sub .recv() .await .expect("child shutdown submission"); assert_eq!(shutdown.op, Op::Shutdown); child_shutdown_tx .send(()) .expect("child shutdown signal should be delivered"); }); let child_codex = Codex { tx_sub: child_tx_sub, rx_event: child_rx_event, agent_status: child_agent_status, session: Arc::new(child_session), session_loop_termination: session_loop_termination_from_handle(child_session_loop_handle), }; parent_session .guardian_review_session .register_ephemeral_for_test(child_codex) .await; parent_codex .shutdown_and_wait() .await .expect("parent shutdown should succeed"); child_shutdown_rx .await .expect("ephemeral guardian review should receive a shutdown op");}async fn make_session_and_context_with_auth_and_config_and_rx<F>( auth: CodexAuth, dynamic_tools: Vec<DynamicToolSpec>, configure_config: F,) -> ( Arc<Session>, Arc<TurnContext>, async_channel::Receiver<Event>,)where F: FnOnce(&mut Config),{ let codex_home = tempfile::tempdir().expect("create temp dir"); make_session_and_context_with_auth_config_home_and_rx( auth, dynamic_tools, codex_home.path(), configure_config, ) .await}async fn make_session_and_context_with_auth_config_home_and_rx<F>( auth: CodexAuth, dynamic_tools: Vec<DynamicToolSpec>, codex_home: &Path, configure_config: F,) -> ( Arc<Session>, Arc<TurnContext>, async_channel::Receiver<Event>,)where F: FnOnce(&mut Config),{ let (tx_event, rx_event) = async_channel::unbounded(); let mut config = build_test_config(codex_home).await; configure_config(&mut config); let state_db = None; let config = Arc::new(config); let thread_id = ThreadId::default(); let auth_manager = AuthManager::from_auth_for_testing(auth); let models_manager = models_manager_with_provider( config.codex_home.to_path_buf(), auth_manager.clone(), config.model_provider.clone(), ); let agent_control = AgentControl::default(); let exec_policy = Arc::new(ExecPolicyManager::default()); let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let model = get_model_offline_for_tests(config.model.as_deref()); let model_info = construct_model_info_offline_for_tests(model.as_str(), &config.to_models_manager_config()); let reasoning_effort = config.model_reasoning_effort.clone(); let collaboration_mode = CollaborationMode { mode: ModeKind::Default, settings: Settings { model, reasoning_effort, developer_instructions: None, }, }; let default_environments = vec![local(config.cwd.clone())]; let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), loaded_agents_md: None, service_tier: None, personality: config.personality, base_instructions: config .base_instructions .clone() .unwrap_or_else(|| model_info.get_model_instructions(config.personality)), compact_prompt: config.compact_prompt.clone(), approval_policy: config.permissions.approval_policy.clone(), approvals_reviewer: config.approvals_reviewer, permission_profile_state: config.permissions.permission_profile_state().clone(), windows_sandbox_level: WindowsSandboxLevel::from_config(&config), environments: TurnEnvironmentSelections::new(config.cwd.clone(), default_environments), workspace_roots: config.workspace_roots.clone(), codex_home: config.codex_home.clone(), thread_name: None, original_config_do_not_use: Arc::clone(&config), metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, forked_from_thread_id: None, parent_thread_id: None, thread_source: None, dynamic_tools, user_shell_override: None, }; let per_turn_config = Session::build_per_turn_config(&session_configuration, session_configuration.cwd().clone()); let model_info = construct_model_info_offline_for_tests( session_configuration.collaboration_mode.model(), &per_turn_config.to_models_manager_config(), ); let session_telemetry = session_telemetry( thread_id, config.as_ref(), &model_info, session_configuration.session_source.clone(), ); let state = SessionState::new(session_configuration.clone()); let (environment_manager, resolved_turn_environments) = resolved_environments_for_configuration(&session_configuration).await; let turn_environments = Arc::new(ThreadEnvironments::new( environment_manager, default_user_shell(), ShellSnapshot::disabled(), resolved_turn_environments.clone(), )); let environment = Arc::clone( &resolved_turn_environments .primary() .expect("primary environment") .environment, ); let plugins_manager = Arc::new(PluginsManager::new(config.codex_home.to_path_buf())); let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager))); let skills_manager = Arc::new(SkillsManager::new( config.codex_home.clone(), /*bundled_skills_enabled*/ true, )); let network_approval = Arc::new(NetworkApprovalService::default()); let services = SessionServices { mcp_connection_manager: Arc::new(arc_swap::ArcSwap::from_pointee( McpConnectionManager::new_uninitialized_with_permission_profile( &config.permissions.approval_policy, config.permissions.permission_profile(), config.prefix_mcp_tool_names(), ), )), mcp_startup_cancellation_token: Mutex::new(CancellationToken::new()), unified_exec_manager: UnifiedExecProcessManager::new( config.background_terminal_max_timeout, ), shell_zsh_path: None, main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(), analytics_events_client: AnalyticsEventsClient::new( Arc::clone(&auth_manager), config.chatgpt_base_url.trim_end_matches('/').to_string(), config.analytics_enabled, ), hooks: arc_swap::ArcSwap::from_pointee(Hooks::new(HooksConfig { legacy_notify_argv: config.notify.clone(), ..HooksConfig::default() })), rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), user_shell: Arc::new(default_user_shell()), show_raw_agent_reasoning: config.show_raw_agent_reasoning, exec_policy, auth_manager: Arc::clone(&auth_manager), session_telemetry: session_telemetry.clone(), models_manager: Arc::clone(&models_manager), tool_approvals: Mutex::new(ApprovalStore::default()), guardian_rejections: Mutex::new(std::collections::HashMap::new()), guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), skills_manager, plugins_manager, mcp_manager, extensions: Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), session_extension_data: codex_extension_api::ExtensionData::new( agent_control.session_id().to_string(), ), thread_extension_data: codex_extension_api::ExtensionData::new(thread_id.to_string()), mcp_thread_init: codex_extension_api::ExtensionDataInit::default(), agent_control, network_proxy: arc_swap::ArcSwapOption::from(None), network_proxy_audit_metadata: crate::config::NetworkProxyAuditMetadata::default(), managed_network_requirements_configured: false, network_approval: Arc::clone(&network_approval), state_db: state_db.clone(), live_thread: None, thread_store: Arc::new(codex_thread_store::LocalThreadStore::new( codex_thread_store::LocalThreadStoreConfig::from_config(config.as_ref()), state_db, )), attestation_provider: None, model_client: ModelClient::new( Some(Arc::clone(&auth_manager)), thread_id, session_configuration.provider.clone(), session_configuration.session_source.clone(), config.model_verbosity, config.features.enabled(Feature::EnableRequestCompression), config.features.enabled(Feature::RuntimeMetrics), Session::build_model_client_beta_features_header(config.as_ref()), /*attestation_provider*/ None, ), code_mode_service: crate::tools::code_mode::CodeModeService::new(), tool_search_handler_cache: Default::default(), turn_environments: Arc::clone(&turn_environments), }; let plugin_outcome = services .plugins_manager .plugins_for_config(&per_turn_config.plugins_config_input()) .await; let effective_skill_roots = plugin_outcome.effective_plugin_skill_roots(); let skills_input = crate::skills_load_input_from_config(&per_turn_config, effective_skill_roots); let skill_fs = environment.get_filesystem(); let skills_outcome = Arc::new( services .skills_manager .skills_for_config(&skills_input, Some(Arc::clone(&skill_fs))) .await, ); let turn_context = Arc::new(Session::make_turn_context( thread_id, SessionId::from(thread_id), Some(Arc::clone(&auth_manager)), &session_telemetry, session_configuration.provider.clone(), &session_configuration, config.multi_agent_version_from_features(), services.user_shell.as_ref(), services.shell_zsh_path.as_ref(), services.main_execve_wrapper_exe.as_ref(), per_turn_config, model_info, &models_manager, /*network*/ None, resolved_turn_environments, session_configuration.cwd().clone(), "turn_id".to_string(), skills_outcome, )); let session = Arc::new(Session { thread_id, installation_id: "11111111-1111-4111-8111-111111111111".to_string(), tx_event, agent_status: agent_status_tx, out_of_band_elicitation_paused: watch::channel(false).0, state: Mutex::new(state), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), multi_agent_version: OnceLock::from(config.multi_agent_version_from_features()), pending_mcp_server_refresh_config: Mutex::new(None), conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), input_queue: super::input_queue::InputQueue::new(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), services, next_internal_sub_id: AtomicU64::new(0), }); (session, turn_context, rx_event)}pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx( dynamic_tools: Vec<DynamicToolSpec>,) -> ( Arc<Session>, Arc<TurnContext>, async_channel::Receiver<Event>,) { make_session_and_context_with_auth_and_config_and_rx( CodexAuth::from_api_key("Test API Key"), dynamic_tools, |_config| {}, ) .await}// Like make_session_and_context, but returns Arc<Session> and the event receiver// so tests can assert on emitted events.pub(crate) async fn make_session_and_context_with_rx() -> ( Arc<Session>, Arc<TurnContext>, async_channel::Receiver<Event>,) { make_session_and_context_with_dynamic_tools_and_rx(Vec::new()).await}#[tokio::test]async fn refresh_mcp_servers_is_deferred_until_next_turn() { let (session, turn_context) = make_session_and_context().await; let old_token = session.mcp_startup_cancellation_token().await; assert!(!old_token.is_cancelled()); let mcp_oauth_credentials_store_mode = serde_json::to_value(OAuthCredentialsStoreMode::Auto).expect("serialize store mode"); let auth_keyring_backend_kind = serde_json::to_value(AuthKeyringBackendKind::Secrets).expect("serialize keyring backend"); let refresh_config = McpServerRefreshConfig { mcp_servers: json!({}), mcp_oauth_credentials_store_mode, auth_keyring_backend_kind, }; { let mut guard = session.pending_mcp_server_refresh_config.lock().await; *guard = Some(refresh_config); } assert!(!old_token.is_cancelled()); assert!( session .pending_mcp_server_refresh_config .lock() .await .is_some() ); session .refresh_mcp_servers_if_requested(&turn_context, /*elicitation_reviewer*/ None) .await; assert!(old_token.is_cancelled()); assert!( session .pending_mcp_server_refresh_config .lock() .await .is_none() ); let new_token = session.mcp_startup_cancellation_token().await; assert!(!new_token.is_cancelled());}#[tokio::test]async fn spawn_task_does_not_update_previous_turn_settings_for_non_run_turn_tasks() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; sess.set_previous_turn_settings(/*previous_turn_settings*/ None) .await; let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], client_id: None, }]; sess.spawn_task( Arc::clone(&tc), input, NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; assert_eq!(sess.previous_turn_settings().await, None);}#[tokio::test]async fn build_settings_update_items_emits_environment_item_for_network_changes() { let (session, previous_context) = make_session_and_context().await; let previous_context = Arc::new(previous_context); let mut current_context = previous_context .with_model( previous_context.model_info.slug.clone(), &session.services.models_manager, ) .await; let mut config = (*current_context.config).clone(); let mut requirements = config.config_layer_stack.requirements().clone(); requirements.network = Some(Sourced::new( NetworkConstraints { domains: Some(NetworkDomainPermissionsToml { entries: std::collections::BTreeMap::from([ ( "api.example.com".to_string(), NetworkDomainPermissionToml::Allow, ), ( "blocked.example.com".to_string(), NetworkDomainPermissionToml::Deny, ), ]), }), ..Default::default() }, RequirementSource::LegacyManagedConfigTomlFromMdm, )); let layers = config .config_layer_stack .get_layers( ConfigLayerStackOrdering::LowestPrecedenceFirst, /*include_disabled*/ true, ) .into_iter() .cloned() .collect(); config.config_layer_stack = ConfigLayerStack::new( layers, requirements, config.config_layer_stack.requirements_toml().clone(), ) .expect("rebuild config layer stack with network requirements"); current_context.config = Arc::new(config); let reference_context_item = previous_context.to_turn_context_item(); let update_items = session .build_settings_update_items(Some(&reference_context_item), ¤t_context) .await; let environment_update = user_input_texts(&update_items) .into_iter() .find(|text| text.contains("<environment_context>")) .expect("environment update item should be emitted"); assert!(environment_update.contains( "<network enabled=\"true\"><allowed>api.example.com</allowed><denied>blocked.example.com</denied></network>" ));}#[tokio::test]async fn environment_context_uses_session_shell_when_environment_shell_is_absent() { let (mut session, mut turn_context) = make_session_and_context().await; session.services.user_shell = Arc::new(crate::shell::Shell { shell_type: crate::shell::ShellType::PowerShell, shell_path: PathBuf::from("powershell"), }); for environment in &mut turn_context.environments.turn_environments { environment.shell = None; } let session_shell = session.user_shell(); let environment_context = crate::context::EnvironmentContext::from_turn_context( &turn_context, session_shell.as_ref(), ) .render(); assert!( environment_context.contains("<shell>powershell</shell>"), "{environment_context}" ); let primary_environment = turn_context .environments .turn_environments .first_mut() .expect("primary environment"); primary_environment.shell = Some(crate::shell::Shell { shell_type: crate::shell::ShellType::Cmd, shell_path: PathBuf::from("cmd"), }); let environment_context = crate::context::EnvironmentContext::from_turn_context( &turn_context, session_shell.as_ref(), ) .render(); assert!( environment_context.contains("<shell>cmd</shell>"), "{environment_context}" );}#[tokio::test]async fn build_settings_update_items_emits_environment_item_for_time_changes() { let (session, previous_context) = make_session_and_context().await; let previous_context = Arc::new(previous_context); let mut current_context = previous_context .with_model( previous_context.model_info.slug.clone(), &session.services.models_manager, ) .await; current_context.current_date = Some("2026-02-27".to_string()); current_context.timezone = Some("Europe/Berlin".to_string()); let reference_context_item = previous_context.to_turn_context_item(); let update_items = session .build_settings_update_items(Some(&reference_context_item), ¤t_context) .await; let environment_update = user_input_texts(&update_items) .into_iter() .find(|text| text.contains("<environment_context>")) .expect("environment update item should be emitted"); assert!(environment_update.contains("<current_date>2026-02-27</current_date>")); assert!(environment_update.contains("<timezone>Europe/Berlin</timezone>"));}#[tokio::test]async fn build_settings_update_items_omits_environment_item_when_disabled() { let (session, previous_context) = make_session_and_context().await; let previous_context = Arc::new(previous_context); let mut current_context = previous_context .with_model( previous_context.model_info.slug.clone(), &session.services.models_manager, ) .await; let mut config = (*current_context.config).clone(); config.include_environment_context = false; current_context.config = Arc::new(config); current_context.current_date = Some("2026-02-27".to_string()); let reference_context_item = previous_context.to_turn_context_item(); let update_items = session .build_settings_update_items(Some(&reference_context_item), ¤t_context) .await; let user_texts = user_input_texts(&update_items); assert!( !user_texts .iter() .any(|text| text.contains("<environment_context>")), "did not expect environment context updates when disabled, got {user_texts:?}" );}#[tokio::test]async fn build_settings_update_items_emits_realtime_start_when_session_becomes_live() { let (session, previous_context) = make_session_and_context().await; let previous_context = Arc::new(previous_context); let mut current_context = previous_context .with_model( previous_context.model_info.slug.clone(), &session.services.models_manager, ) .await; current_context.realtime_active = true; let update_items = session .build_settings_update_items( Some(&previous_context.to_turn_context_item()), ¤t_context, ) .await; let developer_texts = developer_input_texts(&update_items); assert!( developer_texts .iter() .any(|text| text.contains("<realtime_conversation>")), "expected a realtime start update, got {developer_texts:?}" );}#[tokio::test]async fn build_settings_update_items_emits_realtime_end_when_session_stops_being_live() { let (session, mut previous_context) = make_session_and_context().await; previous_context.realtime_active = true; let mut current_context = previous_context .with_model( previous_context.model_info.slug.clone(), &session.services.models_manager, ) .await; current_context.realtime_active = false; let update_items = session .build_settings_update_items( Some(&previous_context.to_turn_context_item()), ¤t_context, ) .await; let developer_texts = developer_input_texts(&update_items); assert!( developer_texts .iter() .any(|text| text.contains("Reason: inactive")), "expected a realtime end update, got {developer_texts:?}" );}#[tokio::test]async fn build_settings_update_items_uses_previous_turn_settings_for_realtime_end() { let (session, previous_context) = make_session_and_context().await; let mut previous_context_item = previous_context.to_turn_context_item(); previous_context_item.realtime_active = None; let previous_turn_settings = PreviousTurnSettings { model: previous_context.model_info.slug.clone(), comp_hash: None, realtime_active: Some(true), }; let mut current_context = previous_context .with_model( previous_context.model_info.slug.clone(), &session.services.models_manager, ) .await; current_context.realtime_active = false; session .set_previous_turn_settings(Some(previous_turn_settings)) .await; let update_items = session .build_settings_update_items(Some(&previous_context_item), ¤t_context) .await; let developer_texts = developer_input_texts(&update_items); assert!( developer_texts .iter() .any(|text| text.contains("Reason: inactive")), "expected a realtime end update from previous turn settings, got {developer_texts:?}" );}#[tokio::test]async fn build_initial_context_uses_previous_realtime_state() { let (session, mut turn_context) = make_session_and_context().await; turn_context.realtime_active = true; let initial_context = session.build_initial_context(&turn_context).await; let developer_texts = developer_input_texts(&initial_context); assert!( developer_texts .iter() .any(|text| text.contains("<realtime_conversation>")), "expected initial context to describe active realtime state, got {developer_texts:?}" ); let previous_context_item = turn_context.to_turn_context_item(); { let mut state = session.state.lock().await; state.set_reference_context_item(Some(previous_context_item)); } let resumed_context = session.build_initial_context(&turn_context).await; let resumed_developer_texts = developer_input_texts(&resumed_context); assert!( !resumed_developer_texts .iter() .any(|text| text.contains("<realtime_conversation>")), "did not expect a duplicate realtime update, got {resumed_developer_texts:?}" );}async fn make_multi_agent_v2_usage_hint_test_session( enable_multi_agent_v2: bool,) -> (Arc<Session>, Arc<TurnContext>) { let (session, turn_context, _rx_event) = make_session_and_context_with_auth_and_config_and_rx( CodexAuth::from_api_key("Test API Key"), Vec::new(), |config| { if enable_multi_agent_v2 { let _ = config.features.enable(Feature::MultiAgentV2); } config.multi_agent_v2.root_agent_usage_hint_text = Some("Root guidance.".to_string()); config.multi_agent_v2.subagent_usage_hint_text = Some("Subagent guidance.".to_string()); }, ) .await; (session, turn_context)}struct PromptExtensionTestContributor;struct PromptExtensionTestState;impl codex_extension_api::ContextContributor for PromptExtensionTestContributor { fn contribute<'a>( &'a self, _session_store: &'a codex_extension_api::ExtensionData, thread_store: &'a codex_extension_api::ExtensionData, ) -> std::pin::Pin< Box<dyn std::future::Future<Output = Vec<codex_extension_api::PromptFragment>> + Send + 'a>, > { Box::pin(async move { thread_store .get::<PromptExtensionTestState>() .is_some() .then(|| { codex_extension_api::PromptFragment::developer_policy( "prompt extension enabled", ) }) .into_iter() .collect() }) }}fn prompt_extension_test_registry()-> Arc<codex_extension_api::ExtensionRegistry<crate::config::Config>> { let mut builder = codex_extension_api::ExtensionRegistryBuilder::new(); builder.prompt_contributor(Arc::new(PromptExtensionTestContributor)); Arc::new(builder.build())}#[tokio::test]async fn build_initial_context_includes_prompt_fragments_from_extensions() { let (mut session, turn_context) = make_session_and_context().await; session.services.extensions = prompt_extension_test_registry(); session .services .thread_extension_data .insert(PromptExtensionTestState); let initial_context = session.build_initial_context(&turn_context).await; let developer_messages = developer_message_texts(&initial_context); assert!( developer_messages .iter() .flatten() .any(|text| *text == "prompt extension enabled"), "expected prompt extension developer text, got {developer_messages:?}" );}#[tokio::test]async fn build_initial_context_omits_prompt_fragments_without_extension_state() { let (mut session, turn_context) = make_session_and_context().await; session.services.extensions = prompt_extension_test_registry(); let initial_context = session.build_initial_context(&turn_context).await; let developer_messages = developer_message_texts(&initial_context); assert!( !developer_messages .iter() .flatten() .any(|text| *text == "prompt extension enabled"), "did not expect prompt extension developer text, got {developer_messages:?}" );}#[tokio::test]async fn build_initial_context_adds_multi_agent_v2_root_usage_hint_as_developer_message() { let (session, turn_context) = make_multi_agent_v2_usage_hint_test_session(/*enable_multi_agent_v2*/ true).await; let initial_context = session.build_initial_context(turn_context.as_ref()).await; let developer_messages = developer_message_texts(&initial_context); assert!( developer_messages .iter() .any(|message| message.as_slice() == ["Root guidance."]), "expected standalone root usage hint developer message, got {developer_messages:?}" ); assert!( !developer_messages .iter() .any(|message| message.as_slice() == ["Subagent guidance."]), "did not expect subagent usage hint for root thread, got {developer_messages:?}" );}#[tokio::test]async fn build_initial_context_adds_multi_agent_v2_subagent_usage_hint_as_developer_message() { let (session, mut turn_context) = make_multi_agent_v2_usage_hint_test_session(/*enable_multi_agent_v2*/ true).await; let session_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: ThreadId::new(), depth: 1, agent_path: Some(AgentPath::try_from("/root/worker").expect("agent path should parse")), agent_nickname: Some("worker".to_string()), agent_role: None, }); session .state .lock() .await .session_configuration .session_source = session_source.clone(); Arc::get_mut(&mut turn_context) .expect("thread settings should not be shared") .session_source = session_source; let initial_context = session.build_initial_context(turn_context.as_ref()).await; let developer_messages = developer_message_texts(&initial_context); assert!( developer_messages .iter() .any(|message| message.as_slice() == ["Subagent guidance."]), "expected standalone subagent usage hint developer message, got {developer_messages:?}" ); assert!( !developer_messages .iter() .any(|message| message.as_slice() == ["Root guidance."]), "did not expect root usage hint for subagent thread, got {developer_messages:?}" );}#[tokio::test]async fn build_initial_context_omits_multi_agent_v2_usage_hints_when_feature_disabled() { let (session, turn_context) = make_multi_agent_v2_usage_hint_test_session(/*enable_multi_agent_v2*/ false).await; let initial_context = session.build_initial_context(turn_context.as_ref()).await; let developer_messages = developer_message_texts(&initial_context); assert!( !developer_messages.iter().any(|message| { matches!( message.as_slice(), ["Root guidance."] | ["Subagent guidance."] ) }), "did not expect multi-agent v2 usage hint developer messages, got {developer_messages:?}" );}#[tokio::test]async fn build_initial_context_omits_multi_agent_v2_usage_hints_when_hint_disabled() { let (session, turn_context, _rx_event) = make_session_and_context_with_auth_and_config_and_rx( CodexAuth::from_api_key("Test API Key"), Vec::new(), |config| { let _ = config.features.enable(Feature::MultiAgentV2); config.multi_agent_v2.usage_hint_enabled = false; config.multi_agent_v2.root_agent_usage_hint_text = Some("Root guidance.".to_string()); config.multi_agent_v2.subagent_usage_hint_text = Some("Subagent guidance.".to_string()); }, ) .await; let initial_context = session.build_initial_context(turn_context.as_ref()).await; let developer_messages = developer_message_texts(&initial_context); assert!( !developer_messages.iter().any(|message| { matches!( message.as_slice(), ["Root guidance."] | ["Subagent guidance."] ) }), "did not expect multi-agent v2 usage hint developer messages, got {developer_messages:?}" );}#[tokio::test]async fn build_initial_context_omits_default_image_save_location_with_image_history() { let (session, turn_context) = make_session_and_context().await; session .replace_history( vec![ResponseItem::ImageGenerationCall { id: "ig-test".to_string(), status: "completed".to_string(), revised_prompt: Some("a tiny blue square".to_string()), result: "Zm9v".to_string(), metadata: None, }], /*reference_context_item*/ None, ) .await; let initial_context = session.build_initial_context(&turn_context).await; let developer_texts = developer_input_texts(&initial_context); assert!( !developer_texts .iter() .any(|text| text.contains("Generated images are saved to")), "expected initial context to omit image save instructions even with image history, got {developer_texts:?}" );}#[tokio::test]async fn build_initial_context_omits_default_image_save_location_without_image_history() { let (session, turn_context) = make_session_and_context().await; let initial_context = session.build_initial_context(&turn_context).await; let developer_texts = developer_input_texts(&initial_context); assert!( !developer_texts .iter() .any(|text| text.contains("Generated images are saved to")), "expected initial context to omit image save instructions without image history, got {developer_texts:?}" );}#[tokio::test]async fn build_initial_context_trims_skill_metadata_from_context_window_budget() { let (session, mut turn_context) = make_session_and_context().await; let mut outcome = SkillLoadOutcome::default(); outcome.skills = vec![ SkillMetadata { name: "admin-skill".to_string(), description: "desc".to_string(), short_description: None, interface: None, dependencies: None, policy: None, path_to_skills_md: test_path_buf("/tmp/admin-skill/SKILL.md").abs(), scope: SkillScope::Admin, plugin_id: None, }, SkillMetadata { name: "repo-skill".to_string(), description: "desc".to_string(), short_description: None, interface: None, dependencies: None, policy: None, path_to_skills_md: test_path_buf("/tmp/repo-skill/SKILL.md").abs(), scope: SkillScope::Repo, plugin_id: None, }, ]; turn_context.model_info.context_window = Some(100); turn_context.turn_skills = TurnSkillsContext::new(Arc::new(outcome)); let initial_context = session.build_initial_context(&turn_context).await; let developer_texts = developer_input_texts(&initial_context); assert!( developer_texts .iter() .all(|text| !text.contains("Exceeded skills context budget")), "expected skill budget warning to stay out of the initial context, got {developer_texts:?}" ); assert!( developer_texts .iter() .all(|text| !text.contains("- admin-skill:") && !text.contains("- repo-skill:")), "expected no skill metadata entries to fit the tiny budget, got {developer_texts:?}" );}#[test]fn emit_thread_start_skill_metrics_records_enabled_kept_and_truncated_values() { let session_telemetry = test_session_telemetry_without_metadata(); let mut outcome = SkillLoadOutcome::default(); outcome.skills = vec![SkillMetadata { name: "repo-skill".to_string(), description: "desc".to_string(), short_description: None, interface: None, dependencies: None, policy: None, path_to_skills_md: test_path_buf("/tmp/repo-skill/SKILL.md").abs(), scope: SkillScope::Repo, plugin_id: None, }]; let rendered = build_available_skills( &outcome, SkillMetadataBudget::Characters(1), SkillRenderSideEffects::ThreadStart { session_telemetry: &session_telemetry, }, ) .expect("skills should render"); assert_eq!( rendered.warning_message, Some( "Exceeded skills context budget. All skill descriptions were removed and 1 additional skill was not included in the model-visible skills list." .to_string() ) ); let snapshot = session_telemetry .snapshot_metrics() .expect("runtime metrics snapshot"); assert_eq!( histogram_sum(&snapshot, THREAD_SKILLS_ENABLED_TOTAL_METRIC), 1 ); assert_eq!(histogram_sum(&snapshot, THREAD_SKILLS_KEPT_TOTAL_METRIC), 0); assert_eq!(histogram_sum(&snapshot, THREAD_SKILLS_TRUNCATED_METRIC), 1); assert_eq!( histogram_sum(&snapshot, THREAD_SKILLS_DESCRIPTION_TRUNCATED_CHARS_METRIC), 4 );}#[test]fn emit_thread_start_skill_metrics_records_description_truncated_chars_without_omitted_skills() { let session_telemetry = test_session_telemetry_without_metadata(); let alpha = SkillMetadata { name: "alpha-skill".to_string(), description: "abcdef".to_string(), short_description: None, interface: None, dependencies: None, policy: None, path_to_skills_md: test_path_buf("/tmp/alpha-skill/SKILL.md").abs(), scope: SkillScope::Repo, plugin_id: None, }; let beta = SkillMetadata { name: "beta-skill".to_string(), description: "uvwxyz".to_string(), short_description: None, interface: None, dependencies: None, policy: None, path_to_skills_md: test_path_buf("/tmp/beta-skill/SKILL.md").abs(), scope: SkillScope::Repo, plugin_id: None, }; let minimum_skill_line_cost = |skill: &SkillMetadata| { let path = skill.path_to_skills_md.to_string_lossy().replace('\\', "/"); format!("- {}: (file: {})\n", skill.name, path) .chars() .count() }; let minimum_budget = minimum_skill_line_cost(&alpha) + minimum_skill_line_cost(&beta); let mut outcome = SkillLoadOutcome::default(); outcome.skills = vec![alpha, beta]; let rendered = build_available_skills( &outcome, SkillMetadataBudget::Characters(minimum_budget + 6), SkillRenderSideEffects::ThreadStart { session_telemetry: &session_telemetry, }, ) .expect("skills should render"); assert_eq!(rendered.report.omitted_count, 0); assert_eq!(rendered.report.truncated_description_chars, 8); let snapshot = session_telemetry .snapshot_metrics() .expect("runtime metrics snapshot"); assert_eq!(histogram_sum(&snapshot, THREAD_SKILLS_TRUNCATED_METRIC), 0); assert_eq!( histogram_sum(&snapshot, THREAD_SKILLS_DESCRIPTION_TRUNCATED_CHARS_METRIC), 8 );}#[tokio::test]async fn build_initial_context_emits_thread_start_skill_warning_on_repeated_builds() { let (session, turn_context, rx) = make_session_and_context_with_rx().await; let mut turn_context = Arc::into_inner(turn_context).expect("sole thread settings owner"); let mut outcome = SkillLoadOutcome::default(); outcome.skills = vec![ SkillMetadata { name: "admin-skill".to_string(), description: "desc".to_string(), short_description: None, interface: None, dependencies: None, policy: None, path_to_skills_md: test_path_buf("/tmp/admin-skill/SKILL.md").abs(), scope: SkillScope::Admin, plugin_id: None, }, SkillMetadata { name: "repo-skill".to_string(), description: "desc".to_string(), short_description: None, interface: None, dependencies: None, policy: None, path_to_skills_md: test_path_buf("/tmp/repo-skill/SKILL.md").abs(), scope: SkillScope::Repo, plugin_id: None, }, ]; turn_context.model_info.context_window = Some(100); turn_context.turn_skills = TurnSkillsContext::new(Arc::new(outcome)); let _ = session.build_initial_context(&turn_context).await; let warning_event = timeout(Duration::from_secs(1), rx.recv()) .await .expect("warning event should arrive") .expect("warning event should be readable"); assert!(matches!( warning_event.msg, EventMsg::Warning(WarningEvent { message }) if message == "Exceeded skills context budget of 2%. All skill descriptions were removed and 2 additional skills were not included in the model-visible skills list." )); let _ = session.build_initial_context(&turn_context).await; let warning_event = timeout(Duration::from_secs(1), rx.recv()) .await .expect("warning event should arrive on repeated build") .expect("warning event should be readable"); assert!(matches!( warning_event.msg, EventMsg::Warning(WarningEvent { message }) if message == "Exceeded skills context budget of 2%. All skill descriptions were removed and 2 additional skills were not included in the model-visible skills list." ));}#[tokio::test]async fn handle_output_item_done_records_image_save_history_message() { let (session, turn_context) = make_session_and_context().await; let session = Arc::new(session); let turn_context = Arc::new(turn_context); let call_id = "ig_history_records_message"; let expected_saved_path = crate::stream_events_utils::image_generation_artifact_path( &turn_context.config.codex_home, &session.thread_id.to_string(), call_id, ); let _ = std::fs::remove_file(&expected_saved_path); let item = ResponseItem::ImageGenerationCall { id: call_id.to_string(), status: "completed".to_string(), revised_prompt: Some("a tiny blue square".to_string()), result: "Zm9v".to_string(), metadata: None, }; let mut ctx = HandleOutputCtx { sess: Arc::clone(&session), turn_context: Arc::clone(&turn_context), turn_store: Arc::new(codex_extension_api::ExtensionData::new( turn_context.sub_id.clone(), )), tool_runtime: test_tool_runtime(Arc::clone(&session), Arc::clone(&turn_context)), cancellation_token: CancellationToken::new(), }; handle_output_item_done(&mut ctx, item.clone(), /*previously_active_item*/ None) .await .expect("image generation item should succeed"); let history = session.clone_history().await; let image_output_path = crate::stream_events_utils::image_generation_artifact_path( &turn_context.config.codex_home, &session.thread_id.to_string(), "<image_id>", ); let image_output_dir = image_output_path .parent() .expect("generated image path should have a parent"); let image_message: ResponseItem = crate::context::ContextualUserFragment::into( crate::context::ImageGenerationInstructions::new( image_output_dir.display(), image_output_path.display(), ), ); let expected = vec![image_message, item]; assert_eq!(history.raw_items(), expected.as_slice()); assert_eq!( std::fs::read(&expected_saved_path).expect("saved file"), b"foo" ); let _ = std::fs::remove_file(&expected_saved_path);}#[tokio::test]async fn handle_output_item_done_skips_image_save_message_when_save_fails() { let (session, turn_context) = make_session_and_context().await; let session = Arc::new(session); let turn_context = Arc::new(turn_context); let call_id = "ig_history_no_message"; let expected_saved_path = crate::stream_events_utils::image_generation_artifact_path( &turn_context.config.codex_home, &session.thread_id.to_string(), call_id, ); let _ = std::fs::remove_file(&expected_saved_path); let item = ResponseItem::ImageGenerationCall { id: call_id.to_string(), status: "completed".to_string(), revised_prompt: Some("broken payload".to_string()), result: "_-8".to_string(), metadata: None, }; let mut ctx = HandleOutputCtx { sess: Arc::clone(&session), turn_context: Arc::clone(&turn_context), turn_store: Arc::new(codex_extension_api::ExtensionData::new( turn_context.sub_id.clone(), )), tool_runtime: test_tool_runtime(Arc::clone(&session), Arc::clone(&turn_context)), cancellation_token: CancellationToken::new(), }; handle_output_item_done(&mut ctx, item.clone(), /*previously_active_item*/ None) .await .expect("image generation item should still complete"); let history = session.clone_history().await; let expected = vec![item]; assert_eq!(history.raw_items(), expected.as_slice()); assert!(!expected_saved_path.exists());}#[tokio::test]async fn build_initial_context_uses_previous_turn_settings_for_realtime_end() { let (session, turn_context) = make_session_and_context().await; let previous_turn_settings = PreviousTurnSettings { model: turn_context.model_info.slug.clone(), comp_hash: None, realtime_active: Some(true), }; session .set_previous_turn_settings(Some(previous_turn_settings)) .await; let initial_context = session.build_initial_context(&turn_context).await; let developer_texts = developer_input_texts(&initial_context); assert!( developer_texts .iter() .any(|text| text.contains("Reason: inactive")), "expected initial context to describe an ended realtime session, got {developer_texts:?}" );}#[tokio::test]async fn build_initial_context_restates_realtime_start_when_reference_context_is_missing() { let (session, mut turn_context) = make_session_and_context().await; turn_context.realtime_active = true; let previous_turn_settings = PreviousTurnSettings { model: turn_context.model_info.slug.clone(), comp_hash: None, realtime_active: Some(true), }; session .set_previous_turn_settings(Some(previous_turn_settings)) .await; let initial_context = session.build_initial_context(&turn_context).await; let developer_texts = developer_input_texts(&initial_context); assert!( developer_texts .iter() .any(|text| text.contains("<realtime_conversation>")), "expected initial context to restate active realtime when the reference context is missing, got {developer_texts:?}" );}fn file_system_policy_with_unreadable_glob(turn_context: &TurnContext) -> FileSystemSandboxPolicy { #[allow(deprecated)] let mut policy = FileSystemSandboxPolicy::from_legacy_sandbox_policy_for_cwd( &turn_context.sandbox_policy(), &turn_context.cwd, ); #[allow(deprecated)] let cwd_display = turn_context.cwd.as_path().display().to_string(); policy.entries.push(FileSystemSandboxEntry { path: FileSystemPath::GlobPattern { pattern: format!("{cwd_display}/**/*.env"), }, access: FileSystemAccessMode::Deny, }); policy}#[tokio::test]async fn turn_context_item_uses_turn_context_comp_hash_snapshot() { let (_session, mut turn_context) = make_session_and_context().await; turn_context.comp_hash = Some("turn-context-hash".to_string()); turn_context.model_info.comp_hash = Some("model-info-hash".to_string()); assert_eq!( turn_context.to_turn_context_item().comp_hash.as_deref(), Some("turn-context-hash") );}#[tokio::test]async fn turn_context_item_omits_legacy_equivalent_file_system_sandbox_policy() { let (_session, turn_context) = make_session_and_context().await; let item = turn_context.to_turn_context_item(); assert_eq!(item.file_system_sandbox_policy, None); assert_eq!( item.permission_profile, Some(turn_context.permission_profile()) );}#[tokio::test]async fn turn_context_item_stores_split_file_system_sandbox_policy_when_different() { let (_session, mut turn_context) = make_session_and_context().await; let file_system_sandbox_policy = file_system_policy_with_unreadable_glob(&turn_context); turn_context.permission_profile = PermissionProfile::from_runtime_permissions_with_enforcement( turn_context.permission_profile.enforcement(), &file_system_sandbox_policy, turn_context.network_sandbox_policy(), ); let item = turn_context.to_turn_context_item(); assert_eq!( item.file_system_sandbox_policy, Some(file_system_sandbox_policy) ); assert_eq!( item.permission_profile, Some(turn_context.permission_profile()) );}#[tokio::test]async fn record_context_updates_and_set_reference_context_item_injects_full_context_when_baseline_missing() { let (session, turn_context) = make_session_and_context().await; session .record_context_updates_and_set_reference_context_item(&turn_context) .await; let history = session.clone_history().await; let initial_context = session.build_initial_context(&turn_context).await; assert_eq!(history.raw_items().to_vec(), initial_context); let current_context = session.reference_context_item().await; assert_eq!( serde_json::to_value(current_context).expect("serialize current context item"), serde_json::to_value(Some(turn_context.to_turn_context_item())) .expect("serialize expected context item") );}#[tokio::test]async fn record_context_updates_and_set_reference_context_item_reinjects_full_context_after_clear(){ let (session, turn_context) = make_session_and_context().await; let compacted_summary = ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: format!("{}\nsummary", crate::compact::SUMMARY_PREFIX), }], phase: None, metadata: None, }; session .record_conversation_items(&turn_context, std::slice::from_ref(&compacted_summary)) .await; session .record_context_updates_and_set_reference_context_item(&turn_context) .await; { let mut state = session.state.lock().await; state.set_reference_context_item(/*item*/ None); } session .replace_history( vec![compacted_summary.clone()], /*reference_context_item*/ None, ) .await; session .record_context_updates_and_set_reference_context_item(&turn_context) .await; let history = session.clone_history().await; let mut expected_history = vec![compacted_summary]; let initial_context = session.build_initial_context(&turn_context).await; expected_history.extend(initial_context); assert_eq!(history.raw_items().to_vec(), expected_history);}#[tokio::test]async fn record_context_updates_and_set_reference_context_item_persists_baseline_without_emitting_diffs() { let (mut session, previous_context) = make_session_and_context().await; let next_model = if previous_context.model_info.slug == "gpt-5.4" { "gpt-5.2" } else { "gpt-5.4" }; let turn_context = previous_context .with_model(next_model.to_string(), &session.services.models_manager) .await; let previous_context_item = previous_context.to_turn_context_item(); { let mut state = session.state.lock().await; state.set_reference_context_item(Some(previous_context_item.clone())); } let rollout_path = attach_thread_persistence(&mut session).await; let update_items = session .build_settings_update_items(Some(&previous_context_item), &turn_context) .await; assert_eq!(update_items, Vec::new()); session .record_context_updates_and_set_reference_context_item(&turn_context) .await; assert_eq!( session.clone_history().await.raw_items().to_vec(), Vec::new() ); assert_eq!( serde_json::to_value(session.reference_context_item().await) .expect("serialize current context item"), serde_json::to_value(Some(turn_context.to_turn_context_item())) .expect("serialize expected context item") ); session.ensure_rollout_materialized().await; session.flush_rollout().await.expect("rollout should flush"); let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path) .await .expect("read rollout history") else { panic!("expected resumed rollout history"); }; let persisted_turn_context = resumed.history.iter().find_map(|item| match item { RolloutItem::TurnContext(ctx) => Some(ctx.clone()), _ => None, }); assert_eq!( serde_json::to_value(persisted_turn_context) .expect("serialize persisted turn context item"), serde_json::to_value(Some(turn_context.to_turn_context_item())) .expect("serialize expected turn context item") );}#[tokio::test]async fn record_context_updates_and_set_reference_context_item_persists_split_file_system_policy_to_rollout() { let (mut session, mut turn_context) = make_session_and_context().await; let file_system_sandbox_policy = file_system_policy_with_unreadable_glob(&turn_context); turn_context.permission_profile = PermissionProfile::from_runtime_permissions_with_enforcement( turn_context.permission_profile.enforcement(), &file_system_sandbox_policy, turn_context.network_sandbox_policy(), ); let rollout_path = attach_thread_persistence(&mut session).await; session .record_context_updates_and_set_reference_context_item(&turn_context) .await; session.ensure_rollout_materialized().await; session.flush_rollout().await.expect("rollout should flush"); let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path) .await .expect("read rollout history") else { panic!("expected resumed rollout history"); }; let persisted_file_system_sandbox_policy = resumed.history.iter().find_map(|item| match item { RolloutItem::TurnContext(ctx) => ctx.file_system_sandbox_policy.clone(), _ => None, }); assert_eq!( persisted_file_system_sandbox_policy, Some(file_system_sandbox_policy) );}#[tokio::test]async fn build_initial_context_prepends_model_switch_message() { let (session, turn_context) = make_session_and_context().await; let previous_turn_settings = PreviousTurnSettings { model: "previous-regular-model".to_string(), comp_hash: None, realtime_active: None, }; session .set_previous_turn_settings(Some(previous_turn_settings)) .await; let initial_context = session.build_initial_context(&turn_context).await; let ResponseItem::Message { role, content, .. } = &initial_context[0] else { panic!("expected developer message"); }; assert_eq!(role, "developer"); let [ContentItem::InputText { text }, ..] = content.as_slice() else { panic!("expected developer text"); }; assert!(text.contains("<model_switch>"));}#[tokio::test]async fn record_context_updates_and_set_reference_context_item_persists_full_reinjection_to_rollout() { let (mut session, previous_context) = make_session_and_context().await; let next_model = if previous_context.model_info.slug == "gpt-5.4" { "gpt-5.2" } else { "gpt-5.4" }; let turn_context = previous_context .with_model(next_model.to_string(), &session.services.models_manager) .await; let rollout_path = attach_thread_persistence(&mut session).await; session .persist_rollout_items(&[RolloutItem::EventMsg(EventMsg::UserMessage( UserMessageEvent { client_id: None, message: "seed rollout".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() }, ))]) .await; { let mut state = session.state.lock().await; state.set_reference_context_item(/*item*/ None); } session .set_previous_turn_settings(Some(PreviousTurnSettings { model: previous_context.model_info.slug.clone(), comp_hash: None, realtime_active: Some(previous_context.realtime_active), })) .await; session .record_context_updates_and_set_reference_context_item(&turn_context) .await; session.ensure_rollout_materialized().await; session.flush_rollout().await.expect("rollout should flush"); let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path) .await .expect("read rollout history") else { panic!("expected resumed rollout history"); }; let persisted_turn_context = resumed.history.iter().find_map(|item| match item { RolloutItem::TurnContext(ctx) => Some(ctx.clone()), _ => None, }); assert_eq!( serde_json::to_value(persisted_turn_context) .expect("serialize persisted turn context item"), serde_json::to_value(Some(turn_context.to_turn_context_item())) .expect("serialize expected turn context item") );}#[tokio::test]async fn run_user_shell_command_does_not_set_reference_context_item() { let (session, _turn_context, rx) = make_session_and_context_with_rx().await; { let mut state = session.state.lock().await; state.set_reference_context_item(/*item*/ None); } handlers::run_user_shell_command(&session, "sub-id".to_string(), "echo shell".to_string()) .await; let deadline = StdDuration::from_secs(15); let start = std::time::Instant::now(); loop { let remaining = deadline.saturating_sub(start.elapsed()); let evt = tokio::time::timeout(remaining, rx.recv()) .await .expect("timeout waiting for event") .expect("event"); if matches!(evt.msg, EventMsg::TurnComplete(_)) { break; } } assert!( session.reference_context_item().await.is_none(), "standalone shell tasks should not mutate previous context" );}#[tokio::test]async fn realtime_conversation_list_voices_emits_builtin_list() { let (session, _turn_context, rx) = make_session_and_context_with_rx().await; handlers::realtime_conversation_list_voices(&session, "sub-id".to_string()).await; let event = rx.recv().await.expect("event"); let voices = match event.msg { EventMsg::RealtimeConversationListVoicesResponse( RealtimeConversationListVoicesResponseEvent { voices }, ) => voices, msg => panic!("expected list voices response, got {msg:?}"), }; assert_eq!( voices, RealtimeVoicesList { v1: vec![ RealtimeVoice::Juniper, RealtimeVoice::Maple, RealtimeVoice::Spruce, RealtimeVoice::Ember, RealtimeVoice::Vale, RealtimeVoice::Breeze, RealtimeVoice::Arbor, RealtimeVoice::Sol, RealtimeVoice::Cove, ], v2: vec![ RealtimeVoice::Alloy, RealtimeVoice::Ash, RealtimeVoice::Ballad, RealtimeVoice::Coral, RealtimeVoice::Echo, RealtimeVoice::Sage, RealtimeVoice::Shimmer, RealtimeVoice::Verse, RealtimeVoice::Marin, RealtimeVoice::Cedar, ], default_v1: RealtimeVoice::Cove, default_v2: RealtimeVoice::Marin, }, );}#[derive(Clone, Copy)]struct CompletingTask;impl SessionTask for CompletingTask { fn kind(&self) -> TaskKind { TaskKind::Regular } fn span_name(&self) -> &'static str { "session_task.completing" } async fn run( self: Arc<Self>, _session: Arc<SessionTaskContext>, _ctx: Arc<TurnContext>, _input: Vec<TurnInput>, _cancellation_token: CancellationToken, ) -> Option<String> { None }}#[derive(Clone, Copy)]struct NeverEndingTask { kind: TaskKind, listen_to_cancellation_token: bool,}impl SessionTask for NeverEndingTask { fn kind(&self) -> TaskKind { self.kind } fn span_name(&self) -> &'static str { "session_task.never_ending" } async fn run( self: Arc<Self>, _session: Arc<SessionTaskContext>, _ctx: Arc<TurnContext>, _input: Vec<TurnInput>, cancellation_token: CancellationToken, ) -> Option<String> { if self.listen_to_cancellation_token { cancellation_token.cancelled().await; return None; } loop { sleep(Duration::from_secs(60)).await; } }}#[derive(Clone, Copy)]struct GuardianDeniedApprovalTask;impl SessionTask for GuardianDeniedApprovalTask { fn kind(&self) -> TaskKind { TaskKind::Regular } fn span_name(&self) -> &'static str { "session_task.guardian_denied_approval" } async fn run( self: Arc<Self>, session: Arc<SessionTaskContext>, ctx: Arc<TurnContext>, _input: Vec<TurnInput>, cancellation_token: CancellationToken, ) -> Option<String> { let session = session.clone_session(); for _ in 0..3 { crate::guardian::record_guardian_denial_for_test(&session, &ctx, &ctx.sub_id).await; } cancellation_token.cancelled().await; None }}#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn guardian_auto_review_interrupts_after_three_consecutive_denials() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "trigger guardian denials".to_string(), text_elements: Vec::new(), }], client_id: None, }]; sess.spawn_task(Arc::clone(&tc), input, GuardianDeniedApprovalTask) .await; let mut observed = Vec::new(); let aborted = tokio::time::timeout(std::time::Duration::from_secs(5), async { loop { let event = rx.recv().await.expect("event"); if let EventMsg::TurnAborted(event) = &event.msg { let event = event.clone(); observed.push(EventMsg::TurnAborted(event.clone())); break event; } observed.push(event.msg); } }) .await .unwrap_or_else(|_| { panic!( "guardian denial circuit breaker should interrupt the turn; observed events: {observed:?}" ) }); assert_eq!(aborted.reason, TurnAbortReason::Interrupted);}#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn guardian_helper_review_interrupts_after_three_consecutive_denials() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "keep turn active for helper reviews".to_string(), text_elements: Vec::new(), }], client_id: None, }]; sess.spawn_task( Arc::clone(&tc), input, NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; let session_for_review = Arc::clone(&sess); let turn_for_review = Arc::clone(&tc); let turn_id = tc.sub_id.clone(); let review_thread = std::thread::spawn(move || { let runtime = tokio::runtime::Builder::new_current_thread() .enable_all() .build() .expect("helper review runtime"); runtime.block_on(async move { for _ in 0..3 { crate::guardian::record_guardian_denial_for_test( &session_for_review, &turn_for_review, &turn_id, ) .await; } }); }); review_thread.join().expect("helper review thread"); let mut observed = Vec::new(); let aborted = timeout(StdDuration::from_secs(5), async { loop { let event = rx.recv().await.expect("event"); if let EventMsg::TurnAborted(event) = &event.msg { let event = event.clone(); observed.push(EventMsg::TurnAborted(event.clone())); break event; } observed.push(event.msg); } }) .await .unwrap_or_else(|_| { panic!( "helper review circuit breaker should interrupt the turn; observed events: {observed:?}" ) }); assert_eq!(aborted.reason, TurnAbortReason::Interrupted);}#[tokio::test(flavor = "multi_thread", worker_threads = 2)]#[test_log::test]async fn abort_regular_task_emits_marker_before_turn_aborted() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], client_id: None, }]; sess.spawn_task( Arc::clone(&tc), input, NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: false, }, ) .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; // Interrupts surface the model-visible `<turn_aborted>` marker before the abort event. let marker_evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("timeout waiting for marker event") .expect("event"); assert!(matches!(marker_evt.msg, EventMsg::RawResponseItem(_))); let evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("timeout waiting for event") .expect("event"); match evt.msg { EventMsg::TurnAborted(e) => assert_eq!(TurnAbortReason::Interrupted, e.reason), other => panic!("unexpected event: {other:?}"), } // No extra events should be emitted after an abort. assert!(rx.try_recv().is_err());}#[tokio::test]async fn abort_gracefully_emits_marker_before_turn_aborted() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], client_id: None, }]; sess.spawn_task( Arc::clone(&tc), input, NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; // Gracefully cancelled tasks surface the model-visible marker before the abort event too. let marker_evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("timeout waiting for marker event") .expect("event"); assert!(matches!(marker_evt.msg, EventMsg::RawResponseItem(_))); let evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("timeout waiting for event") .expect("event"); match evt.msg { EventMsg::TurnAborted(e) => assert_eq!(TurnAbortReason::Interrupted, e.reason), other => panic!("unexpected event: {other:?}"), } // No extra events should be emitted after an abort. assert!(rx.try_recv().is_err());}#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], client_id: None, }]; sess.spawn_task( Arc::clone(&tc), input, NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: false, }, ) .await; while rx.try_recv().is_ok() {} let text_element = codex_protocol::user_input::TextElement::new( codex_protocol::user_input::ByteRange { start: 5, end: 12 }, Some("pending marker".to_string()), ); let pending_user_input = vec![UserInput::Text { text: "late pending input".to_string(), text_elements: vec![text_element.clone()], }]; sess.steer_input( pending_user_input.clone(), /*additional_context*/ Default::default(), Some(&tc.sub_id), /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, ) .await .expect("steer pending input into active turn"); sess.on_task_finished(Arc::clone(&tc), /*last_agent_message*/ None) .await; let history = sess.clone_history().await; let expected = ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "late pending input".to_string(), }], phase: None, metadata: None, }; assert!( history.raw_items().iter().any(|item| item == &expected), "expected pending input to be persisted into history on turn completion" ); let first = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("expected raw response item event") .expect("channel open"); assert!(matches!(first.msg, EventMsg::RawResponseItem(_))); let second = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("expected item started event") .expect("channel open"); assert!(matches!( second.msg, EventMsg::ItemStarted(ItemStartedEvent { item: TurnItem::UserMessage(UserMessageItem { content, .. }), .. }) if content == pending_user_input )); let third = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("expected item completed event") .expect("channel open"); assert!(matches!( third.msg, EventMsg::ItemCompleted(ItemCompletedEvent { item: TurnItem::UserMessage(UserMessageItem { content, .. }), .. }) if content == pending_user_input )); let fourth = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("expected legacy user message event") .expect("channel open"); assert!(matches!( fourth.msg, EventMsg::UserMessage(UserMessageEvent { client_id: None, message, images, text_elements, local_images, .. }) if message == "late pending input" && images == Some(Vec::new()) && text_elements == vec![text_element] && local_images.is_empty() )); let fifth = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("expected turn complete event") .expect("channel open"); assert!(matches!( fifth.msg, EventMsg::TurnComplete(TurnCompleteEvent { turn_id, last_agent_message: None, time_to_first_token_ms: None, .. }) if turn_id == tc.sub_id ));}#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn task_finish_emits_thread_idle_lifecycle_after_active_turn_clears() { struct ThreadIdleRecorder { calls: Arc<std::sync::atomic::AtomicUsize>, idle_tx: async_channel::Sender<()>, expected_thread_id: ThreadId, } impl codex_extension_api::ThreadLifecycleContributor<crate::config::Config> for ThreadIdleRecorder { fn on_thread_idle<'a>( &'a self, input: codex_extension_api::ThreadIdleInput<'a>, ) -> codex_extension_api::ExtensionFuture<'a, ()> { Box::pin(async move { assert_eq!( self.expected_thread_id.to_string(), input.thread_store.level_id() ); self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); self.idle_tx.send(()).await.expect("idle receiver open"); }) } } let (mut session, turn_context) = make_session_and_context().await; let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let (idle_tx, idle_rx) = async_channel::bounded(1); let mut builder = codex_extension_api::ExtensionRegistryBuilder::<crate::config::Config>::new(); builder.thread_lifecycle_contributor(Arc::new(ThreadIdleRecorder { calls: Arc::clone(&calls), idle_tx, expected_thread_id: session.thread_id, })); session.services.extensions = Arc::new(builder.build()); let session = Arc::new(session); session .spawn_task(Arc::new(turn_context), Vec::new(), CompletingTask) .await; timeout(StdDuration::from_secs(2), idle_rx.recv()) .await .expect("thread idle lifecycle") .expect("idle receiver open"); assert_eq!(1, calls.load(std::sync::atomic::Ordering::SeqCst)); assert!(session.active_turn.lock().await.is_none());}#[tokio::test]async fn thread_idle_lifecycle_waits_for_trigger_turn_mailbox_work() { struct ThreadIdleRecorder { calls: Arc<std::sync::atomic::AtomicUsize>, } impl codex_extension_api::ThreadLifecycleContributor<crate::config::Config> for ThreadIdleRecorder { fn on_thread_idle<'a>( &'a self, _input: codex_extension_api::ThreadIdleInput<'a>, ) -> codex_extension_api::ExtensionFuture<'a, ()> { Box::pin(async move { self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); }) } } let (mut session, _turn_context) = make_session_and_context().await; let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let mut builder = codex_extension_api::ExtensionRegistryBuilder::<crate::config::Config>::new(); builder.thread_lifecycle_contributor(Arc::new(ThreadIdleRecorder { calls: Arc::clone(&calls), })); session.services.extensions = Arc::new(builder.build()); session .input_queue .enqueue_mailbox_communication(InterAgentCommunication::new( AgentPath::root(), AgentPath::root(), Vec::new(), "pending trigger".to_string(), /*trigger_turn*/ true, )) .await; session.emit_thread_idle_lifecycle_if_idle().await; assert_eq!(0, calls.load(std::sync::atomic::Ordering::SeqCst));}#[tokio::test]async fn try_start_turn_if_idle_rejects_active_turn_without_injecting() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; sess.spawn_task( Arc::clone(&tc), Vec::new(), NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; let item = user_message("synthetic idle input"); let err = sess .try_start_turn_if_idle(vec![item.clone()]) .await .expect_err("active turn should reject idle-only input"); assert_eq!(TryStartTurnIfIdleRejectionReason::Busy, err.reason()); assert_eq!(vec![item], err.into_input()); assert_eq!( Vec::<TurnInput>::new(), sess.input_queue.get_pending_input(&sess.active_turn).await ); sess.abort_all_tasks(TurnAbortReason::Interrupted).await;}#[tokio::test]async fn try_start_turn_if_idle_rejects_plan_mode_without_injecting() { let (sess, _tc, _rx) = make_session_and_context_with_rx().await; let mut collaboration_mode = sess.collaboration_mode().await; collaboration_mode.mode = ModeKind::Plan; { let mut state = sess.state.lock().await; state.session_configuration.collaboration_mode = collaboration_mode; } let item = user_message("synthetic idle input"); let err = sess .try_start_turn_if_idle(vec![item.clone()]) .await .expect_err("plan mode should reject automatic idle input"); assert_eq!(TryStartTurnIfIdleRejectionReason::PlanMode, err.reason()); assert_eq!(vec![item], err.into_input()); assert!(sess.active_turn.lock().await.is_none()); assert_eq!( Vec::<TurnInput>::new(), sess.input_queue.get_pending_input(&sess.active_turn).await );}#[tokio::test]async fn try_start_turn_if_idle_rejects_pending_trigger_turn_without_injecting() { let (sess, _tc, _rx) = make_session_and_context_with_rx().await; sess.input_queue .enqueue_mailbox_communication(InterAgentCommunication::new( AgentPath::root(), AgentPath::root(), Vec::new(), "pending trigger".to_string(), /*trigger_turn*/ true, )) .await; let item = user_message("synthetic idle input"); let err = sess .try_start_turn_if_idle(vec![item.clone()]) .await .expect_err("pending trigger-turn mail should reject automatic idle input"); assert_eq!( TryStartTurnIfIdleRejectionReason::PendingTriggerTurn, err.reason() ); assert_eq!(vec![item], err.into_input()); assert!(sess.active_turn.lock().await.is_none()); assert!(sess.input_queue.has_trigger_turn_mailbox_items().await);}#[tokio::test]async fn try_start_turn_if_idle_rejects_active_review_turn_without_injecting() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; sess.spawn_task( Arc::clone(&tc), Vec::new(), NeverEndingTask { kind: TaskKind::Review, listen_to_cancellation_token: true, }, ) .await; let item = user_message("synthetic idle input"); let err = sess .try_start_turn_if_idle(vec![item.clone()]) .await .expect_err("active review turn should reject automatic idle input"); assert_eq!(TryStartTurnIfIdleRejectionReason::Busy, err.reason()); assert_eq!(vec![item], err.into_input()); assert_eq!( Vec::<TurnInput>::new(), sess.input_queue.get_pending_input(&sess.active_turn).await ); sess.abort_all_tasks(TurnAbortReason::Interrupted).await;}#[tokio::test]async fn steer_input_requires_active_turn() { let (sess, _tc, _rx) = make_session_and_context_with_rx().await; let input = vec![UserInput::Text { text: "steer".to_string(), text_elements: Vec::new(), }]; let err = sess .steer_input( input, /*additional_context*/ Default::default(), /*expected_turn_id*/ None, /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, ) .await .expect_err("steering without active turn should fail"); assert!(matches!(err, SteerInputError::NoActiveTurn(_)));}#[tokio::test]async fn steer_input_enforces_expected_turn_id() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], client_id: None, }]; sess.spawn_task( Arc::clone(&tc), input, NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: false, }, ) .await; let steer_input = vec![UserInput::Text { text: "steer".to_string(), text_elements: Vec::new(), }]; let err = sess .steer_input( steer_input, /*additional_context*/ Default::default(), Some("different-turn-id"), /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, ) .await .expect_err("mismatched expected turn id should fail"); match err { SteerInputError::ExpectedTurnMismatch { expected, actual } => { assert_eq!( (expected, actual), ("different-turn-id".to_string(), tc.sub_id.clone()) ); } other => panic!("unexpected error: {other:?}"), }}#[tokio::test]async fn steer_input_rejects_non_regular_turns() { for (task_kind, turn_kind) in [ (TaskKind::Review, NonSteerableTurnKind::Review), (TaskKind::Compact, NonSteerableTurnKind::Compact), ] { let (sess, _tc, _rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], client_id: None, }]; let turn_context = sess.new_default_turn_with_sub_id("turn".to_string()).await; sess.spawn_task( turn_context, input, NeverEndingTask { kind: task_kind, listen_to_cancellation_token: true, }, ) .await; let steer_input = vec![UserInput::Text { text: "steer".to_string(), text_elements: Vec::new(), }]; let err = sess .steer_input( steer_input, /*additional_context*/ Default::default(), /*expected_turn_id*/ None, /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, ) .await .expect_err("steering a non-regular turn should fail"); assert_eq!(err, SteerInputError::ActiveTurnNotSteerable { turn_kind }); sess.abort_all_tasks(TurnAbortReason::Interrupted).await; }}#[tokio::test]async fn steer_input_returns_active_turn_id() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], client_id: None, }]; sess.spawn_task( Arc::clone(&tc), input, NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: false, }, ) .await; let steer_input = vec![UserInput::Text { text: "steer".to_string(), text_elements: Vec::new(), }]; let turn_id = sess .steer_input( steer_input, /*additional_context*/ Default::default(), Some(&tc.sub_id), /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, ) .await .expect("steering with matching expected turn id should succeed"); assert_eq!(turn_id, tc.sub_id); assert!(sess.input_queue.has_pending_input(&sess.active_turn).await);}#[tokio::test]async fn abort_empty_active_turn_preserves_pending_input() { let (sess, _tc, _rx) = make_session_and_context_with_rx().await; let pending_item = ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "late pending input".to_string(), }], phase: None, metadata: None, }; let turn_state = { let mut active = sess.active_turn.lock().await; let active_turn = active.get_or_insert_with(ActiveTurn::default); Arc::clone(&active_turn.turn_state) }; sess.input_queue .extend_pending_input_for_turn_state( turn_state.as_ref(), vec![TurnInput::ResponseItem(pending_item.clone())], ) .await; sess.abort_all_tasks(TurnAbortReason::Replaced).await; assert!(sess.active_turn.lock().await.is_none()); assert_eq!( sess.input_queue .take_pending_input_for_turn_state(turn_state.as_ref()) .await, vec![TurnInput::ResponseItem(pending_item)] );}async fn set_total_token_usage(sess: &Session, total_token_usage: TokenUsage) { let mut state = sess.state.lock().await; state.set_token_info(Some(TokenUsageInfo { total_token_usage, last_token_usage: TokenUsage::default(), model_context_window: None, }));}#[tokio::test]async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; let communication = InterAgentCommunication::new( AgentPath::try_from("/root/worker").expect("worker path should parse"), AgentPath::root(), Vec::new(), "late queue-only update".to_string(), /*trigger_turn*/ false, ); sess.spawn_task( Arc::clone(&tc), Vec::new(), NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) .await; sess.input_queue .enqueue_mailbox_communication(communication.clone()) .await; assert!( !sess.input_queue.has_pending_input(&sess.active_turn).await, "queue-only mailbox mail should stay buffered once the current turn emitted its answer" ); assert_eq!( sess.input_queue.get_pending_input(&sess.active_turn).await, Vec::new() ); sess.abort_all_tasks(TurnAbortReason::Replaced).await; assert_eq!( sess.input_queue.get_pending_input(&sess.active_turn).await, vec![TurnInput::InterAgentCommunication(communication)], );}#[tokio::test]async fn trigger_turn_mailbox_mail_waits_for_next_turn_after_answer_boundary() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; sess.spawn_task( Arc::clone(&tc), Vec::new(), NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) .await; sess.input_queue .enqueue_mailbox_communication(InterAgentCommunication::new( AgentPath::try_from("/root/worker").expect("worker path should parse"), AgentPath::root(), Vec::new(), "late trigger update".to_string(), /*trigger_turn*/ true, )) .await; assert!( !sess.input_queue.has_pending_input(&sess.active_turn).await, "trigger-turn mailbox mail should not extend the current turn after its answer boundary" ); sess.abort_all_tasks(TurnAbortReason::Replaced).await; assert!(sess.input_queue.has_trigger_turn_mailbox_items().await);}#[tokio::test]async fn steered_input_reopens_mailbox_delivery_for_current_turn() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; let communication = InterAgentCommunication::new( AgentPath::try_from("/root/worker").expect("worker path should parse"), AgentPath::root(), Vec::new(), "queued child update".to_string(), /*trigger_turn*/ false, ); sess.spawn_task( Arc::clone(&tc), Vec::new(), NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) .await; sess.input_queue .enqueue_mailbox_communication(communication.clone()) .await; sess.steer_input( vec![UserInput::Text { text: "follow up".to_string(), text_elements: Vec::new(), }], /*additional_context*/ Default::default(), Some(&tc.sub_id), /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, ) .await .expect("steered input should be accepted"); assert_eq!( sess.input_queue.get_pending_input(&sess.active_turn).await, vec![ TurnInput::UserInput { content: vec![UserInput::Text { text: "follow up".to_string(), text_elements: Vec::new(), }], client_id: None }, TurnInput::InterAgentCommunication(communication), ], );}#[tokio::test]async fn stale_defer_mailbox_delivery_does_not_override_steered_input() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; let communication = InterAgentCommunication::new( AgentPath::try_from("/root/worker").expect("worker path should parse"), AgentPath::root(), Vec::new(), "queued child update".to_string(), /*trigger_turn*/ false, ); sess.spawn_task( Arc::clone(&tc), Vec::new(), NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) .await; sess.input_queue .enqueue_mailbox_communication(communication.clone()) .await; sess.steer_input( vec![UserInput::Text { text: "follow up".to_string(), text_elements: Vec::new(), }], /*additional_context*/ Default::default(), Some(&tc.sub_id), /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, ) .await .expect("steered input should be accepted"); sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) .await; assert_eq!( sess.input_queue.get_pending_input(&sess.active_turn).await, vec![ TurnInput::UserInput { content: vec![UserInput::Text { text: "follow up".to_string(), text_elements: Vec::new(), }], client_id: None }, TurnInput::InterAgentCommunication(communication), ], );}#[tokio::test]async fn tool_calls_reopen_mailbox_delivery_for_current_turn() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; let communication = InterAgentCommunication::new( AgentPath::try_from("/root/worker").expect("worker path should parse"), AgentPath::root(), Vec::new(), "queued child update".to_string(), /*trigger_turn*/ false, ); sess.spawn_task( Arc::clone(&tc), Vec::new(), NeverEndingTask { kind: TaskKind::Regular, listen_to_cancellation_token: true, }, ) .await; sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) .await; sess.input_queue .enqueue_mailbox_communication(communication.clone()) .await; let item = ResponseItem::FunctionCall { id: None, name: "test_tool".to_string(), namespace: None, arguments: "{}".to_string(), call_id: "call-1".to_string(), metadata: None, }; let mut ctx = HandleOutputCtx { sess: Arc::clone(&sess), turn_context: Arc::clone(&tc), turn_store: Arc::new(codex_extension_api::ExtensionData::new(tc.sub_id.clone())), tool_runtime: test_tool_runtime(Arc::clone(&sess), Arc::clone(&tc)), cancellation_token: CancellationToken::new(), }; let output = handle_output_item_done(&mut ctx, item, /*previously_active_item*/ None) .await .expect("tool call should be handled"); assert!(output.needs_follow_up); assert!(output.tool_future.is_some()); assert_eq!( sess.input_queue.get_pending_input(&sess.active_turn).await, vec![TurnInput::InterAgentCommunication(communication)], );}#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn abort_review_task_emits_exited_then_aborted_and_records_history() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "start review".to_string(), text_elements: Vec::new(), }], client_id: None, }]; sess.spawn_task(Arc::clone(&tc), input, ReviewTask::new()) .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; // Aborting a review task should exit review mode before surfacing the abort to the client. // We scan for these events (rather than relying on fixed ordering) since unrelated events // may interleave. let mut exited_review_mode_idx = None; let mut turn_aborted_idx = None; let mut idx = 0usize; let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(3); while tokio::time::Instant::now() < deadline { let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); let evt = tokio::time::timeout(remaining, rx.recv()) .await .expect("timeout waiting for event") .expect("event"); let event_idx = idx; idx = idx.saturating_add(1); match evt.msg { EventMsg::ExitedReviewMode(ev) => { assert!(ev.review_output.is_none()); exited_review_mode_idx = Some(event_idx); } EventMsg::TurnAborted(ev) => { assert_eq!(TurnAbortReason::Interrupted, ev.reason); turn_aborted_idx = Some(event_idx); break; } _ => {} } } assert!( exited_review_mode_idx.is_some(), "expected ExitedReviewMode after abort" ); assert!( turn_aborted_idx.is_some(), "expected TurnAborted after abort" ); assert!( exited_review_mode_idx.unwrap() < turn_aborted_idx.unwrap(), "expected ExitedReviewMode before TurnAborted" ); let history = sess.clone_history().await; // Verify the `<turn_aborted>` marker is still recorded in history for the model. assert!( history.raw_items().iter().any(|item| { let ResponseItem::Message { role, content, .. } = item else { return false; }; if role != "user" { return false; } content.iter().any(|content_item| { let ContentItem::InputText { text } = content_item else { return false; }; TurnAborted::matches_text(text) }) }), "expected a model-visible turn aborted marker in history after interrupt" );}#[tokio::test]async fn fatal_tool_error_stops_turn_and_reports_error() { let (session, turn_context, _rx) = make_session_and_context_with_rx().await; let tools = { session .services .mcp_connection_manager .load_full() .list_all_tools() .await }; let deferred_mcp_tools = Some(tools.clone()); let router = ToolRouter::from_turn_context( &turn_context, crate::tools::router::ToolRouterParams { deferred_mcp_tools, mcp_tools: Some(tools), discoverable_tools: None, extension_tool_executors: Vec::new(), dynamic_tools: turn_context.dynamic_tools.as_slice(), }, &Default::default(), ); let item = ResponseItem::CustomToolCall { id: None, status: None, call_id: "call-1".to_string(), name: "shell_command".to_string(), input: "{}".to_string(), metadata: None, }; let call = ToolRouter::build_tool_call(item.clone()) .expect("build tool call") .expect("tool call present"); let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); let err = router .dispatch_tool_call_with_code_mode_result( Arc::clone(&session), Arc::clone(&turn_context), CancellationToken::new(), tracker, call, ToolCallSource::Direct, ) .await .err() .expect("expected fatal error"); match err { FunctionCallError::Fatal(message) => { assert_eq!( message, "tool shell_command invoked with incompatible payload" ); } other => panic!("expected FunctionCallError::Fatal, got {other:?}"), }}async fn sample_rollout( session: &Session, _turn_context: &TurnContext,) -> (Vec<RolloutItem>, Vec<ResponseItem>) { let mut rollout_items = Vec::new(); let mut live_history = ContextManager::new(); // Use the same turn_context source as record_initial_history so model_info (and thus // personality_spec) matches reconstruction. let reconstruction_turn = session.new_default_turn().await; let mut initial_context = session .build_initial_context(reconstruction_turn.as_ref()) .await; // Ensure personality_spec is present when Personality is enabled, so expected matches // what reconstruction produces (build_initial_context may omit it when baked into model). if !initial_context.iter().any(|m| { matches!(m, ResponseItem::Message { role, content, .. } if role == "developer" && content.iter().any(|c| { matches!(c, ContentItem::InputText { text } if text.contains("<personality_spec>")) })) }) && let Some(p) = reconstruction_turn.personality && session.features.enabled(Feature::Personality) && let Some(personality_message) = reconstruction_turn .model_info .model_messages .as_ref() .and_then(|m| m.get_personality_message(Some(p)).filter(|s| !s.is_empty())) { let msg = crate::context::ContextualUserFragment::into( crate::context::PersonalitySpecInstructions::new(personality_message), ); let insert_at = initial_context .iter() .position(|m| matches!(m, ResponseItem::Message { role, .. } if role == "developer")) .map(|i| i + 1) .unwrap_or(0); initial_context.insert(insert_at, msg); } for item in &initial_context { rollout_items.push(RolloutItem::ResponseItem(item.clone())); } live_history.record_items( initial_context.iter(), reconstruction_turn.truncation_policy, ); let user1 = ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "first user".to_string(), }], phase: None, metadata: None, }; live_history.record_items( std::iter::once(&user1), reconstruction_turn.truncation_policy, ); rollout_items.push(RolloutItem::ResponseItem(user1.clone())); let assistant1 = ResponseItem::Message { id: None, role: "assistant".to_string(), content: vec![ContentItem::OutputText { text: "assistant reply one".to_string(), }], phase: None, metadata: None, }; live_history.record_items( std::iter::once(&assistant1), reconstruction_turn.truncation_policy, ); rollout_items.push(RolloutItem::ResponseItem(assistant1.clone())); let summary1 = "summary one"; let snapshot1 = live_history .clone() .for_prompt(&reconstruction_turn.model_info.input_modalities); let user_messages1 = collect_user_messages(&snapshot1); let rebuilt1 = compact::build_compacted_history(Vec::new(), &user_messages1, summary1); live_history.replace(rebuilt1); rollout_items.push(RolloutItem::Compacted(CompactedItem { message: summary1.to_string(), replacement_history: None, window_id: None, })); let user2 = ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "second user".to_string(), }], phase: None, metadata: None, }; live_history.record_items( std::iter::once(&user2), reconstruction_turn.truncation_policy, ); rollout_items.push(RolloutItem::ResponseItem(user2.clone())); let assistant2 = ResponseItem::Message { id: None, role: "assistant".to_string(), content: vec![ContentItem::OutputText { text: "assistant reply two".to_string(), }], phase: None, metadata: None, }; live_history.record_items( std::iter::once(&assistant2), reconstruction_turn.truncation_policy, ); rollout_items.push(RolloutItem::ResponseItem(assistant2.clone())); let summary2 = "summary two"; let snapshot2 = live_history .clone() .for_prompt(&reconstruction_turn.model_info.input_modalities); let user_messages2 = collect_user_messages(&snapshot2); let rebuilt2 = compact::build_compacted_history(Vec::new(), &user_messages2, summary2); live_history.replace(rebuilt2); rollout_items.push(RolloutItem::Compacted(CompactedItem { message: summary2.to_string(), replacement_history: None, window_id: None, })); let user3 = ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "third user".to_string(), }], phase: None, metadata: None, }; live_history.record_items( std::iter::once(&user3), reconstruction_turn.truncation_policy, ); rollout_items.push(RolloutItem::ResponseItem(user3)); let assistant3 = ResponseItem::Message { id: None, role: "assistant".to_string(), content: vec![ContentItem::OutputText { text: "assistant reply three".to_string(), }], phase: None, metadata: None, }; live_history.record_items( std::iter::once(&assistant3), reconstruction_turn.truncation_policy, ); rollout_items.push(RolloutItem::ResponseItem(assistant3)); ( rollout_items, live_history.for_prompt(&reconstruction_turn.model_info.input_modalities), )}#[tokio::test]async fn rejects_escalated_permissions_when_policy_not_on_request() { use crate::exec_policy::ExecApprovalRequest; use crate::sandboxing::SandboxPermissions; use crate::tools::sandboxing::ExecApprovalRequirement; use crate::turn_diff_tracker::TurnDiffTracker; use codex_protocol::protocol::AskForApproval; use codex_tools::ShellCommandBackendConfig; let (session, mut turn_context_raw) = make_session_and_context().await; // Ensure policy is NOT OnRequest so the early rejection path triggers turn_context_raw .approval_policy .set(AskForApproval::OnFailure) .expect("test setup should allow updating approval policy"); let session = Arc::new(session); let mut turn_context = Arc::new(turn_context_raw); let command_script = "echo hi"; let timeout_ms = 1000; let sandbox_permissions = SandboxPermissions::RequireEscalated; let turn_diff_tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); let tool_name = "shell_command"; let call_id = "test-call".to_string(); let handler = ShellCommandHandler::from(ShellCommandBackendConfig::Classic); #[allow(deprecated)] let workdir = Some(turn_context.cwd.to_string_lossy().to_string()); let resp = handler .handle(ToolInvocation { session: Arc::clone(&session), turn: Arc::clone(&turn_context), cancellation_token: CancellationToken::new(), tracker: Arc::clone(&turn_diff_tracker), call_id, tool_name: codex_tools::ToolName::plain(tool_name), source: crate::tools::context::ToolCallSource::Direct, payload: ToolPayload::Function { arguments: serde_json::json!({ "command": command_script, "workdir": workdir, "timeout_ms": timeout_ms, "sandbox_permissions": sandbox_permissions, "justification": Some("test"), }) .to_string(), }, }) .await; let Err(FunctionCallError::RespondToModel(output)) = resp else { panic!("expected error result"); }; let expected = format!( "approval policy is {policy:?}; reject command — you should not ask for escalated permissions if the approval policy is {policy:?}", policy = turn_context.approval_policy.value() ); pretty_assertions::assert_eq!(output, expected); pretty_assertions::assert_eq!( session .granted_turn_permissions(codex_exec_server::LOCAL_ENVIRONMENT_ID) .await, None ); // The rejection should not poison the non-escalated path for the same // command. Force DangerFullAccess so this check stays focused on approval // policy rather than platform-specific sandbox behavior. let turn_context_mut = Arc::get_mut(&mut turn_context).expect("unique thread settings Arc"); turn_context_mut.permission_profile = PermissionProfile::Disabled; let command = session.user_shell().derive_exec_args( command_script, turn_context.config.permissions.allow_login_shell, ); let exec_approval_requirement = session .services .exec_policy .create_exec_approval_requirement_for_command(ExecApprovalRequest { command: &command, approval_policy: turn_context.approval_policy.value(), permission_profile: turn_context.permission_profile(), windows_sandbox_level: turn_context.windows_sandbox_level, sandbox_permissions: SandboxPermissions::UseDefault, prefix_rule: None, }) .await; assert!(matches!( exec_approval_requirement, ExecApprovalRequirement::Skip { .. } ));}#[cfg(unix)]#[tokio::test]async fn shell_tool_cancellation_waits_for_runtime_cleanup() -> anyhow::Result<()> { let session = make_session_with_config(|config| { let cwd = config.cwd.clone(); config .permissions .set_legacy_sandbox_policy(SandboxPolicy::DangerFullAccess, cwd.as_path()) .expect("test setup should allow sandbox policy"); }) .await?; let turn_context = session.new_default_turn().await; let session = Arc::new(session); let turn_context = Arc::new(turn_context); let temp_dir = tempfile::TempDir::new()?; let ready_marker = temp_dir.path().join("ready"); let cleanup_marker = temp_dir.path().join("cleanup"); // Interrupt after the shell starts, then verify dispatch waits for its TERM cleanup trap. let command = format!( r#"trap 'printf cleaned > "{}"; exit 0' TERMprintf ready > "{}"while :; do sleep 1; done"#, cleanup_marker.display(), ready_marker.display(), ); let item = ResponseItem::FunctionCall { id: None, name: "shell_command".to_string(), namespace: None, arguments: serde_json::json!({ "command": command, "timeout_ms": 60_000, }) .to_string(), call_id: "shell-cleanup-call".to_string(), metadata: None, }; let call = ToolRouter::build_tool_call(item)? .expect("shell command response item should build a tool call"); let cancellation_token = CancellationToken::new(); let cancellation_tx = cancellation_token.clone(); let handle = tokio::spawn( test_tool_runtime(Arc::clone(&session), Arc::clone(&turn_context)) .handle_tool_call(call, cancellation_token), ); let mut ready = false; for _ in 0..50 { if ready_marker.exists() { ready = true; break; } tokio::time::sleep(Duration::from_millis(20)).await; } if !ready { cancellation_tx.cancel(); let _ = timeout(Duration::from_secs(5), handle).await; anyhow::bail!("shell command should reach the ready marker"); } cancellation_tx.cancel(); timeout(Duration::from_secs(5), handle) .await .expect("cancelled shell tool should finish promptly") .expect("shell tool task should join") .expect("cancelled shell tool should return a response item"); assert_eq!(std::fs::read_to_string(cleanup_marker)?, "cleaned"); Ok(())}#[tokio::test]async fn unified_exec_rejects_escalated_permissions_when_policy_not_on_request() { use crate::sandboxing::SandboxPermissions; use crate::turn_diff_tracker::TurnDiffTracker; use codex_protocol::protocol::AskForApproval; let (session, mut turn_context_raw) = make_session_and_context().await; turn_context_raw .approval_policy .set(AskForApproval::OnFailure) .expect("test setup should allow updating approval policy"); let session = Arc::new(session); let turn_context = Arc::new(turn_context_raw); let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); let handler = ExecCommandHandler::default(); let resp = handler .handle(ToolInvocation { session: Arc::clone(&session), turn: Arc::clone(&turn_context), cancellation_token: CancellationToken::new(), tracker: Arc::clone(&tracker), call_id: "exec-call".to_string(), tool_name: codex_tools::ToolName::plain("exec_command"), source: crate::tools::context::ToolCallSource::Direct, payload: ToolPayload::Function { arguments: serde_json::json!({ "cmd": "echo hi", "sandbox_permissions": SandboxPermissions::RequireEscalated, "justification": "need unsandboxed execution", }) .to_string(), }, }) .await; let Err(FunctionCallError::RespondToModel(output)) = resp else { panic!("expected error result"); }; let expected = format!( "approval policy is {policy:?}; reject command — you cannot ask for escalated permissions if the approval policy is {policy:?}", policy = turn_context.approval_policy.value() ); pretty_assertions::assert_eq!(output, expected);}#[tokio::test]async fn session_start_hooks_only_load_from_trusted_project_layers() -> std::io::Result<()> { let temp = tempfile::tempdir()?; let codex_home = temp.path().join("home"); let project_root = temp.path().join("project"); let nested = project_root.join("nested"); let root_dot_codex = project_root.join(".codex"); let nested_dot_codex = nested.join(".codex"); std::fs::create_dir_all(&codex_home)?; std::fs::create_dir_all(&nested_dot_codex)?; std::fs::write(project_root.join(".git"), "gitdir: here")?; write_project_hooks(&root_dot_codex)?; write_project_hooks(&nested_dot_codex)?; write_project_trust_config(&codex_home, &[(&nested, TrustLevel::Trusted)]).await?; let config = ConfigBuilder::default() .codex_home(codex_home) .fallback_cwd(Some(nested)) .build() .await?; let hook_list = codex_hooks::list_hooks(codex_hooks::HooksConfig { feature_enabled: true, config_layer_stack: Some(config.config_layer_stack.clone()), ..codex_hooks::HooksConfig::default() }); let expected_source_path = codex_utils_absolute_path::AbsolutePathBuf::from_absolute_path( nested_dot_codex.join("hooks.json"), )?; assert_eq!( hook_list .hooks .iter() .map(|hook| &hook.source_path) .collect::<Vec<_>>(), vec![&expected_source_path], ); assert_eq!( hook_list.hooks[0].trust_status, codex_protocol::protocol::HookTrustStatus::Untrusted ); assert!(preview_session_start_hooks(&config).await?.is_empty()); Ok(())}#[tokio::test]async fn session_start_hooks_require_project_trust_without_config_toml() -> std::io::Result<()> { let temp = tempfile::tempdir()?; let project_root = temp.path().join("project"); let nested = project_root.join("nested"); let dot_codex = project_root.join(".codex"); std::fs::create_dir_all(&nested)?; std::fs::write(project_root.join(".git"), "gitdir: here")?; write_project_hooks(&dot_codex)?; let cases = [ ("unknown", Vec::<(&Path, TrustLevel)>::new(), 0_usize), ( "untrusted", vec![(&project_root as &Path, TrustLevel::Untrusted)], 0_usize, ), ( "trusted", vec![(&project_root as &Path, TrustLevel::Trusted)], 1_usize, ), ]; for (name, trust_entries, expected_hooks) in cases { let codex_home = temp.path().join(format!("home_{name}")); std::fs::create_dir_all(&codex_home)?; write_project_trust_config(&codex_home, &trust_entries).await?; let config = ConfigBuilder::default() .codex_home(codex_home) .fallback_cwd(Some(nested.clone())) .build() .await?; let hook_list = codex_hooks::list_hooks(codex_hooks::HooksConfig { feature_enabled: true, config_layer_stack: Some(config.config_layer_stack.clone()), ..codex_hooks::HooksConfig::default() }); assert_eq!( hook_list.hooks.len(), expected_hooks, "unexpected discovered hook count for {name}", ); assert!(preview_session_start_hooks(&config).await?.is_empty()); if expected_hooks == 1 { assert_eq!( hook_list.hooks[0].trust_status, codex_protocol::protocol::HookTrustStatus::Untrusted ); } } Ok(())}