From 9eeac78b3f0c510ed9c53d329d466cf18e1af8af Mon Sep 17 00:00:00 2001 From: rka-oai Date: Thu, 30 Jul 2026 16:39:01 +0000 Subject: [PATCH] Refresh environment dates from the configured clock (#36187) ## Why The `` in environment context could diverge from the clock used by current-time features, especially when a host provides an external clock. ## What changed - Read the current time from the session's time provider whenever world state is built, and render its local date in environment context. - Use each session's own thread ID for clock reads, including guardian sessions. - Propagate clock-read failures through turn error events and stop before model inference. ## Testing - Cover date updates across turns and app-server requests. - Verify guardian clock reads use the corresponding root or guardian thread ID and do not run during prewarm. - Verify clock failures emit an error without sending an inference request. GitOrigin-RevId: bb7f32ccc0229b7bc6f14d7356db00bcce78066d --- .../app-server/tests/suite/v2/current_time.rs | 69 ++++++++----- codex-rs/app-server/tests/suite/v2/sleep.rs | 1 + codex-rs/core/src/codex_thread.rs | 2 +- codex-rs/core/src/compact_tests.rs | 7 +- codex-rs/core/src/compact_token_budget.rs | 4 +- .../src/context/world_state/environment.rs | 3 +- codex-rs/core/src/prompt_debug.rs | 2 +- codex-rs/core/src/session/mod.rs | 16 +-- .../session/rollout_reconstruction_tests.rs | 3 +- codex-rs/core/src/session/tests.rs | 49 ++++++--- .../core/src/session/tests/guardian_tests.rs | 7 +- codex-rs/core/src/session/turn.rs | 5 +- codex-rs/core/src/session/world_state.rs | 16 ++- codex-rs/core/src/tasks/mod.rs | 11 +++ .../core/tests/suite/current_time_reminder.rs | 57 ++++++++++- codex-rs/core/tests/suite/guardian_review.rs | 99 +++++++++++++++++-- 16 files changed, 283 insertions(+), 68 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/current_time.rs b/codex-rs/app-server/tests/suite/v2/current_time.rs index d3a012f0ea..9f8ae49d46 100644 --- a/codex-rs/app-server/tests/suite/v2/current_time.rs +++ b/codex-rs/app-server/tests/suite/v2/current_time.rs @@ -2,8 +2,12 @@ use anyhow::Result; use app_test_support::MockResponsesConfig; use app_test_support::TestAppServer; use app_test_support::create_final_assistant_message_sse_response; +use chrono::DateTime; +use chrono::Local; +use chrono::Utc; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::CurrentTimeReadResponse; +use codex_app_server_protocol::JSONRPCMessage; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; @@ -68,35 +72,54 @@ clock_source = "external" }) .await?; - let server_request = timeout( - DEFAULT_READ_TIMEOUT, - app_server.read_stream_until_request_message(), - ) - .await??; - let ServerRequest::CurrentTimeRead { request_id, params } = server_request else { - panic!("expected CurrentTimeRead request, got: {server_request:?}"); - }; - assert_eq!(params.thread_id, thread.id); - app_server - .send_response( - request_id, - serde_json::to_value(CurrentTimeReadResponse { - current_time_at: CURRENT_TIME_AT, - })?, - ) - .await?; - timeout( - DEFAULT_READ_TIMEOUT, - app_server.read_stream_until_notification_message("turn/completed"), - ) + let mut current_time_reads = 0; + timeout(DEFAULT_READ_TIMEOUT, async { + loop { + match app_server.read_next_message().await? { + JSONRPCMessage::Request(request) => { + let server_request = ServerRequest::try_from(request)?; + let ServerRequest::CurrentTimeRead { request_id, params } = server_request + else { + panic!("expected CurrentTimeRead request, got: {server_request:?}"); + }; + assert_eq!(params.thread_id, thread.id); + current_time_reads += 1; + app_server + .send_response( + request_id, + serde_json::to_value(CurrentTimeReadResponse { + current_time_at: CURRENT_TIME_AT, + })?, + ) + .await?; + } + JSONRPCMessage::Notification(notification) + if notification.method == "turn/completed" => + { + break Ok::<_, anyhow::Error>(()); + } + _ => {} + } + } + }) .await??; + assert!(current_time_reads >= 2); + let request = response_mock.single_request(); assert!( - response_mock - .single_request() + request .message_input_texts("developer") .iter() .any(|text| text == CURRENT_TIME_REMINDER) ); + let current_date = DateTime::::from_timestamp(CURRENT_TIME_AT, 0) + .expect("test timestamp should be valid") + .with_timezone(&Local) + .format("%Y-%m-%d") + .to_string(); + assert!(request.message_input_texts("user").iter().any(|text| { + text.contains("") + && text.contains(&format!("{current_date}")) + })); Ok(()) } diff --git a/codex-rs/app-server/tests/suite/v2/sleep.rs b/codex-rs/app-server/tests/suite/v2/sleep.rs index a886125645..b0b4a939f2 100644 --- a/codex-rs/app-server/tests/suite/v2/sleep.rs +++ b/codex-rs/app-server/tests/suite/v2/sleep.rs @@ -54,6 +54,7 @@ async fn external_sleep_polls_current_time_and_emits_items() -> Result<()> { let codex_home = TempDir::new()?; MockResponsesConfig::new(&server.uri()) + .with_root_config("include_environment_context = false") .with_extra_config( r#"[features.current_time_reminder] enabled = true diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 63f2c69aff..074719114d 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -496,7 +496,7 @@ impl CodexThread { .await?; self.session .record_context_updates_and_set_reference_context_item(step_context.as_ref()) - .await; + .await?; } self.session .inject_no_new_turn(items, Some(turn_context.as_ref())) diff --git a/codex-rs/core/src/compact_tests.rs b/codex-rs/core/src/compact_tests.rs index 6fcec9bb27..ac7cc87bee 100644 --- a/codex-rs/core/src/compact_tests.rs +++ b/codex-rs/core/src/compact_tests.rs @@ -18,7 +18,12 @@ async fn process_compacted_history_with_test_session( .await; let step_context = crate::session::step_context::StepContext::for_test(Arc::clone(&turn_context)); - let world_state = Arc::new(session.build_world_state_for_step(&step_context).await); + let world_state = Arc::new( + session + .build_world_state_for_step(&step_context) + .await + .expect("world state should build"), + ); let initial_context = session .build_initial_context_with_world_state(&turn_context, world_state.as_ref()) .await; diff --git a/codex-rs/core/src/compact_token_budget.rs b/codex-rs/core/src/compact_token_budget.rs index 1e47496638..2abb6b0b24 100644 --- a/codex-rs/core/src/compact_token_budget.rs +++ b/codex-rs/core/src/compact_token_budget.rs @@ -40,7 +40,7 @@ pub(crate) async fn run_manual_compact_task( let step_context = sess .capture_step_context(Arc::clone(&turn_context), &CancellationToken::new()) .await?; - let world_state = Arc::new(sess.build_world_state_for_step(&step_context).await); + let world_state = Arc::new(sess.build_world_state_for_step(&step_context).await?); run_compact_task_inner(&sess, &step_context, world_state, CompactionTrigger::Manual).await } @@ -57,7 +57,7 @@ pub(crate) async fn run_inline_auto_compact_task( let world_state = match initial_context_injection { InitialContextInjection::BeforeLastUserMessage { world_state, .. } => world_state, InitialContextInjection::DoNotInject => { - Arc::new(sess.build_world_state_for_step(&step_context).await) + Arc::new(sess.build_world_state_for_step(&step_context).await?) } }; run_compact_task_inner(&sess, &step_context, world_state, CompactionTrigger::Auto).await diff --git a/codex-rs/core/src/context/world_state/environment.rs b/codex-rs/core/src/context/world_state/environment.rs index 735f864a6c..c039544ca0 100644 --- a/codex-rs/core/src/context/world_state/environment.rs +++ b/codex-rs/core/src/context/world_state/environment.rs @@ -27,6 +27,7 @@ impl EnvironmentsState { pub(crate) fn from_turn_context_with_environments( turn_context: &TurnContext, environments: &TurnEnvironmentSnapshot, + current_date: Option, ) -> Self { let workspace_roots = environments .primary() @@ -34,7 +35,7 @@ impl EnvironmentsState { .unwrap_or_default(); Self { environments: environment_states(environments), - current_date: turn_context.current_date.clone(), + current_date, timezone: turn_context.timezone.clone(), network: network_from_turn_context(turn_context), filesystem: Some(FileSystemContext::from_permission_profile( diff --git a/codex-rs/core/src/prompt_debug.rs b/codex-rs/core/src/prompt_debug.rs index 3fbaf1d683..b97a5a2c7c 100644 --- a/codex-rs/core/src/prompt_debug.rs +++ b/codex-rs/core/src/prompt_debug.rs @@ -88,7 +88,7 @@ pub(crate) async fn build_prompt_input_from_session( .capture_step_context(Arc::clone(&turn_context), &CancellationToken::new()) .await?; sess.record_context_updates_and_set_reference_context_item(step_context.as_ref()) - .await; + .await?; if !input.is_empty() { let response_item = sess.response_item_from_user_input(input); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 1e81eecdbd..0c7b438464 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -2992,10 +2992,10 @@ impl Session { &self, previous_world_state: &Arc, step_context: &step_context::StepContext, - ) -> Arc { + ) -> CodexResult> { let turn_context = step_context.turn.as_ref(); // Render model-visible state from the same step used to build and run tools. - let world_state = Arc::new(self.build_world_state_for_step(step_context).await); + let world_state = Arc::new(self.build_world_state_for_step(step_context).await?); // Derive the model update and persisted patch from the same two snapshots. let previous_snapshot = previous_world_state.snapshot(); let world_state_snapshot = world_state.snapshot(); @@ -3020,7 +3020,7 @@ impl Session { self.persist_rollout_items(&[RolloutItem::WorldState(world_state_item)]) .await; } - world_state + Ok(world_state) } /// Captures one request-scoped view of dynamic state. @@ -3711,7 +3711,7 @@ impl Session { pub(crate) async fn record_context_updates_and_set_reference_context_item( &self, step_context: &StepContext, - ) -> Arc { + ) -> CodexResult> { let turn_context = step_context.turn.as_ref(); let reference_context_item = { let state = self.state.lock().await; @@ -3720,7 +3720,7 @@ impl Session { let turn_context_item = turn_context.to_turn_context_item(); let turn_context_changed = reference_context_item.as_ref() != Some(&turn_context_item); let should_inject_full_context = reference_context_item.is_none(); - let world_state = Arc::new(self.build_world_state_for_step(step_context).await); + let world_state = Arc::new(self.build_world_state_for_step(step_context).await?); // Full initial context resets the baseline; later turns persist only its changes. let (mut context_items, world_state_item) = if should_inject_full_context { let context_items = self @@ -3757,7 +3757,7 @@ impl Session { // A snapshot can change without producing model-visible or TurnContext updates. let only_world_state_changed = !turn_context_changed && context_items.is_empty(); if only_world_state_changed && world_state_item.is_none() { - return world_state; + return Ok(world_state); } if !context_items.is_empty() { self.record_conversation_items(turn_context, &context_items) @@ -3770,7 +3770,7 @@ impl Session { } // A snapshot-only change does not require a duplicate TurnContext record. if only_world_state_changed { - return world_state; + return Ok(world_state); } // Persist one `TurnContextItem` per real user turn so resume/lazy replay can recover the // latest durable baseline even when this turn emitted no model-visible context diffs. @@ -3781,7 +3781,7 @@ impl Session { // context items. let mut state = self.state.lock().await; state.set_reference_context_item(Some(turn_context_item)); - world_state + Ok(world_state) } pub(crate) async fn update_token_usage_info( diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index f5bf358d5a..952d4714e6 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -165,7 +165,8 @@ async fn record_initial_history_restores_world_state_baseline() { let step_context = StepContext::for_test(Arc::clone(&turn_context)); session .record_context_updates_and_set_reference_context_item(&step_context) - .await; + .await + .expect("world state should build"); assert_eq!( session.clone_history().await.raw_items(), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index d0378c4bf1..a71c37fa2f 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -2575,7 +2575,8 @@ async fn resumed_history_injects_initial_context_on_first_context_update_only() let step_context = StepContext::for_test(Arc::clone(&turn_context)); session .record_context_updates_and_set_reference_context_item(&step_context) - .await; + .await + .expect("world state should build"); let initial_context = build_initial_context(&session, &turn_context).await; expected.extend(initial_context); let history_after_seed = session.clone_history().await; @@ -2586,7 +2587,8 @@ async fn resumed_history_injects_initial_context_on_first_context_update_only() session .record_context_updates_and_set_reference_context_item(&step_context) - .await; + .await + .expect("world state should build"); let history_after_second_seed = session.clone_history().await; assert_eq!( history_after_seed.raw_items(), @@ -3208,7 +3210,12 @@ async fn start_new_context_window_assigns_and_persists_item_ids() { .capture_step_context(Arc::clone(&turn_context), &CancellationToken::new()) .await .expect("a fresh cancellation token cannot be cancelled"); - let world_state = Arc::new(session.build_world_state_for_step(&step_context).await); + let world_state = Arc::new( + session + .build_world_state_for_step(&step_context) + .await + .expect("world state should build"), + ); session .start_new_context_window(&step_context, world_state) @@ -5616,7 +5623,10 @@ pub(crate) async fn build_world_state_from_turn_context( turn_context: &Arc, ) -> WorldState { let step_context = StepContext::for_test(Arc::clone(turn_context)); - session.build_world_state_for_step(&step_context).await + session + .build_world_state_for_step(&step_context) + .await + .expect("world state should build") } // todo: use online model info @@ -8716,7 +8726,6 @@ async fn record_context_updates_emits_environment_item_for_time_changes() { &session.services.models_manager, ) .await; - current_context.current_date = Some("2026-02-27".to_string()); current_context.timezone = Some("Europe/Berlin".to_string()); let update_items = @@ -8726,7 +8735,8 @@ async fn record_context_updates_emits_environment_item_for_time_changes() { .into_iter() .find(|text| text.contains("")) .expect("environment update item should be emitted"); - assert!(environment_update.contains("2026-02-27")); + let current_date = chrono::Local::now().format("%Y-%m-%d").to_string(); + assert!(environment_update.contains(&format!("{current_date}"))); assert!(environment_update.contains("Europe/Berlin")); } @@ -8777,13 +8787,15 @@ async fn record_context_update_items( let previous_step = StepContext::for_test(previous_context); session .record_context_updates_and_set_reference_context_item(&previous_step) - .await; + .await + .expect("world state should build"); let previous_len = session.clone_history().await.raw_items().len(); let current_step = StepContext::for_test(Arc::new(current_context)); session .record_context_updates_and_set_reference_context_item(¤t_step) - .await; + .await + .expect("world state should build"); let history = session.clone_history().await; history.raw_items()[previous_len..].to_vec() } @@ -9009,7 +9021,8 @@ async fn record_context_updates_includes_turn_context_fragments_on_steady_state_ let step_context = StepContext::for_test(Arc::clone(&turn_context)); session .record_context_updates_and_set_reference_context_item(&step_context) - .await; + .await + .expect("world state should build"); let history = session.clone_history().await; let developer_messages = developer_message_texts(history.raw_items()); @@ -9467,7 +9480,8 @@ async fn record_context_updates_and_set_reference_context_item_injects_full_cont let step_context = StepContext::for_test(Arc::clone(&turn_context)); session .record_context_updates_and_set_reference_context_item(&step_context) - .await; + .await + .expect("world state should build"); let history = session.clone_history().await; let initial_context = build_initial_context(&session, &turn_context).await; assert_eq!( @@ -9503,7 +9517,8 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co .await; session .record_context_updates_and_set_reference_context_item(&step_context) - .await; + .await + .expect("world state should build"); { let mut state = session.state.lock().await; state.set_reference_context_item(/*item*/ None); @@ -9517,7 +9532,8 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co session .record_context_updates_and_set_reference_context_item(&step_context) - .await; + .await + .expect("world state should build"); let history = session.clone_history().await; let mut expected_history = vec![compacted_summary]; @@ -9562,7 +9578,8 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline let step_context = StepContext::for_test(Arc::clone(&turn_context)); session .record_context_updates_and_set_reference_context_item(&step_context) - .await; + .await + .expect("world state should build"); assert_eq!( session.clone_history().await.raw_items().to_vec(), @@ -9611,7 +9628,8 @@ async fn record_context_updates_and_set_reference_context_item_persists_split_fi let step_context = StepContext::for_test(Arc::clone(&turn_context)); session .record_context_updates_and_set_reference_context_item(&step_context) - .await; + .await + .expect("world state should build"); session.ensure_rollout_materialized().await; session.flush_rollout().await.expect("rollout should flush"); @@ -9698,7 +9716,8 @@ async fn record_context_updates_and_set_reference_context_item_persists_full_rei let step_context = StepContext::for_test(Arc::clone(&turn_context)); session .record_context_updates_and_set_reference_context_item(&step_context) - .await; + .await + .expect("world state should build"); session.ensure_rollout_materialized().await; session.flush_rollout().await.expect("rollout should flush"); diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index 15ac87b7d1..5e84027b43 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -540,7 +540,12 @@ async fn process_compacted_history_preserves_separate_guardian_developer_message turn_context.developer_instructions = Some(guardian_policy.clone()); let turn_context = Arc::new(turn_context); let step_context = StepContext::for_test(Arc::clone(&turn_context)); - let world_state = Arc::new(session.build_world_state_for_step(&step_context).await); + let world_state = Arc::new( + session + .build_world_state_for_step(&step_context) + .await + .expect("world state should build"), + ); let initial_context_injection = InitialContextInjection::BeforeLastUserMessage { world_state, step_context, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 2a51f892e3..09358b82a8 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -210,10 +210,11 @@ pub(crate) async fn run_turn( Err(err) => return Err(err), }; // Keep the exact model-visible state used by this turn and its inline compactions. - let (mut world_state, display_roots) = tokio::join!( + let (world_state, display_roots) = tokio::join!( sess.record_context_updates_and_set_reference_context_item(first_step_context.as_ref()), turn_diff_display_roots(first_step_context.as_ref()), ); + let mut world_state = world_state?; let Some((injection_items, explicitly_enabled_connectors)) = build_skills_and_plugins( &sess, @@ -325,7 +326,7 @@ pub(crate) async fn run_turn( world_state = sess .record_step_world_state_if_changed(&world_state, step_context.as_ref()) - .await; + .await?; // Construct the input that we will send to the model. let sampling_request_input: Vec = async { diff --git a/codex-rs/core/src/session/world_state.rs b/codex-rs/core/src/session/world_state.rs index 7663beb7c9..f60ef8398e 100644 --- a/codex-rs/core/src/session/world_state.rs +++ b/codex-rs/core/src/session/world_state.rs @@ -20,13 +20,15 @@ use crate::context::world_state::ToolsState; use crate::context::world_state::WorldState; use codex_extension_api::WorldStateContributionInput; use codex_features::Feature; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result as CodexResult; impl Session { #[tracing::instrument(name = "world_state.build", level = "info", skip_all)] pub(crate) async fn build_world_state_for_step( &self, step_context: &StepContext, - ) -> WorldState { + ) -> CodexResult { let turn_context = step_context.turn.as_ref(); tracing::trace!( selected_capability_root_count = step_context.selected_capability_roots.len(), @@ -134,10 +136,20 @@ impl Session { world_state.add_section(collaboration_mode); } if turn_context.config.include_environment_context { + let current_date = self + .services + .time_provider + .current_time(self.thread_id()) + .await + .map_err(|err| CodexErr::Fatal(format!("failed to read current time: {err:#}")))? + .with_timezone(&chrono::Local) + .format("%Y-%m-%d") + .to_string(); world_state.add_section( EnvironmentsState::from_turn_context_with_environments( turn_context, &step_context.environments, + Some(current_date), ) .with_subagents(environment_subagents), ); @@ -205,6 +217,6 @@ impl Session { world_state.add_section(MultiAgentModeState::new( super::multi_agents::effective_multi_agent_mode(turn_context), )); - world_state + Ok(world_state) } } diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index e12e2bfbe1..83ca4512a4 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -607,6 +607,17 @@ impl Session { } Err(err) => { warn!(%err, "session task returned an unexpected error"); + self.emit_turn_error_lifecycle( + turn_context.as_ref(), + err.to_codex_protocol_error(), + ) + .await; + self.track_turn_codex_error(turn_context.as_ref(), &err); + self.send_event( + turn_context.as_ref(), + EventMsg::Error(err.to_error_event(/*message_prefix*/ None)), + ) + .await; (None, None) } }; diff --git a/codex-rs/core/tests/suite/current_time_reminder.rs b/codex-rs/core/tests/suite/current_time_reminder.rs index 9a9440fb33..4e1da82e5f 100644 --- a/codex-rs/core/tests/suite/current_time_reminder.rs +++ b/codex-rs/core/tests/suite/current_time_reminder.rs @@ -7,6 +7,7 @@ use std::time::Duration; use anyhow::Result; use anyhow::anyhow; use chrono::DateTime; +use chrono::Local; use chrono::Utc; use codex_core::SleepFuture; use codex_core::TimeFuture; @@ -100,6 +101,7 @@ fn enable_current_time_reminder( interval: u64, clock_source: CurrentTimeSource, ) { + config.include_environment_context = false; config .features .enable(Feature::CurrentTimeReminder) @@ -111,6 +113,56 @@ fn enable_current_time_reminder( }); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn environment_context_uses_external_current_time_on_each_turn() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let responses = mount_sse_sequence( + &server, + vec![ + sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), + sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]), + ], + ) + .await; + let time_provider = Arc::new(TestTimeProvider::default()); + let test = test_codex() + .with_config(|config| { + enable_current_time_reminder(config, /*interval*/ 0, CurrentTimeSource::External); + config.include_environment_context = true; + }) + .with_external_time_provider(time_provider.clone()) + .build_with_auto_env(&server) + .await?; + + test.submit_turn("first simulated day").await?; + time_provider + .current_time + .store(FIRST_TIME_UNIX_SECONDS + 86_400, Ordering::Relaxed); + test.submit_turn("second simulated day").await?; + + let requests = responses.requests(); + assert_eq!(requests.len(), 2); + for (request, timestamp) in requests + .iter() + .zip([FIRST_TIME_UNIX_SECONDS, FIRST_TIME_UNIX_SECONDS + 86_400]) + { + let current_date = DateTime::::from_timestamp(timestamp, 0) + .expect("test timestamp should be valid") + .with_timezone(&Local) + .format("%Y-%m-%d") + .to_string(); + assert!(request.message_input_texts("user").iter().any(|text| { + text.contains("") + && text.contains(&format!("{current_date}")) + })); + } + assert_eq!(current_time_reminders(&requests[0]), vec![SECOND_REMINDER]); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn current_time_reminders_follow_time_interval_and_persist_in_history() -> Result<()> { skip_if_no_network!(Ok(())); @@ -368,10 +420,11 @@ async fn time_provider_failure_stops_before_inference() -> Result<()> { .await; let test = test_codex() .with_config(|config| { - enable_current_time_reminder(config, /*interval*/ 1, CurrentTimeSource::External) + enable_current_time_reminder(config, /*interval*/ 1, CurrentTimeSource::External); + config.include_environment_context = true; }) .with_external_time_provider(Arc::new(FailingTimeProvider)) - .build(&server) + .build_with_auto_env(&server) .await?; test.codex diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index de64275c45..aaf736ab8e 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -1,8 +1,18 @@ #![cfg(not(target_os = "windows"))] use anyhow::Result; +use chrono::DateTime; +use chrono::Local; +use chrono::Utc; +use codex_core::SleepFuture; +use codex_core::TimeFuture; +use codex_core::TimeProvider; use codex_core::config::Constrained; +use codex_core::config::CurrentTimeReminderConfig; use codex_core::sandboxing::SandboxPermissions; +use codex_features::CurrentTimeSource; +use codex_features::Feature; +use codex_protocol::ThreadId; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::openai_models::AutoReviewMessages; use codex_protocol::openai_models::ModelsResponse; @@ -33,9 +43,34 @@ use serde_json::Value; use serde_json::json; use std::fs; use std::os::unix::fs::PermissionsExt; +use std::sync::Arc; +use std::sync::Mutex; use std::time::Duration; use tempfile::TempDir; +const CURRENT_TIME_AT: i64 = 1_781_717_655; + +struct RecordingTimeProvider { + thread_ids: Mutex>, +} + +impl TimeProvider for RecordingTimeProvider { + fn current_time(&self, thread_id: ThreadId) -> TimeFuture<'_> { + self.thread_ids + .lock() + .expect("time-provider thread ids lock should not be poisoned") + .push(thread_id); + Box::pin(async { + Ok(DateTime::::from_timestamp(CURRENT_TIME_AT, 0) + .expect("test timestamp should be valid")) + }) + } + + fn sleep(&self, _thread_id: ThreadId, _duration: Duration) -> SleepFuture<'_> { + Box::pin(async { Ok(()) }) + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn guardian_session_prewarms_and_is_reused_for_first_review() -> Result<()> { skip_if_no_network!(Ok(())); @@ -85,17 +120,31 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review() -> Result<() ]], ]) .await; - let mut builder = test_codex().with_config(move |config| { - config.model_catalog = Some(ModelsResponse { - models: vec![review_model], - }); - config.model_context_window = Some(900_000); - config.model_auto_compact_token_limit = Some(600_000); - config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); - config.approvals_reviewer = ApprovalsReviewer::AutoReview; + let time_provider = Arc::new(RecordingTimeProvider { + thread_ids: Mutex::new(Vec::new()), }); + let mut builder = test_codex() + .with_config(move |config| { + config.model_catalog = Some(ModelsResponse { + models: vec![review_model], + }); + config.model_context_window = Some(900_000); + config.model_auto_compact_token_limit = Some(600_000); + config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + config.approvals_reviewer = ApprovalsReviewer::AutoReview; + config + .features + .enable(Feature::CurrentTimeReminder) + .expect("test config should allow current-time reminders"); + config.current_time_reminder = Some(CurrentTimeReminderConfig { + clock_source: CurrentTimeSource::External, + ..CurrentTimeReminderConfig::default() + }); + }) + .with_external_time_provider(time_provider.clone()); let test = builder.build_with_websocket_server(&server).await?; + let root_thread_id = test.session_configured.thread_id; let (first, second) = tokio::time::timeout(Duration::from_secs(5), async { tokio::join!( server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 0), @@ -103,6 +152,14 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review() -> Result<() ) }) .await?; + assert!( + time_provider + .thread_ids + .lock() + .expect("time-provider thread ids lock should not be poisoned") + .is_empty(), + "startup prewarm must not request the external clock" + ); let prewarm_requests = [first.body_json(), second.body_json()]; let guardian_prewarm = prewarm_requests .iter() @@ -145,6 +202,32 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review() -> Result<() guardian_review["client_metadata"]["thread_id"].as_str(), Some(guardian_thread_id) ); + let current_date = DateTime::::from_timestamp(CURRENT_TIME_AT, 0) + .expect("test timestamp should be valid") + .with_timezone(&Local) + .format("%Y-%m-%d") + .to_string(); + assert!( + guardian_review + .to_string() + .contains(&format!("{current_date}")), + "guardian's environment context should use the simulated current date" + ); + let guardian_thread_id = ThreadId::from_string(guardian_thread_id)?; + { + let thread_ids = time_provider + .thread_ids + .lock() + .expect("time-provider thread ids lock should not be poisoned"); + assert!(thread_ids.contains(&root_thread_id)); + assert!(thread_ids.contains(&guardian_thread_id)); + assert!( + thread_ids + .iter() + .all(|thread_id| thread_id == &root_thread_id || thread_id == &guardian_thread_id), + "clock requests should use the corresponding agent's own thread id: {thread_ids:?}" + ); + } assert_eq!(guardian_review.get("generate"), None); let guardian_rollout_path = test