openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Backfill exec completion items through paginated history (#57800)

Browse Source 
## Why

`codex exec` recovers turn items at completion because item notifications can be dropped under backpressure. This recovery needs to respect the thread's history mode.

## What changed

For paginated threads, fetch every page from `thread/items/list`, scoped to the completed turn. Preserve items already present in the completion notification when merging the recovered history, and stop recovery if a cursor repeats. Keep `thread/read` recovery for other history modes.

## Testing

Add regression coverage for recovering a final answer across multiple item pages and for a failed fork recovering only its own turn.

GitOrigin-RevId: f005ca2cf9d8313bbbf32ed4c228ebb5e51428a2
This commit is contained in:
Brent Traut
2026-10-28 18:03:57 +00:00
committed by copyberry
parent 22aec5f2b0
commit c92700d299
2 changed files with 130 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/exec/src/lib.rs
+61 -8
View File
@@ -39,6 +39,10 @@ use codex_app_server_protocol::Thread as AppServerThread;
use codex_app_server_protocol::ThreadForkParams;
use codex_app_server_protocol::ThreadForkResponse;
use codex_app_server_protocol::ThreadHistoryMode;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadItemsListCursor;
use codex_app_server_protocol::ThreadItemsListParams;
use codex_app_server_protocol::ThreadItemsListResponse;
use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadListResponse;
use codex_app_server_protocol::ThreadReadParams;
@@ -155,6 +159,7 @@ pub use exec_events::Usage;
pub use exec_events::WebSearchItem;
use serde_json::Value;
use std::collections::HashMap;
use std::collections::HashSet;
use std::future::Future;
use std::io::IsTerminal;
use std::io::Read;
@@ -1028,7 +1033,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
let mut daybreak_enabled = config.daybreak_enabled;
// Resolve resume and fork through existing app-server thread lifecycle APIs.
let (primary_thread_id, fallback_session_configured) = if let Some(ExecCommand::Resume(args)) =
let (history_mode, fallback_session_configured) = if let Some(ExecCommand::Resume(args)) =
command.as_ref()
{
if let Some(thread_id) =
@@ -1054,7 +1059,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
daybreak_enabled = daybreak_override
.or(response.thread.daybreak_enabled)
.unwrap_or(false);
(session_configured.thread_id, session_configured)
(response.thread.history_mode, session_configured)
} else {
let response = start_thread(&client, &mut request_ids, &config, &thread_source)
.await
@@ -1062,7 +1067,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
let session_configured =
session_configured_from_thread_start_response(&response, &config)
.map_err(anyhow::Error::msg)?;
(session_configured.thread_id, session_configured)
(response.thread.history_mode, session_configured)
}
} else if let Some(ExecCommand::Fork(args)) = command.as_ref() {
let source_args = crate::cli::ResumeArgs {
@@ -1150,16 +1155,17 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
.or(response.thread.daybreak_enabled)
.or(source_daybreak)
.unwrap_or(false);
(session_configured.thread_id, session_configured)
(response.thread.history_mode, session_configured)
} else {
let response = start_thread(&client, &mut request_ids, &config, &thread_source)
.await
.map_err(anyhow::Error::msg)?;
let session_configured = session_configured_from_thread_start_response(&response, &config)
.map_err(anyhow::Error::msg)?;
(session_configured.thread_id, session_configured)
(response.thread.history_mode, session_configured)
};
let primary_thread_id = fallback_session_configured.thread_id;
if let Some(worktree) = managed_worktree.as_ref() {
worktree
.manager
@@ -1367,6 +1373,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
) {
maybe_backfill_turn_completed_items(
config.ephemeral,
history_mode,
&client,
&mut request_ids,
&mut notification,
@@ -1737,14 +1744,13 @@ fn should_process_notification(
async fn maybe_backfill_turn_completed_items(
thread_ephemeral: bool,
history_mode: ThreadHistoryMode,
client: &InProcessAppServerClient,
request_ids: &mut RequestIdSequencer,
notification: &mut ServerNotification,
) {
// In-process delivery may drop non-terminal item notifications under backpressure while still
// guaranteeing `turn/completed`. Because app-server currently emits that completion with an
// empty `turn.items`, exec does one last `thread/read` here so human/json output can recover
// the final message and reconcile any still-running items before shutdown.
// guaranteeing `turn/completed`. Recover the completed turn's items before shutdown.
if !should_backfill_turn_completed_items(thread_ephemeral, notification) {
return;
}
@@ -1753,6 +1759,53 @@ async fn maybe_backfill_turn_completed_items(
return;
};
if history_mode == ThreadHistoryMode::Paginated {
let completion_items = payload.turn.items.clone();
let mut items = Vec::new();
let mut cursor = None;
let mut seen_cursors = HashSet::new();
loop {
let page: ThreadItemsListResponse = match client
.request_typed(ClientRequest::ThreadItemsList {
request_id: request_ids.next(),
params: ThreadItemsListParams {
thread_id: payload.thread_id.clone(),
turn_id: Some(payload.turn.id.clone()),
cursor,
limit: None,
sort_direction: None,
},
})
.await
{
Ok(page) => page,
Err(err) => {
warn!("thread/items/list failed during turn completion: {err}");
return;
}
};
items.extend(page.data.into_iter().map(ThreadItem::from));
let Some(next) = page.next_cursor else { break };
if !seen_cursors.insert(next.clone()) {
warn!("thread/items/list returned a repeated cursor during turn completion");
return;
}
cursor = Some(ThreadItemsListCursor::Opaque(next));
}
for completion_item in completion_items {
if let Some(item) = items
.iter_mut()
.find(|item| item.id() == completion_item.id())
{
*item = completion_item;
} else {
items.push(completion_item);
}
}
payload.turn.items = items;
return;
}
let response = send_request_with_response::<ThreadReadResponse>(
client,
ClientRequest::ThreadRead {
codex-rs/exec/tests/suite/completion_backfill_tests.rs
+69 -6
View File
@@ -4,6 +4,9 @@ use core_test_support::test_codex_exec::test_codex_exec;
use pretty_assertions::assert_eq;
use serde_json::json;
const TRACE_REQUESTS: &str =
"codex_app_server::message_processor=trace,codex_app_server::outgoing_message=trace";
const PARENT_PROMPT: &str = "spawn a child and wait for it";
const CHILD_PROMPT: &str = "child: finish first";
const SPAWN_CALL_ID: &str = "spawn-call";
@@ -86,10 +89,7 @@ async fn ignores_unrelated_turn_completion_before_backfilling_primary_turn() ->
);
let output = test
.cmd()
.env(
"RUST_LOG",
"codex_app_server::message_processor=trace,codex_app_server::outgoing_message=trace",
)
.env("RUST_LOG", TRACE_REQUESTS)
.arg("--skip-git-repo-check")
.arg("--json")
.arg("-c")
@@ -138,7 +138,7 @@ async fn ignores_unrelated_turn_completion_before_backfilling_primary_turn() ->
.filter(|line| line.contains("app-server typed request"))
.count(),
0,
"the unrelated completion must not issue thread/read: {stderr}"
"the unrelated completion must not read history: {stderr}"
);
assert_eq!(
lines[*primary_completion + 1..]
@@ -146,7 +146,7 @@ async fn ignores_unrelated_turn_completion_before_backfilling_primary_turn() ->
.filter(|line| line.contains("app-server typed request"))
.count(),
2,
"the primary completion should issue thread/read and thread/unsubscribe: {stderr}"
"the primary completion should read its items and unsubscribe: {stderr}"
);
assert_eq!(
@@ -160,3 +160,66 @@ async fn ignores_unrelated_turn_completion_before_backfilling_primary_turn() ->
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_fork_recovers_only_its_own_turn() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let test = test_codex_exec();
let last_message = test.cwd_path().join("last-message");
let mut events = vec![responses::ev_response_created("parent")];
// The parent's final answer must survive recovery across multiple item pages.
for index in 0..120 {
events.push(responses::ev_assistant_message(
&format!("msg-{index}"),
"working",
));
}
events.push(responses::ev_assistant_message("answer", "parent done"));
events.push(responses::ev_completed("parent"));
let _parent_response = responses::mount_sse_once(&server, responses::sse(events)).await;
let parent = test
.cmd_with_server(&server)
.args(["--skip-git-repo-check", "--json", "--output-last-message"])
.arg(&last_message)
.arg("Produce a long turn")
.assert()
.success();
assert_eq!(std::fs::read_to_string(last_message)?, "parent done");
let stdout = std::str::from_utf8(&parent.get_output().stdout)?;
let started: serde_json::Value = serde_json::from_str(stdout.lines().next().unwrap())?;
let parent_id = started["thread_id"].as_str().expect("parent thread ID");
let error = wiremock::ResponseTemplate::new(400).set_body_json(json!({
"error": {"code": "model_not_found", "message": "Unknown model", "type": "invalid_request_error"}
}));
let _fork_response = responses::mount_response_once(&server, error).await;
let fork = test
.cmd_with_server(&server)
.env("RUST_LOG", TRACE_REQUESTS)
.args(["--skip-git-repo-check", "--json", "fork", parent_id, "Fail"])
.assert()
.code(1);
let stdout = std::str::from_utf8(&fork.get_output().stdout)?;
let terminal: serde_json::Value = serde_json::from_str(stdout.lines().last().unwrap())?;
assert_eq!(
terminal,
json!({"type": "turn.failed", "error": {"message": "Unknown model"}})
);
let stderr = std::str::from_utf8(&fork.get_output().stderr)?;
let (_, completion) = stderr
.split_once("app-server event: turn/completed")
.expect("fork completed");
assert!(
!completion.contains("app-server event: deprecationNotice"),
"{completion}"
);
assert_eq!(
completion.matches("app-server typed request").count(),
2,
"the failed turn needs one item page and an unsubscribe: {completion}"
);
Ok(())
}