diff --git a/codex-rs/analytics/src/tests/suite/turns/events_tests.rs b/codex-rs/analytics/src/tests/suite/turns/events_tests.rs index 5a86e9a058..5336744bbe 100644 --- a/codex-rs/analytics/src/tests/suite/turns/events_tests.rs +++ b/codex-rs/analytics/src/tests/suite/turns/events_tests.rs @@ -191,6 +191,8 @@ fn turn_event_serializes_expected_shape() { "critical_path_status": "disabled", "critical_path_duration_ms": null, "critical_path_inference_ms": null, + "critical_path_sampling_ms": null, + "critical_path_compaction_ms": null, "critical_path_auto_review_ms": null, "critical_path_user_wait_ms": null, "critical_path_tools_ms": null, @@ -982,35 +984,52 @@ async fn critical_path_payload_preserves_capture_status_and_emits_once() { for (critical_path, expected) in [ ( TurnCriticalPath::Disabled, - json!(["disabled", null, null, null, null, null, null, null]), + json!([ + "disabled", null, null, null, null, null, null, null, null, null + ]), ), ( TurnCriticalPath::Incomplete, - json!(["incomplete", null, null, null, null, null, null, null]), + json!([ + "incomplete", + null, + null, + null, + null, + null, + null, + null, + null, + null + ]), ), ( TurnCriticalPath::Complete(TurnCriticalPathDurations { duration_ms: 1234, inference_ms: 700, + sampling_ms: 500, + compaction_ms: 200, auto_review_ms: 200, user_wait_ms: 100, tools_ms: 150, overhead_ms: 80, unknown_ms: 4, }), - json!(["complete", 1234, 700, 200, 100, 150, 80, 4]), + json!(["complete", 1234, 700, 500, 200, 200, 100, 150, 80, 4]), ), ( TurnCriticalPath::Complete(TurnCriticalPathDurations { duration_ms: 1234, inference_ms: 1234, + sampling_ms: 1234, + compaction_ms: 0, auto_review_ms: 0, user_wait_ms: 0, tools_ms: 0, overhead_ms: 0, unknown_ms: 0, }), - json!(["complete", 1234, 1234, 0, 0, 0, 0, 0]), + json!(["complete", 1234, 1234, 1234, 0, 0, 0, 0, 0, 0]), ), ] { let mut reducer = AnalyticsReducer::default(); @@ -1058,6 +1077,8 @@ async fn critical_path_payload_preserves_capture_status_and_emits_once() { params["critical_path_status"], params["critical_path_duration_ms"], params["critical_path_inference_ms"], + params["critical_path_sampling_ms"], + params["critical_path_compaction_ms"], params["critical_path_auto_review_ms"], params["critical_path_user_wait_ms"], params["critical_path_tools_ms"], diff --git a/codex-rs/analytics/src/turn_critical_path.rs b/codex-rs/analytics/src/turn_critical_path.rs index f66ae642ec..4d10fca22c 100644 --- a/codex-rs/analytics/src/turn_critical_path.rs +++ b/codex-rs/analytics/src/turn_critical_path.rs @@ -30,12 +30,15 @@ impl TurnCriticalPath { } /// Category totals over the server turn interval, truncated to milliseconds. -/// Sampling and compaction both contribute to inference. Independent truncation -/// can leave the category sum up to five milliseconds below duration_ms. +/// Inference is the compatibility total of sampling and compaction, not another leaf. +/// Independent truncation can leave the seven leaf categories up to six milliseconds +/// below duration_ms, and sampling_ms + compaction_ms up to one below inference_ms. #[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct TurnCriticalPathDurations { pub duration_ms: u64, pub inference_ms: u64, + pub sampling_ms: u64, + pub compaction_ms: u64, pub auto_review_ms: u64, pub user_wait_ms: u64, pub tools_ms: u64, @@ -48,6 +51,8 @@ pub(crate) struct CriticalPathEventParams { critical_path_status: &'static str, critical_path_duration_ms: Option, critical_path_inference_ms: Option, + critical_path_sampling_ms: Option, + critical_path_compaction_ms: Option, critical_path_auto_review_ms: Option, critical_path_user_wait_ms: Option, critical_path_tools_ms: Option, @@ -62,6 +67,8 @@ impl From for CriticalPathEventParams { critical_path_status: path.status(), critical_path_duration_ms: durations.map(|value| value.duration_ms), critical_path_inference_ms: durations.map(|value| value.inference_ms), + critical_path_sampling_ms: durations.map(|value| value.sampling_ms), + critical_path_compaction_ms: durations.map(|value| value.compaction_ms), critical_path_auto_review_ms: durations.map(|value| value.auto_review_ms), critical_path_user_wait_ms: durations.map(|value| value.user_wait_ms), critical_path_tools_ms: durations.map(|value| value.tools_ms), diff --git a/codex-rs/app-server/tests/suite/v2/critical_path_analytics_tests.rs b/codex-rs/app-server/tests/suite/v2/critical_path_analytics_tests.rs index 426ad3d858..d681b72a87 100644 --- a/codex-rs/app-server/tests/suite/v2/critical_path_analytics_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/critical_path_analytics_tests.rs @@ -137,7 +137,8 @@ async fn critical_path_reaches_turn_event(finish: Finish, capture: Capture) -> R ); let fields = [ "critical_path_duration_ms", - "critical_path_inference_ms", + "critical_path_sampling_ms", + "critical_path_compaction_ms", "critical_path_auto_review_ms", "critical_path_user_wait_ms", "critical_path_tools_ms", @@ -146,6 +147,16 @@ async fn critical_path_reaches_turn_event(finish: Finish, capture: Capture) -> R ]; if matches!(capture, Capture::Enabled) { assert_eq!(p["critical_path_status"], "complete"); + let inference = p["critical_path_inference_ms"] + .as_u64() + .expect("inference duration"); + let phases = p["critical_path_sampling_ms"] + .as_u64() + .expect("sampling duration") + + p["critical_path_compaction_ms"] + .as_u64() + .expect("compaction duration"); + assert!(phases <= inference && inference - phases <= 1); assert_eq!(p[fields[0]], p["duration_ms"]); let total = p[fields[0]] .as_u64() @@ -158,7 +169,7 @@ async fn critical_path_reaches_turn_event(finish: Finish, capture: Capture) -> R .expect("complete capture has numeric categories") }) .sum(); - assert!(sum <= total && total - sum <= 5); + assert!(sum <= total && total - sum <= 6); assert_eq!( json!([ p["critical_path_user_wait_ms"], @@ -170,9 +181,13 @@ async fn critical_path_reaches_turn_event(finish: Finish, capture: Capture) -> R ); } else { assert_eq!(p["critical_path_status"], "disabled"); + assert_eq!( + p.get("critical_path_inference_ms"), + Some(&serde_json::Value::Null) + ); assert_eq!( fields.map(|key| p.get(key)), - [Some(&serde_json::Value::Null); 7] + [Some(&serde_json::Value::Null); 8] ); } Ok(()) diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 0c2f12e450..064d8e975b 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -4461,14 +4461,6 @@ impl Session { communication: impl Into, ) { let communication = communication.into(); - // Agent-message dependencies are not captured by direct-tool timing. - if turn_context - .config - .features - .enabled(Feature::CriticalPathTiming) - { - turn_context.turn_timing_state.invalidate_critical_path(); - } let mut response_item = communication.to_model_input_item(); if let ResponseItem::AgentMessage { content, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index f9c25a35c0..3704a9079b 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -2834,22 +2834,7 @@ async fn inter_agent_communication_waits_for_confirmed_delivery_persistence() { #[tokio::test] async fn record_inter_agent_communication_sets_turn_id_in_rollout_and_resume() { - let (mut session, mut turn_context) = make_session_and_context().await; - Arc::make_mut(&mut turn_context.config) - .features - .enable(Feature::CriticalPathTiming) - .unwrap(); - turn_context.turn_timing_state.enable_critical_path(); - turn_context - .turn_timing_state - .mark_turn_started(std::time::Instant::now()) - .await; - assert!( - turn_context - .turn_timing_state - .snapshot_path(std::time::Instant::now()) - .is_some() - ); + let (mut session, turn_context) = make_session_and_context().await; let rollout_path = attach_thread_persistence(&mut session).await; let communication = InterAgentCommunication::new( AgentPath::root().join("worker").expect("worker path"), @@ -2870,12 +2855,6 @@ async fn record_inter_agent_communication_sets_turn_id_in_rollout_and_resume() { .record_inter_agent_communication(&turn_context, turn_context.model_info(), communication) .await; - assert!( - turn_context - .turn_timing_state - .snapshot_path(std::time::Instant::now()) - .is_none() - ); let recorded_history = session.clone_history().await; let recorded_items = raw_history_items(&recorded_history); let create_time = recorded_items[0] diff --git a/codex-rs/core/src/tasks/mod_tests.rs b/codex-rs/core/src/tasks/mod_tests.rs index 03b71f49a0..64c9fff2e3 100644 --- a/codex-rs/core/src/tasks/mod_tests.rs +++ b/codex-rs/core/src/tasks/mod_tests.rs @@ -675,6 +675,8 @@ async fn turn_critical_path_metrics_count_each_capture_once() { let complete = TurnCriticalPath::Complete(TurnCriticalPathDurations { duration_ms: 60, inference_ms: 31, + sampling_ms: 31, + compaction_ms: 0, auto_review_ms: 2, tools_ms: 10, user_wait_ms: 0, diff --git a/codex-rs/core/src/tasks/regular.rs b/codex-rs/core/src/tasks/regular.rs index 718acfd862..fd91a1d557 100644 --- a/codex-rs/core/src/tasks/regular.rs +++ b/codex-rs/core/src/tasks/regular.rs @@ -2,6 +2,9 @@ use std::sync::Arc; use codex_async_utils::OrCancelExt; use codex_extension_api::TurnStartPhase; +use codex_protocol::protocol::InternalSessionSource; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; use futures::future::BoxFuture; use tokio_util::sync::CancellationToken; @@ -56,7 +59,11 @@ impl SessionTask for RegularTask { .config .features .enabled(codex_features::Feature::CriticalPathTiming) - && !ctx.session_source.is_non_root_agent() + && !matches!( + ctx.session_source, + SessionSource::Internal(InternalSessionSource::MemoryConsolidation) + | SessionSource::SubAgent(SubAgentSource::MemoryConsolidation) + ) { ctx.turn_timing_state.enable_critical_path(); } diff --git a/codex-rs/core/src/test_support.rs b/codex-rs/core/src/test_support.rs index f6ba23047a..b122aba784 100644 --- a/codex-rs/core/src/test_support.rs +++ b/codex-rs/core/src/test_support.rs @@ -143,7 +143,7 @@ pub async fn observe_turn_timing( timing.snapshot_path(end).map(|path| { let totals = path.durations(start, end); [ - totals.inference, + totals.sampling.saturating_add(totals.compaction), totals.auto_review, totals.tools, totals.overhead, diff --git a/codex-rs/core/src/tools/parallel.rs b/codex-rs/core/src/tools/parallel.rs index 4361a91aa7..8696ad4e66 100644 --- a/codex-rs/core/src/tools/parallel.rs +++ b/codex-rs/core/src/tools/parallel.rs @@ -61,15 +61,16 @@ struct DirectToolTiming { batch: ToolBatchTiming, } -// Remove this boundary when CodeMode and subagent dependencies have runtime capture. +// Direct spawn and wait calls use this turn's tool time; child work stays on its own turn. +// CodeMode and follow-up agent operations still lack runtime capture. fn direct_path_supported(tool_name: &codex_tools::ToolName, source: &ToolCallSource) -> bool { !(matches!(source, ToolCallSource::CodeMode { .. }) || (tool_name.is_default_namespace() && matches!(tool_name.name.as_str(), crate::tools::code_mode::PUBLIC_TOOL_NAME | crate::tools::code_mode::WAIT_TOOL_NAME)) // Multi-agent tool namespaces can be overridden by the tool catalog. - || matches!(tool_name.name.as_str(), "spawn_agent" | "send_input" - | "close_agent" | "resume_agent" | "wait_agent" | "list_agents" + || matches!(tool_name.name.as_str(), "send_input" + | "close_agent" | "resume_agent" | "list_agents" | "followup_task" | "interrupt_agent" | "send_message")) } @@ -630,7 +631,7 @@ mod tests { use tracing_test::internal::MockWriter; #[test] - fn unsupported_handlers_remain_incomplete_after_namespace_resolution() { + fn direct_tool_support_follows_namespace_resolution() { let handler = crate::tools::multi_agent_tool::multi_agent_v2_handler( crate::tools::handlers::multi_agents_v2::WaitAgentHandler::default(), Some("custom_collaboration"), @@ -644,13 +645,15 @@ mod tests { "wait_agent", )) .unwrap(); - assert!(!direct_path_supported( + assert!(direct_path_supported( ®istered.runtime.tool_name(), &ToolCallSource::Direct )); for name in [ crate::tools::code_mode::PUBLIC_TOOL_NAME, crate::tools::code_mode::WAIT_TOOL_NAME, + "followup_task", + "send_message", ] { assert!(!direct_path_supported( &codex_tools::ToolName::plain(name).with_default_namespace(), diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index aaae1cd7d8..ddac4878fa 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -415,11 +415,8 @@ impl TurnProfileState { fn extend_path(&self, path: CriticalPath, previous: Instant, now: Instant) -> CriticalPath { match self.active_phase { - // Compaction is model-related elapsed work, including local bookkeeping; - // the existing profile still exposes its separate compaction_ms duration. - Some(TurnProfilePhase::Sampling | TurnProfilePhase::Compaction) => { - path.work(previous, now, WorkKind::Inference) - } + Some(TurnProfilePhase::Sampling) => path.work(previous, now, WorkKind::Sampling), + Some(TurnProfilePhase::Compaction) => path.work(previous, now, WorkKind::Compaction), Some(TurnProfilePhase::ToolBlocking) => { let dependency = self .tool_dependency diff --git a/codex-rs/core/src/turn_timing/report.rs b/codex-rs/core/src/turn_timing/report.rs index 0a00a58353..21703e114b 100644 --- a/codex-rs/core/src/turn_timing/report.rs +++ b/codex-rs/core/src/turn_timing/report.rs @@ -19,7 +19,9 @@ pub(super) fn completed_critical_path(state: &TurnProfileState, end: Instant) -> let totals = path.durations(start, end); TurnCriticalPath::Complete(TurnCriticalPathDurations { duration_ms: duration_to_u64_ms(end.saturating_duration_since(start)), - inference_ms: duration_to_u64_ms(totals.inference), + inference_ms: duration_to_u64_ms(totals.sampling.saturating_add(totals.compaction)), + sampling_ms: duration_to_u64_ms(totals.sampling), + compaction_ms: duration_to_u64_ms(totals.compaction), auto_review_ms: duration_to_u64_ms(totals.auto_review), user_wait_ms: duration_to_u64_ms(totals.user_wait), tools_ms: duration_to_u64_ms(totals.tools), diff --git a/codex-rs/core/src/turn_timing_tests.rs b/codex-rs/core/src/turn_timing_tests.rs index d821fc6c2a..7c8b3d633d 100644 --- a/codex-rs/core/src/turn_timing_tests.rs +++ b/codex-rs/core/src/turn_timing_tests.rs @@ -288,6 +288,8 @@ fn turn_profile_counts_compaction_as_an_exclusive_phase() { codex_analytics::TurnCriticalPathDurations { duration_ms: 1100, inference_ms: 800, + sampling_ms: 300, + compaction_ms: 500, auto_review_ms: 0, user_wait_ms: 0, tools_ms: 0, @@ -311,7 +313,8 @@ fn turn_profile_counts_compaction_as_an_exclusive_phase() { .as_ref() .unwrap() .durations(started_at, started_at + Duration::from_millis(1_100)); - assert_eq!(totals.inference, Duration::from_millis(800)); + assert_eq!(totals.sampling, Duration::from_millis(300)); + assert_eq!(totals.compaction, Duration::from_millis(500)); assert_eq!(totals.overhead, Duration::from_millis(300)); assert_eq!(totals.user_wait, Duration::ZERO); } @@ -332,6 +335,8 @@ async fn turn_profile_and_duration_share_a_completion_instant() { codex_analytics::TurnCriticalPath::Complete(codex_analytics::TurnCriticalPathDurations { duration_ms: duration_ms.unwrap() as u64, inference_ms: 0, + sampling_ms: 0, + compaction_ms: 0, auto_review_ms: 0, user_wait_ms: 0, tools_ms: 0, @@ -373,7 +378,7 @@ fn capture_preserves_phases_and_freezes_missing_dependencies() { .unwrap() .durations(started_at, time_after_turn_start(60)), PathDurations { - inference: Duration::from_millis(10), + sampling: Duration::from_millis(10), unknown: Duration::from_millis(30), overhead: Duration::from_millis(20), ..PathDurations::default() @@ -384,6 +389,8 @@ fn capture_preserves_phases_and_freezes_missing_dependencies() { codex_analytics::TurnCriticalPath::Complete(codex_analytics::TurnCriticalPathDurations { duration_ms: 60, inference_ms: 10, + sampling_ms: 10, + compaction_ms: 0, auto_review_ms: 0, user_wait_ms: 0, tools_ms: 0, @@ -467,7 +474,10 @@ fn critical_path_report_retains_fractional_millisecond_rounding_and_compaction() ..Default::default() }; state.start(start); - state.begin_compaction(start); + state.begin_sampling(start); + let compact_at = start + Duration::from_nanos(999_999); + state.end_phase(compact_at, TurnProfilePhase::Sampling); + state.begin_compaction(compact_at); let end = start + Duration::from_nanos(1_999_999); state.end_phase(end, TurnProfilePhase::Compaction); let profile = state.complete(end); @@ -476,6 +486,8 @@ fn critical_path_report_retains_fractional_millisecond_rounding_and_compaction() codex_analytics::TurnCriticalPath::Complete(codex_analytics::TurnCriticalPathDurations { duration_ms: 1, inference_ms: 1, + sampling_ms: 0, + compaction_ms: 1, auto_review_ms: 0, user_wait_ms: 0, tools_ms: 0, diff --git a/codex-rs/core/tests/suite/critical_path_runtime.rs b/codex-rs/core/tests/suite/critical_path_runtime.rs index 49d3506ea8..f286d292f3 100644 --- a/codex-rs/core/tests/suite/critical_path_runtime.rs +++ b/codex-rs/core/tests/suite/critical_path_runtime.rs @@ -1,17 +1,26 @@ -//! Exercises dependency capture through real tool dispatch and admission gates. +//! Exercises per-turn timing through real tool dispatch and admission gates. //! Channels hold work open and establish overlap without scheduling sleeps. use std::sync::Arc; use std::time::Duration; use std::time::Instant; +use codex_core::StartThreadOptions; use codex_core::TurnInputRequest; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::IsolatedSessionExtensions; +use codex_extension_api::SessionIsolation; use codex_extension_api::ToolContributor; use codex_features::Feature; +use codex_protocol::AgentPath; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::InterAgentCommunication; +use codex_protocol::protocol::InternalSessionSource; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::Op; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; use codex_protocol::user_input::UserInput; use codex_tools::JsonToolOutput; use codex_tools::ToolCall; @@ -34,6 +43,14 @@ enum Admission { Unsupported, } +#[derive(Clone, Copy)] +enum CaptureSource { + Root, + Guardian, + Subagent, + MemoryConsolidation, +} + #[derive(Clone)] struct GatedTool { admission: Admission, @@ -54,7 +71,7 @@ impl ToolContributor for GatedTool { impl<'call> ToolExecutor> for GatedTool { fn tool_name(&self) -> ToolName { ToolName::plain(if matches!(self.admission, Admission::Unsupported) { - "wait_agent" + "followup_task" } else { "timing_gate" }) @@ -92,14 +109,20 @@ impl<'call> ToolExecutor> for GatedTool { } } -#[test_case::test_case(Admission::Serial, false; "default_disabled")] -#[test_case::test_case(Admission::Unsupported, true; "unsupported_enabled")] -#[test_case::test_case(Admission::Serial, true; "serial_enabled")] -#[test_case::test_case(Admission::Parallel, true; "overlapping_enabled")] +#[test_case::test_case(Admission::Serial, false, CaptureSource::Root; "default_disabled")] +#[test_case::test_case(Admission::Unsupported, true, CaptureSource::Root; "unsupported_enabled")] +#[test_case::test_case(Admission::Serial, true, CaptureSource::Root; "serial_enabled")] +#[test_case::test_case(Admission::Parallel, true, CaptureSource::Root; "overlapping_enabled")] +#[test_case::test_case(Admission::Serial, true, CaptureSource::Guardian; "guardian_enabled")] +#[test_case::test_case(Admission::Serial, false, CaptureSource::Guardian; "guardian_disabled")] +#[test_case::test_case(Admission::Parallel, true, CaptureSource::Subagent; "subagent_enabled")] +#[test_case::test_case(Admission::Unsupported, true, CaptureSource::Subagent; "subagent_unsupported")] +#[test_case::test_case(Admission::Serial, true, CaptureSource::MemoryConsolidation; "memory_excluded")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn critical_path_follows_runtime_admission( admission: Admission, critical_path_timing: bool, + source: CaptureSource, ) -> anyhow::Result<()> { let server = responses::start_mock_server().await; let (entered, mut entries) = mpsc::unbounded_channel(); @@ -111,18 +134,60 @@ async fn critical_path_follows_runtime_admission( let name = gate.tool_name().name; let mut extensions = ExtensionRegistryBuilder::new(); extensions.tool_contributor(Arc::new(gate.clone())); + let extensions = Arc::new(extensions.build()); let test = test_codex() - .with_extensions(Arc::new(extensions.build())) + .with_extensions(extensions.clone()) .with_config(move |config| { if critical_path_timing { config .features .enable(Feature::CriticalPathTiming) .expect("critical-path capture is configurable"); + } else { + config + .features + .disable(Feature::CriticalPathTiming) + .expect("disable capture"); } }) .build_with_auto_env(&server) .await?; + let session_source = match source { + CaptureSource::Root => SessionSource::Exec, + CaptureSource::Guardian => SessionSource::Internal(InternalSessionSource::Guardian), + CaptureSource::Subagent => SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id: test.session_configured.thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }), + CaptureSource::MemoryConsolidation => { + SessionSource::Internal(InternalSessionSource::MemoryConsolidation) + } + }; + let mut options = StartThreadOptions { + session_source: Some(session_source), + environments: Some(vec![test.executor_environment().request()]), + ..StartThreadOptions::new(test.config.clone()) + }; + if matches!(source, CaptureSource::Guardian) { + // Guardian inherits neither host extensions nor an unrestricted tool policy. + options + .thread_extension_init + .insert(SessionIsolation::Isolated); + options + .thread_extension_init + .insert(IsolatedSessionExtensions(extensions)); + let mut policy = codex_guardian_reviewer::reviewer_tool_policy(); + policy + .allowed_tools + .as_mut() + .expect("Guardian tool allowlist") + .push(gate.tool_name()); + options.thread_extension_init.insert(policy); + } + let thread = test.thread_manager.start_thread(options).await?.thread; let first = responses::mount_sse_once( &server, responses::sse(vec![ @@ -134,14 +199,18 @@ async fn critical_path_follows_runtime_admission( ) .await; let start = Instant::now(); - test.codex + thread .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { text: "Run both tools".to_string(), text_elements: Vec::new(), }])) .await?; - let (first_id, first_start) = entries.recv().await.expect("first admitted tool"); - let observe = codex_core::test_support::observe_turn_timing(&test.codex) + let (first_id, first_start) = + tokio::time::timeout(Duration::from_secs(/* secs */ 20), entries.recv()) + .await + .expect("first tool should enter its handler") + .expect("first admitted tool"); + let observe = codex_core::test_support::observe_turn_timing(&thread) .await .expect("active turn"); let reply = responses::mount_sse_once( @@ -154,7 +223,11 @@ async fn critical_path_follows_runtime_admission( ) .await; let (second_start, held_start, held_end) = if matches!(admission, Admission::Parallel) { - let (second_id, second_start) = entries.recv().await.expect("second overlapping tool"); + let (second_id, second_start) = + tokio::time::timeout(Duration::from_secs(/* secs */ 20), entries.recv()) + .await + .expect("second tool should overlap the first") + .expect("second overlapping tool"); assert_ne!(first_id, second_id); // Both handlers are inside their gate before either can finish. let held_end = Instant::now(); @@ -167,15 +240,22 @@ async fn critical_path_follows_runtime_admission( ); let held_end = Instant::now(); gate.release.add_permits(/* n */ 1); - let (_, second_start) = entries.recv().await.expect("second serial tool"); + let (_, second_start) = + tokio::time::timeout(Duration::from_secs(/* secs */ 20), entries.recv()) + .await + .expect("second tool should enter after the first releases admission") + .expect("second serial tool"); gate.release.add_permits(/* n */ 1); (second_start, first_start, held_end) }; - wait_for_event(&test.codex, |event| { - matches!(event, EventMsg::TurnComplete(_)) - }) - .await; - first.single_request(); + wait_for_event(&thread, |event| matches!(event, EventMsg::TurnComplete(_))).await; + let first_request = first.single_request(); + if matches!(source, CaptureSource::Guardian) { + assert_eq!( + first_request.body_json()["client_metadata"]["x-openai-subagent"], + "guardian" + ); + } let request = reply.single_request(); for id in ["a", "b"] { assert!( @@ -186,7 +266,10 @@ async fn critical_path_follows_runtime_admission( ); } let end = Instant::now(); - if matches!(admission, Admission::Unsupported) || !critical_path_timing { + if matches!(admission, Admission::Unsupported) + || !critical_path_timing + || matches!(source, CaptureSource::MemoryConsolidation) + { assert_eq!(observe(start, end), None); } else { let [inference, review, tools, overhead, unknown] = @@ -249,3 +332,264 @@ async fn standalone_user_shell_omits_uncaptured_path() -> anyhow::Result<()> { assert_eq!(observe(now, now), None); Ok(()) } + +#[derive(Clone, Copy)] +enum AgentMessageTiming { + InitialOnly, + DuringTurn, +} + +#[test_case::test_case(true, AgentMessageTiming::InitialOnly; "initial_spawn_input")] +#[test_case::test_case(true, AgentMessageTiming::DuringTurn; "midturn_message")] +#[test_case::test_case(false, AgentMessageTiming::InitialOnly; "gate_disabled")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn spawned_child_has_independent_timing_and_parent_wait_is_tools( + critical_path_timing: bool, + message_timing: AgentMessageTiming, +) -> anyhow::Result<()> { + let server = responses::start_mock_server().await; + let (entered, mut entries) = mpsc::unbounded_channel(); + let gate = GatedTool { + admission: Admission::Serial, + entered, + release: Arc::new(Semaphore::new(/* permits */ 0)), + }; + let mut extensions = ExtensionRegistryBuilder::new(); + extensions.tool_contributor(Arc::new(gate.clone())); + let test = test_codex() + .with_extensions(Arc::new(extensions.build())) + .with_model_info_override("test-gpt-5.1-codex", |model| { + model.multi_agent_version = Some(MultiAgentVersion::V2); + }) + .with_config(move |config| { + for feature in [Feature::Collab, Feature::MultiAgentV2] { + config.features.enable(feature).expect("enable multi-agent"); + } + if critical_path_timing { + config + .features + .enable(Feature::CriticalPathTiming) + .expect("enable capture"); + } else { + config + .features + .disable(Feature::CriticalPathTiming) + .expect("disable capture"); + } + config + .features + .disable(Feature::EnableRequestCompression) + .expect("match uncompressed mock requests"); + }) + .build_with_auto_env(&server) + .await?; + let root_id = test.session_configured.thread_id; + let is_root = move |request: &wiremock::Request| { + request + .body_json::() + .is_ok_and(|body| body["client_metadata"]["thread_id"] == serde_json::json!(root_id)) + }; + let is_child = move |request: &wiremock::Request| { + request.body_json::().is_ok_and(|body| { + body["client_metadata"]["x-codex-parent-thread-id"] == serde_json::json!(root_id) + }) + }; + responses::mount_sse_once_match( + &server, + is_root, + responses::sse(vec![ + responses::ev_function_call_with_namespace( + "spawn-worker", + "collaboration", + "spawn_agent", + r#"{"task_name":"worker","fork_turns":"none","message":"Run the timing gate."}"#, + ), + responses::ev_completed("root-spawn"), + ]), + ) + .await; + responses::mount_sse_once_match( + &server, + is_root, + responses::sse(vec![ + responses::ev_function_call_with_namespace( + "wait-worker", + "collaboration", + "wait_agent", + "{}", + ), + responses::ev_completed("root-wait"), + ]), + ) + .await; + responses::mount_sse_once_match( + &server, + is_root, + responses::sse(vec![ + responses::ev_assistant_message("root-answer", "Worker started."), + responses::ev_completed("root-done"), + ]), + ) + .await; + let child_input = responses::mount_sse_once_match( + &server, + is_child, + responses::sse(vec![ + responses::ev_function_call("child-gate", "timing_gate", "{}"), + responses::ev_completed("child-tools"), + ]), + ) + .await; + let (sampled, mut sampling_windows) = mpsc::unbounded_channel(); + let child_reply = responses::sse(vec![ + responses::ev_assistant_message("child-answer", "Done."), + responses::ev_completed("child-done"), + ]); + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path_regex(".*/responses$")) + .and(is_child) + .respond_with(move |request: &wiremock::Request| { + // The child is sampling and the parent is waiting until this response is sent. + let start = Instant::now(); + let response = wiremock::ResponseTemplate::new(/* status_code */ 200) + .insert_header("content-type", "text/event-stream") + .set_body_string(child_reply.clone()); + let end = Instant::now(); + sampled + .send(( + start, + end, + request + .body_json::() + .expect("child request JSON"), + )) + .expect("sampling observer"); + response + }) + .up_to_n_times(/* n */ 1) + .mount(&server) + .await; + let mut created = test.thread_manager.subscribe_thread_created(); + let start = Instant::now(); + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Spawn a worker.".into(), + text_elements: Vec::new(), + }])) + .await?; + let child_id = tokio::time::timeout(Duration::from_secs(20), created.recv()).await??; + let child = test.thread_manager.get_thread(child_id).await?; + let (call_id, _) = tokio::time::timeout(Duration::from_secs(20), entries.recv()) + .await? + .expect("child tool should enter"); + assert_eq!(call_id, "child-gate"); + let observe = codex_core::test_support::observe_turn_timing(&child) + .await + .expect("active child turn"); + assert_eq!( + observe(start, Instant::now()).is_some(), + critical_path_timing + ); + let initial_messages = child_input.single_request().inputs_of_type("agent_message"); + assert_eq!(initial_messages.len(), 1); + assert_eq!(initial_messages[0]["author"], "/root"); + assert_eq!(initial_messages[0]["recipient"], "/root/worker"); + if matches!(message_timing, AgentMessageTiming::DuringTurn) { + child + .submit(Op::InterAgentCommunication { + communication: InterAgentCommunication::new( + AgentPath::root(), + AgentPath::root().join("worker").expect("child path"), + Vec::new(), + "Use this later result too.".into(), + /*trigger_turn*/ true, + ), + start_options: Default::default(), + }) + .await?; + // The response fences the submission queue before the tool can finish. + child.submit(Op::RealtimeConversationListVoices).await?; + wait_for_event(&child, |event| { + matches!(event, EventMsg::RealtimeConversationListVoicesResponse(_)) + }) + .await; + } + wait_for_event(&test.codex, |event| matches!(event, + EventMsg::ItemStarted(event) if matches!(&event.item, codex_protocol::items::TurnItem::CollabAgentToolCall(item) if item.id == "wait-worker") + )).await; + let observe_parent = codex_core::test_support::observe_turn_timing(&test.codex) + .await + .expect("waiting parent"); + gate.release.add_permits(/* n */ 1); + wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await; + let end = Instant::now(); + let expected_messages = match message_timing { + AgentMessageTiming::InitialOnly => 1, + AgentMessageTiming::DuringTurn => 2, + }; + let (sampling_start, sampling_end, child_request) = sampling_windows + .recv() + .await + .expect("child sampling window"); + assert_eq!( + child_request["input"] + .as_array() + .expect("request input") + .iter() + .filter(|item| item["type"] == "agent_message") + .count(), + expected_messages + ); + if critical_path_timing { + let [inference, review, tools, overhead, unknown] = + observe(start, end).expect("child's own turn has a complete path"); + assert!(inference > Duration::ZERO); + assert!(tools > Duration::ZERO); + assert_eq!((review, unknown), (Duration::ZERO, Duration::ZERO)); + assert_eq!( + inference + review + tools + overhead + unknown, + end.duration_since(start) + ); + } else { + assert_eq!(observe(start, end), None); + } + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let parent_end = Instant::now(); + let parent_path = observe_parent(start, parent_end); + assert_eq!(parent_path.is_some(), critical_path_timing); + if let Some([inference, review, tools, overhead, unknown]) = parent_path { + assert_eq!(unknown, Duration::ZERO); + assert_eq!( + inference + review + tools + overhead + unknown, + parent_end.duration_since(start) + ); + let sampling = sampling_end.duration_since(sampling_start); + assert!(sampling > Duration::ZERO); + assert_eq!( + observe(sampling_start, sampling_end), + Some([ + sampling, + Duration::ZERO, + Duration::ZERO, + Duration::ZERO, + Duration::ZERO + ]) + ); + assert_eq!( + observe_parent(sampling_start, sampling_end), + Some([ + Duration::ZERO, + Duration::ZERO, + sampling, + Duration::ZERO, + Duration::ZERO + ]) + ); + } + child.shutdown_and_wait().await?; + test.codex.shutdown_and_wait().await?; + Ok(()) +} diff --git a/codex-rs/direct-tool-timing/src/direct_tool_timing_tests.rs b/codex-rs/direct-tool-timing/src/direct_tool_timing_tests.rs index fb738af8c6..7a6398774b 100644 --- a/codex-rs/direct-tool-timing/src/direct_tool_timing_tests.rs +++ b/codex-rs/direct-tool-timing/src/direct_tool_timing_tests.rs @@ -8,7 +8,7 @@ fn parallel_winner_retains_its_review_and_shared_initial_path() { let now = Instant::now(); let start = now - Duration::from_secs(10); let shared_initial_path = - CriticalPath::default().work(start, start + Duration::from_secs(1), WorkKind::Inference); + CriticalPath::default().work(start, start + Duration::from_secs(1), WorkKind::Sampling); let batch = ToolBatchTiming::default(); let timings = DirectToolTimings::default(); let losing = timings.start_direct_call( @@ -38,7 +38,7 @@ fn parallel_winner_retains_its_review_and_shared_initial_path() { assert_eq!( completed.path.unwrap().durations(start, completed.ended_at), PathDurations { - inference: Duration::from_secs(1), + sampling: Duration::from_secs(1), auto_review: Duration::from_secs(3), tools: completed.ended_at.duration_since(start) - Duration::from_secs(4), ..PathDurations::default() diff --git a/codex-rs/direct-tool-timing/src/recorder/runtime_tests.rs b/codex-rs/direct-tool-timing/src/recorder/runtime_tests.rs index e4571318b9..b620fc9b88 100644 --- a/codex-rs/direct-tool-timing/src/recorder/runtime_tests.rs +++ b/codex-rs/direct-tool-timing/src/recorder/runtime_tests.rs @@ -21,7 +21,7 @@ use tokio_util::either::Either; async fn lock_handoff_publishes_review_history_before_admitting_waiter(admission: ToolAdmission) { let start = Instant::now() - Duration::from_secs(10); let tool_start = start + Duration::from_secs(1); - let shared_initial_path = CriticalPath::default().work(start, tool_start, WorkKind::Inference); + let shared_initial_path = CriticalPath::default().work(start, tool_start, WorkKind::Sampling); let timings = DirectToolTimings::default(); let batch = ToolBatchTiming::default(); let gate = ToolGateTiming::default(); @@ -89,7 +89,7 @@ async fn lock_handoff_publishes_review_history_before_admitting_waiter(admission assert_eq!( completed.path.unwrap().durations(start, completed.ended_at), PathDurations { - inference: Duration::from_secs(1), + sampling: Duration::from_secs(1), auto_review: Duration::from_secs(3), tools: completed.ended_at.duration_since(start) - Duration::from_secs(4), ..PathDurations::default() diff --git a/codex-rs/utils/critical-path/src/critical_path_tests.rs b/codex-rs/utils/critical-path/src/critical_path_tests.rs index 937d5bba8e..6cb8020a09 100644 --- a/codex-rs/utils/critical-path/src/critical_path_tests.rs +++ b/codex-rs/utils/critical-path/src/critical_path_tests.rs @@ -10,7 +10,7 @@ fn join_replaces_overlapping_work_with_the_full_winning_branch() { let shared_initial_path = CriticalPath::default().work( time_after_turn_start(0), time_after_turn_start(100), - WorkKind::Inference, + WorkKind::Sampling, ); let reviewed_tool_path = shared_initial_path .clone() @@ -33,7 +33,7 @@ fn join_replaces_overlapping_work_with_the_full_winning_branch() { .work( time_after_turn_start(100), time_after_turn_start(200), - WorkKind::Inference, + WorkKind::Sampling, ) .join( time_after_turn_start(200), @@ -43,7 +43,7 @@ fn join_replaces_overlapping_work_with_the_full_winning_branch() { assert_eq!( critical_path.durations(time_after_turn_start(0), time_after_turn_start(700)), PathDurations { - inference: Duration::from_millis(100), + sampling: Duration::from_millis(100), tools: Duration::from_millis(600), ..PathDurations::default() } @@ -57,7 +57,7 @@ fn nested_serial_dependencies_preserve_exclusive_categories() { let shared_initial_path = CriticalPath::default().work( time_after_turn_start(0), time_after_turn_start(100), - WorkKind::Inference, + WorkKind::Sampling, ); let child_path = shared_initial_path .clone() @@ -74,13 +74,13 @@ fn nested_serial_dependencies_preserve_exclusive_categories() { .work( time_after_turn_start(400), time_after_turn_start(500), - WorkKind::Inference, + WorkKind::Compaction, ); let critical_path = shared_initial_path .work( time_after_turn_start(100), time_after_turn_start(200), - WorkKind::Inference, + WorkKind::Sampling, ) .join( time_after_turn_start(200), @@ -90,7 +90,8 @@ fn nested_serial_dependencies_preserve_exclusive_categories() { assert_eq!( critical_path.durations(time_after_turn_start(0), time_after_turn_start(600)), PathDurations { - inference: Duration::from_millis(200), + sampling: Duration::from_millis(100), + compaction: Duration::from_millis(100), auto_review: Duration::from_millis(200), tools: Duration::from_millis(200), ..PathDurations::default() @@ -104,7 +105,7 @@ fn equal_finish_times_select_one_path_without_double_counting() { let turn_end = turn_start + Duration::from_millis(100); let reviewed_tool_path = CriticalPath::default().work(turn_start, turn_end, WorkKind::AutoReview); - let inference_path = CriticalPath::default().work(turn_start, turn_end, WorkKind::Inference); + let inference_path = CriticalPath::default().work(turn_start, turn_end, WorkKind::Sampling); // A tie has multiple valid paths. Stable left selection does not sum both. assert_eq!( reviewed_tool_path @@ -123,7 +124,7 @@ fn path_retention_exhaustion_marks_unknown_instead_of_claiming_zero() { critical_path = critical_path.work( turn_start + Duration::from_micros(segment_index as u64), turn_start + Duration::from_micros(segment_index as u64 + 1), - WorkKind::Inference, + WorkKind::Sampling, ); } let turn_end = turn_start + Duration::from_micros(MAX_PATH_DEPTH as u64 + 1); @@ -143,7 +144,7 @@ fn reviewed_branch_wins_even_when_some_review_overlapped_parent_inference() { let shared_initial_path = CriticalPath::default().work( time_after_turn_start(0), time_after_turn_start(100), - WorkKind::Inference, + WorkKind::Sampling, ); let reviewed_tool_path = shared_initial_path .clone() @@ -166,7 +167,7 @@ fn reviewed_branch_wins_even_when_some_review_overlapped_parent_inference() { .work( time_after_turn_start(100), time_after_turn_start(200), - WorkKind::Inference, + WorkKind::Sampling, ) .join( time_after_turn_start(200), @@ -176,7 +177,7 @@ fn reviewed_branch_wins_even_when_some_review_overlapped_parent_inference() { assert_eq!( critical_path.durations(time_after_turn_start(0), time_after_turn_start(400)), PathDurations { - inference: Duration::from_millis(100), + sampling: Duration::from_millis(100), auto_review: Duration::from_millis(200), tools: Duration::from_millis(100), ..Default::default() @@ -192,7 +193,7 @@ fn durations_clip_work_to_the_window_and_attribute_gaps_to_overhead() { .work( time_after_turn_start(10), time_after_turn_start(30), - WorkKind::Inference, + WorkKind::Sampling, ) .work( time_after_turn_start(50), @@ -203,7 +204,7 @@ fn durations_clip_work_to_the_window_and_attribute_gaps_to_overhead() { assert_eq!( critical_path.durations(time_after_turn_start(0), time_after_turn_start(100)), PathDurations { - inference: Duration::from_millis(20), + sampling: Duration::from_millis(20), auto_review: Duration::from_millis(40), overhead: Duration::from_millis(40), ..PathDurations::default() @@ -212,7 +213,7 @@ fn durations_clip_work_to_the_window_and_attribute_gaps_to_overhead() { assert_eq!( critical_path.durations(time_after_turn_start(20), time_after_turn_start(80)), PathDurations { - inference: Duration::from_millis(10), + sampling: Duration::from_millis(10), auto_review: Duration::from_millis(30), overhead: Duration::from_millis(20), ..PathDurations::default() @@ -228,7 +229,7 @@ fn overlapping_work_only_attributes_time_after_the_existing_path() { .work( time_after_turn_start(0), time_after_turn_start(10), - WorkKind::Inference, + WorkKind::Sampling, ) .work( time_after_turn_start(5), @@ -239,7 +240,7 @@ fn overlapping_work_only_attributes_time_after_the_existing_path() { assert_eq!( critical_path.durations(time_after_turn_start(0), time_after_turn_start(20)), PathDurations { - inference: Duration::from_millis(10), + sampling: Duration::from_millis(10), tools: Duration::from_millis(10), ..PathDurations::default() } @@ -274,7 +275,7 @@ fn long_path_debug_output_is_bounded_on_a_small_stack() { path = path.work( start + Duration::from_micros(index), start + Duration::from_micros(index + 1), - WorkKind::Inference, + WorkKind::Sampling, ); } assert!(format!("{path:?}").len() < 512); @@ -295,7 +296,7 @@ fn retained_prefixes_can_be_released_concurrently_on_small_stacks() { path = path.work( start + Duration::from_micros(index as u64), start + Duration::from_micros(index as u64 + 1), - WorkKind::Inference, + WorkKind::Sampling, ); prefixes[index % 2].push(path.clone()); } @@ -328,7 +329,7 @@ fn human_wait_is_retained_only_on_the_selected_parallel_path() { let start = Instant::now(); let time_after_turn_start = |milliseconds| start + Duration::from_millis(milliseconds); let prefix = - CriticalPath::default().work(start, time_after_turn_start(100), WorkKind::Inference); + CriticalPath::default().work(start, time_after_turn_start(100), WorkKind::Sampling); let reviewed_tool = prefix .clone() .work( @@ -359,7 +360,7 @@ fn human_wait_is_retained_only_on_the_selected_parallel_path() { assert_eq!( path.durations(start, time_after_turn_start(600)), PathDurations { - inference: Duration::from_millis(100), + sampling: Duration::from_millis(100), auto_review: Duration::from_millis(100), user_wait: Duration::from_millis(300), tools: Duration::from_millis(100), diff --git a/codex-rs/utils/critical-path/src/lib.rs b/codex-rs/utils/critical-path/src/lib.rs index 439d008120..b8b18d2d7b 100644 --- a/codex-rs/utils/critical-path/src/lib.rs +++ b/codex-rs/utils/critical-path/src/lib.rs @@ -10,8 +10,10 @@ const MAX_PATH_DEPTH: usize = 16_384; /// Category attributed to an observed segment of dependency work. #[derive(Clone, Copy, Debug)] pub enum WorkKind { - /// Model inference work on the selected dependency path. - Inference, + /// Model sampling work on the selected dependency path. + Sampling, + /// Compaction work, including its local bookkeeping. + Compaction, /// Automatic review work, such as reviewing a tool call before execution. AutoReview, /// Time awaiting a human answer or approval on the selected dependency path. @@ -66,7 +68,8 @@ struct Node { /// Exclusive time attributed to each category within the requested interval. #[derive(Debug, Default, PartialEq, Eq)] pub struct PathDurations { - pub inference: Duration, + pub sampling: Duration, + pub compaction: Duration, pub auto_review: Duration, pub user_wait: Duration, pub tools: Duration, @@ -77,7 +80,8 @@ pub struct PathDurations { impl PathDurations { fn record(&mut self, kind: WorkKind, duration: Duration) { let total = match kind { - WorkKind::Inference => &mut self.inference, + WorkKind::Sampling => &mut self.sampling, + WorkKind::Compaction => &mut self.compaction, WorkKind::AutoReview => &mut self.auto_review, WorkKind::UserWait => &mut self.user_wait, WorkKind::Tools => &mut self.tools,