agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

[codex] Add interruptible sleep tool (#28429)

Browse Source 
## Why

Models sometimes need to pause briefly while waiting for external work,
but using a shell command for that delay ties the wait to a process and
does not naturally resume when new turn input arrives.

## What changed

- add a built-in `sleep` tool behind the under-development `sleep_tool`
feature
- accept a bounded `duration_ms` argument, matching the millisecond
convention used by unified exec
- end the sleep early when either steered user input or mailbox input
arrives
- include elapsed wall-clock time in completed and interrupted outputs
- emit a dedicated core `SleepItem` through `item/started` and
`item/completed`
- expose the sleep item as app-server v2 `ThreadItem::Sleep` and retain
it in reconstructed thread history
- regenerate the configuration schema for the new feature flag
- regenerate app-server JSON and TypeScript schema fixtures

## Test plan

- `just test -p codex-core sleep_tool_follows_feature_gate`
- `just test -p codex-core any_new_input_interrupts_sleep`
- `just test -p codex-app-server-protocol`
- `just test -p codex-app-server
sleep_emits_started_and_completed_items`
This commit is contained in:
pakrym-oai
2026-06-15 21:39:21 -07:00
committed by GitHub
Unverified
parent 022f1221e8
commit 08901fc8e1
37 changed files with 1099 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/tests/suite/pending_input.rs
+188 -11
View File
@@ -4,12 +4,15 @@ use std::sync::Arc;
use codex_core::CodexThread;
use codex_features::Feature;
use codex_protocol::AgentPath;
use codex_protocol::items::SleepItem;
use codex_protocol::items::TurnItem;
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::user_input::UserInput;
use core_test_support::context_snapshot;
use core_test_support::context_snapshot::ContextSnapshotOptions;
@@ -65,6 +68,34 @@ fn message_input_texts(body: &Value, role: &str) -> Vec<String> {
.collect()
}
fn function_call_output_text<'a>(body: &'a Value, call_id: &str) -> Option<&'a str> {
body.get("input")
.and_then(Value::as_array)?
.iter()
.find(|item| {
item.get("type").and_then(Value::as_str) == Some("function_call_output")
&& item.get("call_id").and_then(Value::as_str) == Some(call_id)
})?
.get("output")?
.as_str()
}
fn assert_interrupted_sleep_output(output: Option<&str>) {
let Some(output) = output else {
panic!("sleep output missing");
};
let Some(wall_time) = output
.strip_prefix("Wall time: ")
.and_then(|output| output.strip_suffix(" seconds\nSleep interrupted by new input."))
else {
panic!("sleep output should include wall time");
};
assert!(
wall_time.parse::<f64>().is_ok(),
"sleep wall time should be a number"
);
}
fn chunk(event: Value) -> StreamingSseChunk {
StreamingSseChunk {
gate: None,
@@ -207,6 +238,54 @@ async fn wait_for_turn_complete(codex: &CodexThread) {
wait_for_event(codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
}
async fn wait_for_sleep_item_started(codex: &CodexThread, call_id: &str, duration_ms: u64) {
let event = wait_for_event(codex, |event| {
matches!(
event,
EventMsg::ItemStarted(started)
if matches!(&started.item, TurnItem::Sleep(item) if item.id == call_id)
)
})
.await;
let EventMsg::ItemStarted(started) = event else {
unreachable!("wait predicate only accepts item/started events");
};
let TurnItem::Sleep(item) = started.item else {
unreachable!("wait predicate only accepts sleep items");
};
assert_eq!(
item,
SleepItem {
id: call_id.to_string(),
duration_ms,
}
);
}
async fn wait_for_sleep_item_completed(codex: &CodexThread, call_id: &str, duration_ms: u64) {
let event = wait_for_event(codex, |event| {
matches!(
event,
EventMsg::ItemCompleted(completed)
if matches!(&completed.item, TurnItem::Sleep(item) if item.id == call_id)
)
})
.await;
let EventMsg::ItemCompleted(completed) = event else {
unreachable!("wait predicate only accepts item/completed events");
};
let TurnItem::Sleep(item) = completed.item else {
unreachable!("wait predicate only accepts sleep items");
};
assert_eq!(
item,
SleepItem {
id: call_id.to_string(),
duration_ms,
}
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn steer_interrupts_wait_agent_and_is_sent_in_follow_up_request() {
const WAIT_CALL_ID: &str = "wait-call";
@@ -257,17 +336,7 @@ async fn steer_interrupts_wait_agent_and_is_sent_in_follow_up_request() {
relevant_user_input,
vec![INITIAL_PROMPT.to_string(), STEER_PROMPT.to_string()]
);
let wait_output = second["input"]
.as_array()
.expect("second request input")
.iter()
.find(|item| {
item.get("type").and_then(Value::as_str) == Some("function_call_output")
&& item.get("call_id").and_then(Value::as_str) == Some(WAIT_CALL_ID)
})
.and_then(|item| item.get("output"))
.and_then(Value::as_str)
.expect("wait_agent output");
let wait_output = function_call_output_text(&second, WAIT_CALL_ID).expect("wait_agent output");
assert_eq!(
serde_json::from_str::<Value>(wait_output).expect("parse wait_agent output"),
json!({
@@ -279,6 +348,114 @@ async fn steer_interrupts_wait_agent_and_is_sent_in_follow_up_request() {
server.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn any_new_input_interrupts_sleep() {
const FIRST_SLEEP_CALL_ID: &str = "sleep-call-1";
const SECOND_SLEEP_CALL_ID: &str = "sleep-call-2";
const SLEEP_DURATION_MS: u64 = 3_600_000;
const INITIAL_PROMPT: &str = "sleep for a while";
const STEER_PROMPT: &str = "stop sleeping and continue";
let sleep_arguments = json!({ "duration_ms": SLEEP_DURATION_MS }).to_string();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_function_call(
FIRST_SLEEP_CALL_ID,
"sleep",
&sleep_arguments,
)),
chunk(ev_completed("resp-1")),
];
let second_chunks = vec![
chunk(ev_response_created("resp-2")),
chunk(ev_function_call(
SECOND_SLEEP_CALL_ID,
"sleep",
&sleep_arguments,
)),
chunk(ev_completed("resp-2")),
];
let (server, _completions) = start_streaming_sse_server(vec![
first_chunks,
second_chunks,
response_completed_chunks("resp-3"),
])
.await;
let codex = test_codex()
.with_model("gpt-5.4")
.with_config(|config| {
config
.features
.enable(Feature::SleepTool)
.expect("test config should allow feature update");
})
.build_with_streaming_server(&server)
.await
.expect("build Codex test session")
.codex;
submit_user_input(&codex, INITIAL_PROMPT).await;
wait_for_sleep_item_started(&codex, FIRST_SLEEP_CALL_ID, SLEEP_DURATION_MS).await;
steer_user_input(&codex, STEER_PROMPT).await;
wait_for_sleep_item_completed(&codex, FIRST_SLEEP_CALL_ID, SLEEP_DURATION_MS).await;
wait_for_sleep_item_started(&codex, SECOND_SLEEP_CALL_ID, SLEEP_DURATION_MS).await;
submit_queue_only_agent_mail(&codex, "new mailbox input").await;
wait_for_sleep_item_completed(&codex, SECOND_SLEEP_CALL_ID, SLEEP_DURATION_MS).await;
wait_for_turn_complete(&codex).await;
let requests = server.requests().await;
assert_eq!(requests.len(), 3);
let second: Value = from_slice(&requests[1]).expect("parse second request");
let relevant_user_input = message_input_texts(&second, "user")
.into_iter()
.filter(|text| text == INITIAL_PROMPT || text == STEER_PROMPT)
.collect::<Vec<_>>();
assert_eq!(
relevant_user_input,
vec![INITIAL_PROMPT.to_string(), STEER_PROMPT.to_string()]
);
assert_interrupted_sleep_output(function_call_output_text(&second, FIRST_SLEEP_CALL_ID));
let third: Value = from_slice(&requests[2]).expect("parse third request");
assert_interrupted_sleep_output(function_call_output_text(&third, SECOND_SLEEP_CALL_ID));
codex.submit(Op::Shutdown).await.expect("shutdown session");
wait_for_event(&codex, |event| matches!(event, EventMsg::ShutdownComplete)).await;
let rollout_path = codex.rollout_path().expect("rollout path");
let rollout = tokio::fs::read_to_string(rollout_path)
.await
.expect("read rollout");
let persisted_sleep_items = rollout
.lines()
.filter_map(|line| serde_json::from_str::<RolloutLine>(line).ok())
.filter_map(|line| match line.item {
RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) => match event.item {
TurnItem::Sleep(item) => Some(item),
_ => None,
},
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(
persisted_sleep_items,
vec![
SleepItem {
id: FIRST_SLEEP_CALL_ID.to_string(),
duration_ms: SLEEP_DURATION_MS,
},
SleepItem {
id: SECOND_SLEEP_CALL_ID.to_string(),
duration_ms: SLEEP_DURATION_MS,
},
]
);
server.shutdown().await;
}
fn assert_two_responses_input_snapshot(snapshot_name: &str, requests: &[Vec<u8>]) {
assert_eq!(requests.len(), 2);
let options = ContextSnapshotOptions::default().strip_capability_instructions();