Backfill completion items only for the active exec turn (#34226)

## Why

Multi-agent exec sessions can receive child `turn/completed` notifications on
the same event stream as the primary turn. Backfilling those unrelated
completions issues unnecessary `thread/read` requests.

## What changed

Run completion-item backfill only after a notification has been matched to the
primary exec thread and turn.

## Testing

Add an integration test that delivers a child completion before the primary
completion and verifies that only the primary completion triggers backfill.

GitOrigin-RevId: 46a41236ee6a39e4b341570ff4826a71fa02bb32

codex-rs/exec/src/lib.rs

@@ -1014,19 +1014,19 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
                     error_seen = true;
                 }
 
-                maybe_backfill_turn_completed_items(
-                    config.ephemeral,
-                    &client,
-                    &mut request_ids,
-                    &mut notification,
-                )
-                .await;
-
                 if should_process_notification(
                     &notification,
                     &primary_thread_id_for_requests,
                     &task_id,
                 ) {
+                    maybe_backfill_turn_completed_items(
+                        config.ephemeral,
+                        &client,
+                        &mut request_ids,
+                        &mut notification,
+                    )
+                    .await;
+
                     match event_processor.process_server_notification(notification) {
                         CodexStatus::Running => {}
                         CodexStatus::InitiateShutdown => {

codex-rs/exec/tests/suite/completion_backfill_tests.rs

@@ -0,0 +1,162 @@
+use core_test_support::responses;
+use core_test_support::skip_if_no_network;
+use core_test_support::test_codex_exec::test_codex_exec;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+
+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";
+const WAIT_CALL_ID: &str = "wait-call";
+
+fn body_contains(request: &wiremock::Request, text: &str) -> bool {
+    std::str::from_utf8(&request.body).is_ok_and(|body| body.contains(text))
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn ignores_unrelated_turn_completion_before_backfilling_primary_turn() -> anyhow::Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let server = responses::start_mock_server().await;
+    let spawn_args = json!({
+        "message": CHILD_PROMPT,
+        "task_name": "worker",
+        "fork_turns": "none",
+    })
+    .to_string();
+    let _parent_turn = responses::mount_sse_once_match(
+        &server,
+        |request: &wiremock::Request| body_contains(request, PARENT_PROMPT),
+        responses::sse(vec![
+            responses::ev_response_created("resp-parent-1"),
+            responses::ev_function_call_with_namespace(
+                SPAWN_CALL_ID,
+                "collaboration",
+                "spawn_agent",
+                &spawn_args,
+            ),
+            responses::ev_completed("resp-parent-1"),
+        ]),
+    )
+    .await;
+    let _child_turn = responses::mount_sse_once_match(
+        &server,
+        |request: &wiremock::Request| {
+            body_contains(request, CHILD_PROMPT) && !body_contains(request, SPAWN_CALL_ID)
+        },
+        responses::sse(vec![
+            responses::ev_response_created("resp-child"),
+            responses::ev_assistant_message("msg-child", "child done"),
+            responses::ev_completed("resp-child"),
+        ]),
+    )
+    .await;
+    let _parent_wait = responses::mount_sse_once_match(
+        &server,
+        |request: &wiremock::Request| {
+            body_contains(request, SPAWN_CALL_ID) && !body_contains(request, WAIT_CALL_ID)
+        },
+        responses::sse(vec![
+            responses::ev_response_created("resp-parent-2"),
+            responses::ev_function_call_with_namespace(
+                WAIT_CALL_ID,
+                "collaboration",
+                "wait_agent",
+                "{}",
+            ),
+            responses::ev_completed("resp-parent-2"),
+        ]),
+    )
+    .await;
+    let _parent_completion = responses::mount_sse_once_match(
+        &server,
+        |request: &wiremock::Request| body_contains(request, WAIT_CALL_ID),
+        responses::sse(vec![
+            responses::ev_response_created("resp-parent-3"),
+            responses::ev_assistant_message("msg-parent", "parent done"),
+            responses::ev_completed("resp-parent-3"),
+        ]),
+    )
+    .await;
+
+    let test = test_codex_exec();
+    let mock_provider = format!(
+        "model_providers.mock_provider={{name=\"Mock provider for test\",base_url=\"{}/v1\",wire_api=\"responses\",supports_websockets=false}}",
+        server.uri()
+    );
+    let output = test
+        .cmd()
+        .env(
+            "RUST_LOG",
+            "codex_app_server::message_processor=trace,codex_app_server::outgoing_message=trace",
+        )
+        .arg("--skip-git-repo-check")
+        .arg("--json")
+        .arg("-c")
+        .arg(mock_provider)
+        .args([
+            "-c",
+            "model_provider=\"mock_provider\"",
+            "-c",
+            "features.multi_agent=true",
+            "-c",
+            "features.multi_agent_v2=true",
+            "-c",
+            "features.enable_request_compression=false",
+        ])
+        .arg(PARENT_PROMPT)
+        .output()?;
+    assert!(output.status.success(), "exec run failed: {output:?}");
+
+    let stdout = String::from_utf8(output.stdout)?;
+    assert!(
+        stdout.contains("parent done"),
+        "primary completion was not processed: {stdout}"
+    );
+    let stderr = String::from_utf8(output.stderr)?;
+    let lines = stderr.lines().collect::<Vec<_>>();
+    let turn_completions = lines
+        .iter()
+        .enumerate()
+        .filter_map(|(index, line)| {
+            line.contains("app-server event: turn/completed")
+                .then_some(index)
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(
+        turn_completions.len(),
+        2,
+        "expected the child completion before the primary completion: {stderr}"
+    );
+
+    let [child_completion, primary_completion] = turn_completions.as_slice() else {
+        unreachable!("checked turn/completed count")
+    };
+    assert_eq!(
+        lines[*child_completion + 1..*primary_completion]
+            .iter()
+            .filter(|line| line.contains("app-server typed request"))
+            .count(),
+        0,
+        "the unrelated completion must not issue thread/read: {stderr}"
+    );
+    assert_eq!(
+        lines[*primary_completion + 1..]
+            .iter()
+            .filter(|line| line.contains("app-server typed request"))
+            .count(),
+        2,
+        "the primary completion should issue thread/read and thread/unsubscribe: {stderr}"
+    );
+
+    assert_eq!(
+        lines
+            .iter()
+            .filter(|line| line.contains("app-server typed request"))
+            .count(),
+        5,
+        "only the primary completion should issue an extra request: {stderr}"
+    );
+
+    Ok(())
+}

codex-rs/exec/tests/suite/mod.rs

@@ -4,6 +4,8 @@ mod agents_md;
 mod apply_patch;
 mod approval_policy;
 mod auth_env;
+#[path = "completion_backfill_tests.rs"]
+mod completion_backfill;
 mod ephemeral;
 mod hooks;
 mod mcp_required_exit;