From c92700d2993707ee16eb63b35a1cdd7fcc563f69 Mon Sep 17 00:00:00 2001 From: Brent Traut Date: Wed, 28 Oct 2026 17:53:37 +0000 Subject: [PATCH] Backfill exec completion items through paginated history (#57800) ## 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 --- codex-rs/exec/src/lib.rs | 69 +++++++++++++++-- .../tests/suite/completion_backfill_tests.rs | 75 +++++++++++++++++-- 2 files changed, 130 insertions(+), 14 deletions(-) diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 821c6f1fe4..43cf70b21d 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -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::( client, ClientRequest::ThreadRead { diff --git a/codex-rs/exec/tests/suite/completion_backfill_tests.rs b/codex-rs/exec/tests/suite/completion_backfill_tests.rs index 939fb1c695..68471b555a 100644 --- a/codex-rs/exec/tests/suite/completion_backfill_tests.rs +++ b/codex-rs/exec/tests/suite/completion_backfill_tests.rs @@ -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(()) +}