mirror of
https://github.com/openai/codex.git
synced 2026-10-19 17:28:46 +00:00
Refresh environment dates from the configured clock (#36187)
## Why The `<current_date>` 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
This commit is contained in:
@@ -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::<Utc>::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("<environment_context>")
|
||||
&& text.contains(&format!("<current_date>{current_date}</current_date>"))
|
||||
}));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()))
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -27,6 +27,7 @@ impl EnvironmentsState {
|
||||
pub(crate) fn from_turn_context_with_environments(
|
||||
turn_context: &TurnContext,
|
||||
environments: &TurnEnvironmentSnapshot,
|
||||
current_date: Option<String>,
|
||||
) -> 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(
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -2992,10 +2992,10 @@ impl Session {
|
||||
&self,
|
||||
previous_world_state: &Arc<WorldState>,
|
||||
step_context: &step_context::StepContext,
|
||||
) -> Arc<WorldState> {
|
||||
) -> CodexResult<Arc<WorldState>> {
|
||||
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<WorldState> {
|
||||
) -> CodexResult<Arc<WorldState>> {
|
||||
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(
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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<TurnContext>,
|
||||
) -> 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("<environment_context>"))
|
||||
.expect("environment update item should be emitted");
|
||||
assert!(environment_update.contains("<current_date>2026-02-27</current_date>"));
|
||||
let current_date = chrono::Local::now().format("%Y-%m-%d").to_string();
|
||||
assert!(environment_update.contains(&format!("<current_date>{current_date}</current_date>")));
|
||||
assert!(environment_update.contains("<timezone>Europe/Berlin</timezone>"));
|
||||
}
|
||||
|
||||
@@ -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");
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<ResponseItem> = async {
|
||||
|
||||
@@ -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<WorldState> {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
};
|
||||
|
||||
@@ -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::<Utc>::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("<environment_context>")
|
||||
&& text.contains(&format!("<current_date>{current_date}</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
|
||||
|
||||
@@ -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<Vec<ThreadId>>,
|
||||
}
|
||||
|
||||
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::<Utc>::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::<Utc>::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>{current_date}</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
|
||||
|
||||
Reference in New Issue
Block a user