openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-14 21:40:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make clock reads synchronous and remove client clock requests (#61382)

Browse Source 
## What changed

- Make `TimeProvider::current_time` and `CodexThread::current_time` synchronous, with clock reads required to avoid yielding or calling back into the session.
- Remove the experimental `currentTime/read` app-server request and its client-backed clock and sleep polling implementation.
- Require a host-provided `TimeProvider` for external clocks. App-server thread startup now fails when `features.current_time_reminder.clock_source = "external"` has no provider.

## Testing

Add coverage for external provider selection and rejection of app-server thread startup without an external provider.

GitOrigin-RevId: 09479d6eb28453b343dee78b3673092e73ae5d73
This commit is contained in:
Ruslan Nigmatullin
2026-11-13 21:51:53 +00:00
committed by copyberry
parent e40b0e62e2
commit 4aac2eacc5
28 changed files with 87 additions and 623 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/agent-message-board-host/src/lib.rs
+1 -1
View File
@@ -212,7 +212,7 @@ impl MessageBoardHost for LocalBoardHost {
Box::pin(async move {
let actor = self.actor(caller).await?;
self.registered_path(&actor).await?;
actor.current_time().await
actor.current_time()
})
}
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server-protocol/src/export.rs
-15
View File
@@ -2229,13 +2229,6 @@ mod tests {
.ok_or_else(|| anyhow::anyhow!("missing MCP login completion fixture"))?,
)?;
assert!(mcp_login_completion_ts.contains("loginId?: string | null"));
let server_request_ts = std::str::from_utf8(
fixture_tree
.get(Path::new("ServerRequest.ts"))
.ok_or_else(|| anyhow::anyhow!("missing ServerRequest.ts fixture"))?,
)?;
assert_eq!(server_request_ts.contains("currentTime/read"), false);
assert_eq!(server_request_ts.contains("CurrentTimeReadParams"), false);
let typescript_index = std::str::from_utf8(
fixture_tree
.get(Path::new("index.ts"))
@@ -2256,14 +2249,6 @@ mod tests {
fixture_tree.contains_key(Path::new("v2/MockExperimentalMethodResponse.ts")),
false
);
assert_eq!(
fixture_tree.contains_key(Path::new("v2/CurrentTimeReadParams.ts")),
false
);
assert_eq!(
fixture_tree.contains_key(Path::new("v2/CurrentTimeReadResponse.ts")),
false
);
assert_eq!(
fixture_tree.contains_key(Path::new("v2/RemoteControlClient.ts")),
false
codex-rs/app-server-protocol/src/protocol/common.rs
-33
View File
@@ -1943,13 +1943,6 @@ server_request_definitions! {
response: v2::AttestationGenerateResponse,
},
#[experimental("currentTime/read")]
/// Read the current time from an external clock owned by the client.
CurrentTimeRead => "currentTime/read" {
params: v2::CurrentTimeReadParams,
response: v2::CurrentTimeReadResponse,
},
/// DEPRECATED APIs below
/// Request to approve a patch.
/// This request is used for Turns started via the legacy APIs (i.e. SendUserTurn, SendUserMessage).
@@ -3131,32 +3124,6 @@ mod tests {
Ok(())
}
#[test]
fn serialize_current_time_read_request() -> Result<()> {
let params = v2::CurrentTimeReadParams {
thread_id: "thread-123".to_string(),
};
let request = ServerRequest::CurrentTimeRead {
request_id: RequestId::Integer(10),
params: params.clone(),
};
assert_eq!(
json!({
"method": "currentTime/read",
"id": 10,
"params": {
"threadId": "thread-123"
}
}),
serde_json::to_value(&request)?,
);
let payload = ServerRequestPayload::CurrentTimeRead(params);
assert_eq!(request.id(), &RequestId::Integer(10));
assert_eq!(payload.request_with_id(RequestId::Integer(10)), request);
Ok(())
}
#[test]
fn serialize_server_response() -> Result<()> {
let response = ServerResponse::CommandExecutionRequestApproval {
codex-rs/app-server-protocol/src/protocol/v2/current_time.rs
-20
View File
@@ -1,20 +0,0 @@
use crate::JsonSchema;
use crate::TS;
use serde::Deserialize;
use serde::Serialize;
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct CurrentTimeReadParams {
pub thread_id: String,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct CurrentTimeReadResponse {
/// Current time as whole Unix seconds.
#[ts(type = "number")]
pub current_time_at: i64,
}
codex-rs/app-server-protocol/src/protocol/v2/mod.rs
-2
View File
@@ -10,7 +10,6 @@ mod collaboration_mode;
mod command_exec;
mod computer_use_config;
mod config;
mod current_time;
mod diagnostics;
mod environment;
mod experimental_feature;
@@ -54,7 +53,6 @@ pub use collaboration_mode::*;
pub use command_exec::*;
pub use computer_use_config::*;
pub use config::*;
pub use current_time::*;
pub use diagnostics::*;
pub use environment::*;
pub use experimental_feature::*;
codex-rs/app-server/src/current_time.rs
-177
View File
@@ -1,177 +0,0 @@
use std::sync::Arc;
use std::sync::Weak;
use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use anyhow::bail;
use chrono::DateTime;
use chrono::Utc;
use codex_app_server_protocol::CurrentTimeReadParams;
use codex_app_server_protocol::CurrentTimeReadResponse;
use codex_app_server_protocol::ServerRequestPayload;
use codex_core::SleepFuture;
use codex_core::TimeFuture;
use codex_core::TimeProvider;
use codex_protocol::ThreadId;
use tokio::time::Duration;
use tokio::time::Instant;
use tokio::time::timeout_at;
use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::OutgoingMessageSender;
use crate::thread_state::ThreadStateManager;
const CURRENT_TIME_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
const CURRENT_TIME_POLL_INTERVAL: Duration = Duration::from_secs(1);
pub(crate) fn app_server_time_provider(
outgoing: Arc<OutgoingMessageSender>,
thread_state_manager: ThreadStateManager,
) -> Arc<dyn TimeProvider> {
Arc::new(AppServerTimeProvider {
outgoing: Arc::downgrade(&outgoing),
thread_state_manager,
})
}
struct AppServerTimeProvider {
outgoing: Weak<OutgoingMessageSender>,
thread_state_manager: ThreadStateManager,
}
impl TimeProvider for AppServerTimeProvider {
fn current_time(&self, thread_id: ThreadId) -> TimeFuture<'_> {
let outgoing = self.outgoing.clone();
let thread_state_manager = self.thread_state_manager.clone();
Box::pin(async move {
let outgoing = outgoing
.upgrade()
.context("app-server current-time provider is unavailable")?;
request_current_time(outgoing, thread_state_manager, thread_id).await
})
}
fn sleep(&self, thread_id: ThreadId, duration: Duration) -> SleepFuture<'_> {
let outgoing = self.outgoing.clone();
let thread_state_manager = self.thread_state_manager.clone();
Box::pin(async move {
let outgoing = outgoing
.upgrade()
.context("app-server current-time provider is unavailable")?;
let started_at =
request_current_time(outgoing.clone(), thread_state_manager.clone(), thread_id)
.await?;
let wake_at = started_at
.checked_add_signed(
chrono::Duration::from_std(duration)
.context("external sleep duration is outside the supported range")?,
)
.context("external sleep deadline is outside the supported range")?;
loop {
tokio::time::sleep(CURRENT_TIME_POLL_INTERVAL).await;
if request_current_time(outgoing.clone(), thread_state_manager.clone(), thread_id)
.await?
>= wake_at
{
return Ok(());
}
}
})
}
}
async fn request_current_time(
outgoing: Arc<OutgoingMessageSender>,
thread_state_manager: ThreadStateManager,
thread_id: ThreadId,
) -> Result<DateTime<Utc>> {
let deadline = Instant::now() + CURRENT_TIME_REQUEST_TIMEOUT;
timeout_at(
deadline,
thread_state_manager.wait_for_thread_subscriber(thread_id),
)
.await
.map_err(|_| {
anyhow!(
"timed out waiting for a client to subscribe to the thread after {}s",
CURRENT_TIME_REQUEST_TIMEOUT.as_secs()
)
})?;
let connection_ids = thread_state_manager
.subscribed_connection_ids(thread_id)
.await;
let connection_id = require_single_current_time_connection(&connection_ids)?;
let connection_ids = [connection_id];
let (request_id, rx) = outgoing
.send_request_to_connections(
Some(&connection_ids),
ServerRequestPayload::CurrentTimeRead(CurrentTimeReadParams {
thread_id: thread_id.to_string(),
}),
/*thread_id*/ None,
)
.await;
let result = match timeout_at(deadline, rx).await {
Ok(Ok(Ok(result))) => result,
Ok(Ok(Err(err))) => {
bail!(
"current-time request failed: code={} message={}",
err.code,
err.message
);
}
Ok(Err(err)) => bail!("current-time request was canceled: {err}"),
Err(_) => {
let _canceled = outgoing.cancel_request(&request_id).await;
bail!(
"current-time request timed out after {}s",
CURRENT_TIME_REQUEST_TIMEOUT.as_secs()
);
}
};
let response: CurrentTimeReadResponse =
serde_json::from_value(result).context("invalid current-time response")?;
DateTime::from_timestamp(response.current_time_at, 0)
.ok_or_else(|| anyhow!("current-time response is outside the supported range"))
}
fn require_single_current_time_connection(connection_ids: &[ConnectionId]) -> Result<ConnectionId> {
// External clocks are not interchangeable, so do not choose one silently.
match connection_ids {
[connection_id] => Ok(*connection_id),
_ => bail!(
"expected exactly one client subscribed to the thread, found {}",
connection_ids.len()
),
}
}
#[cfg(test)]
mod tests {
use super::require_single_current_time_connection;
use crate::outgoing_message::ConnectionId;
#[test]
fn current_time_connection_must_be_unambiguous() {
assert_eq!(
require_single_current_time_connection(&[ConnectionId(7)]).unwrap(),
ConnectionId(7)
);
assert_eq!(
require_single_current_time_connection(&[])
.unwrap_err()
.to_string(),
"expected exactly one client subscribed to the thread, found 0"
);
assert_eq!(
require_single_current_time_connection(&[ConnectionId(7), ConnectionId(8)])
.unwrap_err()
.to_string(),
"expected exactly one client subscribed to the thread, found 2"
);
}
}
codex-rs/app-server/src/lib.rs
-1
View File
@@ -116,7 +116,6 @@ mod config_manager;
mod config_manager_service;
mod connection_cleanup;
mod connection_rpc_gate;
mod current_time;
mod daemon_thread_recovery;
mod dynamic_tools;
mod effective_plugin_change;
codex-rs/app-server/src/message_processor.rs
+1 -6
View File
@@ -8,7 +8,6 @@ use std::sync::atomic::AtomicBool;
use crate::attestation::app_server_attestation_provider;
use crate::config_manager::ConfigManager;
use crate::connection_rpc_gate::ConnectionRpcGate;
use crate::current_time::app_server_time_provider;
use crate::error_code::internal_error;
use crate::error_code::invalid_params;
use crate::error_code::invalid_request;
@@ -417,11 +416,7 @@ impl MessageProcessor {
outgoing.clone(),
thread_state_manager.clone(),
)),
{
let time_provider =
app_server_time_provider(outgoing.clone(), thread_state_manager.clone());
Some(time_provider)
},
None,
);
match code_mode_session_provider {
Some(provider) => manager.with_code_mode_session_provider(provider),
codex-rs/app-server/tests/suite/v2/current_time.rs
+17 -103
View File
@@ -1,126 +1,40 @@
//! Explicit external time requires a native clock on the app-server.
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::RequestId;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput;
use core_test_support::responses;
use core_test_support::skip_if_no_network;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use tokio::time::Duration;
use tokio::time::timeout;
#[cfg(windows)]
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(25);
#[cfg(not(windows))]
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10);
const CURRENT_TIME_AT: i64 = 1_781_717_655;
const CURRENT_TIME_REMINDER: &str =
"<current_time_reminder>It is 2026-06-17 17:34:15 UTC.</current_time_reminder>";
#[tokio::test]
async fn current_time_read_round_trip_adds_reminder_to_model_input() -> Result<()> {
async fn external_time_without_native_provider_fails_thread_start() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let response_mock = responses::mount_sse_once(
&server,
create_final_assistant_message_sse_response("Done")?,
)
.await;
let codex_home = TempDir::new()?;
let home = TempDir::new()?;
MockResponsesConfig::new(&server.uri())
.with_extra_config(
r#"[features.current_time_reminder]
enabled = true
reminder_interval_seconds = 1
clock_source = "external"
"#,
"[features.current_time_reminder]\nenabled = true\nclock_source = \"external\"",
)
.write(codex_home.path())?;
.write(home.path())?;
let mut app_server = TestAppServer::builder()
.with_codex_home(codex_home.path())
.with_codex_home(home.path())
.build_initialized()
.await?;
let ThreadStartResponse { thread, .. } = app_server
.start_thread(ThreadStartParams::default())
let id = app_server
.send_thread_start_request_with_auto_env(ThreadStartParams::default())
.await?;
let _: TurnStartResponse = app_server
.request(|request_id| ClientRequest::TurnStart {
request_id,
params: TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "What time is it?".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
},
})
let error = app_server
.read_stream_until_error_message(RequestId::Integer(id))
.await?;
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!(
request
.message_input_texts("developer")
.iter()
.any(|text| text == CURRENT_TIME_REMINDER)
error
.error
.message
.contains("no external current-time provider"),
"{}",
error.error.message
);
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(())
}
codex-rs/app-server/tests/suite/v2/sleep.rs
-196
View File
@@ -2,12 +2,7 @@ use anyhow::Result;
use app_test_support::MockResponsesConfig;
use app_test_support::TestAppServer;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::CurrentTimeReadResponse;
use codex_app_server_protocol::ItemCompletedNotification;
use codex_app_server_protocol::ItemStartedNotification;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::SleepItem;
use codex_app_server_protocol::ThreadItemPayload;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
@@ -21,7 +16,6 @@ use pretty_assertions::assert_eq;
use serde_json::json;
use std::time::Duration;
use tempfile::TempDir;
use test_case::test_case;
use tokio::time::timeout;
use super::analytics::captured_analytics_events;
@@ -32,7 +26,6 @@ use super::analytics::wait_for_matching_analytics_event;
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(25);
#[cfg(not(windows))]
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10);
const CURRENT_TIME_AT: i64 = 1_781_717_655;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn clock_tools_emit_control_tool_analytics() -> Result<()> {
@@ -163,157 +156,6 @@ async fn clock_tools_emit_control_tool_analytics() -> Result<()> {
Ok(())
}
#[test_case(0; "initial read fails")]
#[test_case(1; "polling read fails")]
#[test_case(3; "successful reads")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn external_sleep_polls_current_time_and_emits_items(
successful_sleep_reads: usize,
) -> Result<()> {
const CALL_ID: &str = "sleep-1";
const DURATION_MS: u64 = 2_000;
let read_fails = successful_sleep_reads < 3;
let server = responses::start_mock_server().await;
let model = responses::mount_sse_sequence(
&server,
vec![
responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_function_call_with_namespace(
CALL_ID,
"clock",
"sleep",
&serde_json::json!({ "duration_ms": DURATION_MS }).to_string(),
),
responses::ev_completed("resp-1"),
]),
responses::sse(vec![
responses::ev_assistant_message("msg-1", "Done"),
responses::ev_completed("resp-2"),
]),
],
)
.await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri())
.with_root_config("include_environment_context = false")
.with_extra_config(&format!(
r#"[features]
nonfatal_clock_read_errors = {read_fails}
[features.current_time_reminder]
enabled = true
sleep_tool = true
clock_source = "external"
"#,
))
.write(codex_home.path())?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let ThreadStartResponse { thread, .. } = mcp
.start_thread(ThreadStartParams {
model: Some("mock-model".to_string()),
..Default::default()
})
.await?;
let TurnStartResponse { turn, .. } = mcp
.request(|request_id| ClientRequest::TurnStart {
request_id,
params: TurnStartParams {
thread_id: thread.id.clone(),
client_user_message_id: None,
input: vec![V2UserInput::Text {
text: "Sleep briefly".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
},
})
.await?;
// Read once for the initial reminder, then once to establish the sleep deadline.
respond_to_current_time_read(&mut mcp, &thread.id, CURRENT_TIME_AT).await?;
let started = wait_for_sleep_started(&mut mcp, CALL_ID).await?;
for current_time_at in [CURRENT_TIME_AT, CURRENT_TIME_AT + 1, CURRENT_TIME_AT + 2]
.into_iter()
.take(successful_sleep_reads)
{
respond_to_current_time_read(&mut mcp, &thread.id, current_time_at).await?;
}
if read_fails {
let request = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_request_message(),
)
.await??;
let ServerRequest::CurrentTimeRead { request_id, params } = request else {
panic!("expected clock read, got {request:?}");
};
assert_eq!(params.thread_id, thread.id);
mcp.send_error(
request_id,
JSONRPCErrorError {
code: -32_000,
message: "PRIVATE_CLOCK_ERROR".to_string(),
data: None,
},
)
.await?;
}
let completed = wait_for_sleep_completed(&mut mcp, CALL_ID).await?;
// The next inference boundary reads the same external clock after the tool returns.
respond_to_current_time_read(&mut mcp, &thread.id, CURRENT_TIME_AT + 2).await?;
let turn_completed: TurnCompletedNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_notification("turn/completed"),
)
.await??;
assert_eq!(turn_completed.turn.status, TurnStatus::Completed);
if read_fails {
let requests = model.requests();
assert_eq!(requests.len(), 2);
assert_eq!(
requests[1].function_call_output_text(CALL_ID).as_deref(),
Some("failed to read current time")
);
}
let expected_item = ThreadItemPayload::Sleep(SleepItem {
id: CALL_ID.to_string(),
duration_ms: DURATION_MS,
});
assert!(completed.completed_at_ms >= started.started_at_ms);
assert_eq!(
started,
ItemStartedNotification::new(
thread.id.clone(),
turn.id.clone(),
expected_item.clone(),
started.started_at_ms
)
);
assert_eq!(
completed,
ItemCompletedNotification::new(
thread.id,
turn.id,
expected_item,
Some(started.started_at_ms),
completed.completed_at_ms
)
);
Ok(())
}
async fn wait_for_sleep_started(
mcp: &mut TestAppServer,
call_id: &str,
@@ -326,41 +168,3 @@ async fn wait_for_sleep_started(
}
}
}
async fn wait_for_sleep_completed(
mcp: &mut TestAppServer,
call_id: &str,
) -> Result<ItemCompletedNotification> {
loop {
let completed: ItemCompletedNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_notification("item/completed"),
)
.await??;
if matches!(&completed.item.payload, ThreadItemPayload::Sleep(item) if item.id == call_id) {
return Ok(completed);
}
}
}
async fn respond_to_current_time_read(
mcp: &mut TestAppServer,
thread_id: &str,
current_time_at: i64,
) -> Result<()> {
let request = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_request_message(),
)
.await??;
let ServerRequest::CurrentTimeRead { request_id, params } = request else {
panic!("expected CurrentTimeRead request, got: {request:?}");
};
assert_eq!(params.thread_id, thread_id);
mcp.send_response(
request_id,
serde_json::to_value(CurrentTimeReadResponse { current_time_at })?,
)
.await?;
Ok(())
}
codex-rs/core/src/codex_thread.rs
+1 -2
View File
@@ -278,12 +278,11 @@ impl CodexThread {
}
/// Reads this thread's configured clock.
pub async fn current_time(&self) -> CodexResult<DateTime<Utc>> {
pub fn current_time(&self) -> CodexResult<DateTime<Utc>> {
self.session
.services
.time_provider
.current_time(self.session.thread_id)
.await
.map_err(|error| CodexErr::Io(std::io::Error::other(error)))
}
codex-rs/core/src/current_time.rs
+8 -4
View File
@@ -12,12 +12,12 @@ use codex_protocol::ThreadId;
use crate::config::CurrentTimeReminderConfig;
pub type TimeFuture<'a> = Pin<Box<dyn Future<Output = Result<DateTime<Utc>>> + Send + 'a>>;
pub type SleepFuture<'a> = Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;
/// Host integration boundary for reading and waiting on the current time.
pub trait TimeProvider: Send + Sync {
fn current_time(&self, thread_id: ThreadId) -> TimeFuture<'_>;
/// Reads the clock without yielding or calling back into the session.
fn current_time(&self, thread_id: ThreadId) -> Result<DateTime<Utc>>;
/// Waits for the given duration on this provider's clock.
///
@@ -28,8 +28,8 @@ pub trait TimeProvider: Send + Sync {
pub(crate) struct SystemTimeProvider;
impl TimeProvider for SystemTimeProvider {
fn current_time(&self, _thread_id: ThreadId) -> TimeFuture<'_> {
Box::pin(async { Ok(Utc::now()) })
fn current_time(&self, _thread_id: ThreadId) -> Result<DateTime<Utc>> {
Ok(Utc::now())
}
fn sleep(&self, _thread_id: ThreadId, duration: Duration) -> SleepFuture<'_> {
@@ -53,3 +53,7 @@ pub(crate) fn resolve_time_provider(
}),
}
}
#[cfg(test)]
#[path = "current_time_tests.rs"]
mod tests;
codex-rs/core/src/current_time_tests.rs
+22
View File
@@ -0,0 +1,22 @@
#![cfg(test)]
use super::*;
#[test]
fn explicit_external_time_requires_a_provider() {
let config = CurrentTimeReminderConfig {
clock_source: CurrentTimeSource::External,
..Default::default()
};
let err = resolve_time_provider(Some(&config), /*external_provider*/ None)
.err()
.expect("missing provider");
assert!(
err.to_string()
.contains("no external current-time provider")
);
let provider: Arc<dyn TimeProvider> = Arc::new(SystemTimeProvider);
let selected = resolve_time_provider(Some(&config), Some(Arc::clone(&provider))).unwrap();
assert!(Arc::ptr_eq(&selected, &provider));
assert!(resolve_time_provider(/*config*/ None, /*external_provider*/ None).is_ok());
}
codex-rs/core/src/lib.rs
-1
View File
@@ -281,7 +281,6 @@ pub use codex_prompts::REVIEW_PROMPT;
pub use codex_runtime_agent_instance::resolve_installation_id;
pub use compact::content_items_to_text;
pub use current_time::SleepFuture;
pub use current_time::TimeFuture;
pub use current_time::TimeProvider;
pub use event_mapping::parse_turn_item;
pub use exec_policy::ExecPolicyError;
codex-rs/core/src/session/time_reminder.rs
+1 -6
View File
@@ -102,12 +102,7 @@ impl Session {
turn_context: &TurnContext,
clock_read: &'static str,
) -> CodexResult<Option<DateTime<Utc>>> {
let error = match self
.services
.time_provider
.current_time(self.thread_id)
.await
{
let error = match self.services.time_provider.current_time(self.thread_id) {
Ok(time) => {
let mut state = self.state.lock().await;
state.current_time_reminder.last_clock_failure = None;
codex-rs/core/src/tools/handlers/current_time.rs
-1
View File
@@ -103,7 +103,6 @@ impl ToolExecutor<ToolInvocation> for CurrentTimeHandler {
.services
.time_provider
.current_time(invocation.session.thread_id)
.await
.map_err(|err| {
if invocation
.turn
codex-rs/core/tests/suite/current_time_reminder.rs
+6 -9
View File
@@ -14,7 +14,6 @@ use chrono::Local;
use chrono::Utc;
use codex_config::test_support::CloudConfigBundleFixture;
use codex_core::SleepFuture;
use codex_core::TimeFuture;
use codex_core::TimeProvider;
use codex_core::TurnInputRequest;
use codex_core::config::CurrentTimeReminderConfig;
@@ -87,7 +86,7 @@ impl Default for TestTimeProvider {
}
impl TimeProvider for TestTimeProvider {
fn current_time(&self, _thread_id: ThreadId) -> TimeFuture<'_> {
fn current_time(&self, _thread_id: ThreadId) -> anyhow::Result<chrono::DateTime<chrono::Utc>> {
let timestamp = if let Some(script) = &self.scripted_current_time {
script
.lock()
@@ -97,11 +96,9 @@ impl TimeProvider for TestTimeProvider {
} else {
Ok(self.current_time.fetch_add(60, Ordering::Relaxed))
};
Box::pin(async move {
let timestamp = timestamp.map_err(anyhow::Error::msg)?;
Ok(DateTime::<Utc>::from_timestamp(timestamp, 0)
.expect("test timestamp should be valid"))
})
let timestamp = timestamp.map_err(anyhow::Error::msg)?;
Ok(DateTime::<Utc>::from_timestamp(timestamp, 0).expect("test timestamp should be valid"))
}
fn sleep(&self, _thread_id: ThreadId, duration: Duration) -> SleepFuture<'_> {
@@ -114,8 +111,8 @@ impl TimeProvider for TestTimeProvider {
struct FailingTimeProvider;
impl TimeProvider for FailingTimeProvider {
fn current_time(&self, _thread_id: ThreadId) -> TimeFuture<'_> {
Box::pin(async { Err(anyhow!("test clock unavailable")) })
fn current_time(&self, _thread_id: ThreadId) -> anyhow::Result<chrono::DateTime<chrono::Utc>> {
Err(anyhow!("test clock unavailable"))
}
fn sleep(&self, _thread_id: ThreadId, _duration: Duration) -> SleepFuture<'_> {
codex-rs/core/tests/suite/guardian_review.rs
+4 -6
View File
@@ -8,7 +8,6 @@ use chrono::Utc;
use codex_config::types::McpServerConfig;
use codex_config::types::OtelExporterKind;
use codex_core::SleepFuture;
use codex_core::TimeFuture;
use codex_core::TimeProvider;
use codex_core::TurnInputRequest;
use codex_core::config::Config;
@@ -128,15 +127,14 @@ impl ToolLifecycleContributor for RecordingToolLifecycleContributor {
}
impl TimeProvider for RecordingTimeProvider {
fn current_time(&self, thread_id: ThreadId) -> TimeFuture<'_> {
fn current_time(&self, thread_id: ThreadId) -> anyhow::Result<chrono::DateTime<chrono::Utc>> {
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"))
})
Ok(DateTime::<Utc>::from_timestamp(CURRENT_TIME_AT, 0)
.expect("test timestamp should be valid"))
}
fn sleep(&self, _thread_id: ThreadId, _duration: Duration) -> SleepFuture<'_> {
codex-rs/core/tests/suite/scenarios_agent_message_board.rs
+11 -10
View File
@@ -42,16 +42,17 @@ enum BoardClock {
}
impl codex_core::TimeProvider for BoardClock {
fn current_time(&self, _thread_id: codex_protocol::ThreadId) -> codex_core::TimeFuture<'_> {
Box::pin(async move {
match self {
Self::Available => Ok(chrono::DateTime::parse_from_rfc3339(
"2026-09-18T12:00:00Z",
)?
.with_timezone(&chrono::Utc)),
Self::Unavailable => Err(anyhow::anyhow!("board clock unavailable")),
}
})
fn current_time(
&self,
_thread_id: codex_protocol::ThreadId,
) -> anyhow::Result<chrono::DateTime<chrono::Utc>> {
match self {
Self::Available => Ok(
chrono::DateTime::parse_from_rfc3339("2026-09-18T12:00:00Z")?
.with_timezone(&chrono::Utc),
),
Self::Unavailable => Err(anyhow::anyhow!("board clock unavailable")),
}
}
fn sleep(
codex-rs/core/tests/suite/subagent_notifications.rs
+7 -5
View File
@@ -1719,12 +1719,14 @@ async fn spawned_full_history_v2_child_uses_model_precedence_without_dropping_co
struct FailFirstClockRead(std::sync::atomic::AtomicBool);
impl codex_core::TimeProvider for FailFirstClockRead {
fn current_time(&self, _thread_id: ThreadId) -> codex_core::TimeFuture<'_> {
fn current_time(
&self,
_thread_id: ThreadId,
) -> anyhow::Result<chrono::DateTime<chrono::Utc>> {
let already_read = self.0.swap(true, std::sync::atomic::Ordering::Relaxed);
Box::pin(async move {
anyhow::ensure!(already_read, "parent clock unavailable");
Ok(chrono::Utc::now())
})
anyhow::ensure!(already_read, "parent clock unavailable");
Ok(chrono::Utc::now())
}
fn sleep(
codex-rs/exec/src/lib.rs
-9
View File
@@ -2315,15 +2315,6 @@ async fn handle_server_request(
)
.await
}
ServerRequest::CurrentTimeRead { request_id, .. } => {
reject_server_request(
client,
request_id,
&method,
"external current time is not supported in exec mode".to_string(),
)
.await
}
ServerRequest::ApplyPatchApproval { request_id, params } => {
reject_server_request(
client,
codex-rs/tui/src/app/agents_overview_tests.rs
+8 -3
View File
@@ -146,7 +146,7 @@ use crate::test_support::test_path_display;
use codex_app_server_client::AppServerClient;
use codex_app_server_client::AppServerEvent;
use codex_app_server_client::CloudClient;
use codex_app_server_protocol::CurrentTimeReadParams;
use codex_app_server_protocol::DynamicToolCallParams;
use codex_app_server_protocol::ReasoningSummaryTextDeltaNotification;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerRequest;
@@ -2016,10 +2016,15 @@ async fn root_switch_preserves_idle_root_with_running_subagent() -> Result<()> {
);
app.handle_app_server_event(
&app_server,
AppServerEvent::ServerRequest(Box::new(ServerRequest::CurrentTimeRead {
AppServerEvent::ServerRequest(Box::new(ServerRequest::DynamicToolCall {
request_id: RequestId::Integer(99),
params: CurrentTimeReadParams {
params: DynamicToolCallParams {
thread_id: idle_child_id.to_string(),
turn_id: "turn".into(),
call_id: "call".into(),
namespace: None,
tool: "test".into(),
arguments: serde_json::Value::Null,
},
})),
)
codex-rs/tui/src/app/app_server_event_targets.rs
-3
View File
@@ -24,9 +24,6 @@ pub(super) fn server_request_thread_id(request: &ServerRequest) -> Option<Thread
ServerRequest::DynamicToolCall { params, .. } => {
ThreadId::from_string(&params.thread_id).ok()
}
ServerRequest::CurrentTimeRead { params, .. } => {
ThreadId::from_string(&params.thread_id).ok()
}
ServerRequest::ChatgptAuthTokensRefresh { .. }
| ServerRequest::AttestationGenerate { .. }
| ServerRequest::ApplyPatchApproval { .. }
codex-rs/tui/src/app/app_server_events.rs
-1
View File
@@ -791,7 +791,6 @@ impl App {
&request,
ServerRequest::DynamicToolCall { .. }
| ServerRequest::AttestationGenerate { .. }
| ServerRequest::CurrentTimeRead { .. }
| ServerRequest::ApplyPatchApproval { .. }
| ServerRequest::ExecCommandApproval { .. }
);
codex-rs/tui/src/app/app_server_requests.rs
-7
View File
@@ -203,12 +203,6 @@ impl PendingAppServerRequests {
message: "Attestation generation is not available in TUI.".to_string(),
})
}
ServerRequest::CurrentTimeRead { request_id, .. } => {
Some(UnsupportedAppServerRequest {
request_id: request_id.clone(),
message: "External current time is not available in TUI.".to_string(),
})
}
ServerRequest::ApplyPatchApproval { request_id, .. } => {
Some(UnsupportedAppServerRequest {
request_id: request_id.clone(),
@@ -530,7 +524,6 @@ impl PendingAppServerRequests {
ServerRequest::ChatgptAuthTokensRefresh { .. } => None,
ServerRequest::DynamicToolCall { .. }
| ServerRequest::AttestationGenerate { .. }
| ServerRequest::CurrentTimeRead { .. }
| ServerRequest::ApplyPatchApproval { .. }
| ServerRequest::ExecCommandApproval { .. } => None,
}
codex-rs/tui/src/app/side.rs
-1
View File
@@ -67,7 +67,6 @@ impl SideParentStatus {
| ServerRequest::ExecCommandApproval { .. } => Some(SideParentStatus::NeedsApproval),
ServerRequest::DynamicToolCall { .. }
| ServerRequest::AttestationGenerate { .. }
| ServerRequest::CurrentTimeRead { .. }
| ServerRequest::ChatgptAuthTokensRefresh { .. } => None,
}
}
codex-rs/tui/src/chatwidget/protocol_requests.rs
-1
View File
@@ -50,7 +50,6 @@ impl ChatWidget {
}
ServerRequest::DynamicToolCall { .. }
| ServerRequest::AttestationGenerate { .. }
| ServerRequest::CurrentTimeRead { .. }
| ServerRequest::ChatgptAuthTokensRefresh { .. }
| ServerRequest::ApplyPatchApproval { .. }
| ServerRequest::ExecCommandApproval { .. } => {