feat: update process ID for event handling (openai#7261)

Author: jif-oai

codex-rs/Cargo.lock

@@ -1020,6 +1020,8 @@ pub enum ThreadItem {
         command: String,
         /// The command's working directory.
         cwd: PathBuf,
+        /// Identifier for the underlying PTY process (when available).
+        process_id: Option<String>,
         status: CommandExecutionStatus,
         /// A best-effort parsing of the command to understand the action(s) it will perform.
         /// This returns a list of CommandAction objects because a single shell command may

codex-rs/app-server-protocol/src/protocol/v2.rs

@@ -449,11 +449,13 @@ pub(crate) async fn apply_bespoke_event_handling(
                 .collect::<Vec<_>>();
             let command = shlex_join(&exec_command_begin_event.command);
             let cwd = exec_command_begin_event.cwd;
+            let process_id = exec_command_begin_event.process_id;
 
             let item = ThreadItem::CommandExecution {
                 id: item_id,
                 command,
                 cwd,
+                process_id,
                 status: CommandExecutionStatus::InProgress,
                 command_actions,
                 aggregated_output: None,
@@ -486,6 +488,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                 command,
                 cwd,
                 parsed_cmd,
+                process_id,
                 aggregated_output,
                 exit_code,
                 duration,
@@ -514,6 +517,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                 id: call_id,
                 command: shlex_join(&command),
                 cwd,
+                process_id,
                 status,
                 command_actions,
                 aggregated_output,
@@ -649,6 +653,7 @@ async fn complete_command_execution_item(
     item_id: String,
     command: String,
     cwd: PathBuf,
+    process_id: Option<String>,
     command_actions: Vec<V2ParsedCommand>,
     status: CommandExecutionStatus,
     outgoing: &OutgoingMessageSender,
@@ -657,6 +662,7 @@ async fn complete_command_execution_item(
         id: item_id,
         command,
         cwd,
+        process_id,
         status,
         command_actions,
         aggregated_output: None,
@@ -1015,6 +1021,7 @@ async fn on_command_execution_request_approval_response(
             item_id.clone(),
             command.clone(),
             cwd.clone(),
+            None,
             command_actions.clone(),
             status,
             outgoing.as_ref(),

codex-rs/app-server/src/bespoke_event_handling.rs

@@ -15,6 +15,7 @@ pub use mcp_process::McpProcess;
 pub use mock_model_server::create_mock_chat_completions_server;
 pub use mock_model_server::create_mock_chat_completions_server_unchecked;
 pub use responses::create_apply_patch_sse_response;
+pub use responses::create_exec_command_sse_response;
 pub use responses::create_final_assistant_message_sse_response;
 pub use responses::create_shell_command_sse_response;
 pub use rollout::create_fake_rollout;

codex-rs/app-server/tests/common/lib.rs

@@ -94,3 +94,42 @@ pub fn create_apply_patch_sse_response(
     );
     Ok(sse)
 }
+
+pub fn create_exec_command_sse_response(call_id: &str) -> anyhow::Result<String> {
+    let (cmd, args) = if cfg!(windows) {
+        ("cmd.exe", vec!["/d", "/c", "echo hi"])
+    } else {
+        ("/bin/sh", vec!["-c", "echo hi"])
+    };
+    let command = std::iter::once(cmd.to_string())
+        .chain(args.into_iter().map(str::to_string))
+        .collect::<Vec<_>>();
+    let tool_call_arguments = serde_json::to_string(&json!({
+        "cmd": command.join(" "),
+        "yield_time_ms": 500
+    }))?;
+    let tool_call = json!({
+        "choices": [
+            {
+                "delta": {
+                    "tool_calls": [
+                        {
+                            "id": call_id,
+                            "function": {
+                                "name": "exec_command",
+                                "arguments": tool_call_arguments
+                            }
+                        }
+                    ]
+                },
+                "finish_reason": "tool_calls"
+            }
+        ]
+    });
+
+    let sse = format!(
+        "data: {}\n\ndata: DONE\n\n",
+        serde_json::to_string(&tool_call)?
+    );
+    Ok(sse)
+}

codex-rs/app-server/tests/common/responses.rs

@@ -1,6 +1,7 @@
 use anyhow::Result;
 use app_test_support::McpProcess;
 use app_test_support::create_apply_patch_sse_response;
+use app_test_support::create_exec_command_sse_response;
 use app_test_support::create_final_assistant_message_sse_response;
 use app_test_support::create_mock_chat_completions_server;
 use app_test_support::create_mock_chat_completions_server_unchecked;
@@ -907,6 +908,134 @@ async fn turn_start_file_change_approval_decline_v2() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+#[cfg_attr(windows, ignore = "process id reporting differs on Windows")]
+async fn command_execution_notifications_include_process_id() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let responses = vec![
+        create_exec_command_sse_response("uexec-1")?,
+        create_final_assistant_message_sse_response("done")?,
+    ];
+    let server = create_mock_chat_completions_server(responses).await;
+    let codex_home = TempDir::new()?;
+    create_config_toml(codex_home.path(), &server.uri(), "never")?;
+    let config_toml = codex_home.path().join("config.toml");
+    let mut config_contents = std::fs::read_to_string(&config_toml)?;
+    config_contents.push_str(
+        r#"
+[features]
+unified_exec = true
+"#,
+    );
+    std::fs::write(&config_toml, config_contents)?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let start_id = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let start_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
+
+    let turn_id = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: "run a command".to_string(),
+            }],
+            ..Default::default()
+        })
+        .await?;
+    let turn_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(turn_id)),
+    )
+    .await??;
+    let TurnStartResponse { turn: _turn } = to_response::<TurnStartResponse>(turn_resp)?;
+
+    let started_command = timeout(DEFAULT_READ_TIMEOUT, async {
+        loop {
+            let notif = mcp
+                .read_stream_until_notification_message("item/started")
+                .await?;
+            let started: ItemStartedNotification = serde_json::from_value(
+                notif
+                    .params
+                    .clone()
+                    .expect("item/started should include params"),
+            )?;
+            if let ThreadItem::CommandExecution { .. } = started.item {
+                return Ok::<ThreadItem, anyhow::Error>(started.item);
+            }
+        }
+    })
+    .await??;
+    let ThreadItem::CommandExecution {
+        id,
+        process_id: started_process_id,
+        status,
+        ..
+    } = started_command
+    else {
+        unreachable!("loop ensures we break on command execution items");
+    };
+    assert_eq!(id, "uexec-1");
+    assert_eq!(status, CommandExecutionStatus::InProgress);
+    let started_process_id = started_process_id.expect("process id should be present");
+
+    let completed_command = timeout(DEFAULT_READ_TIMEOUT, async {
+        loop {
+            let notif = mcp
+                .read_stream_until_notification_message("item/completed")
+                .await?;
+            let completed: ItemCompletedNotification = serde_json::from_value(
+                notif
+                    .params
+                    .clone()
+                    .expect("item/completed should include params"),
+            )?;
+            if let ThreadItem::CommandExecution { .. } = completed.item {
+                return Ok::<ThreadItem, anyhow::Error>(completed.item);
+            }
+        }
+    })
+    .await??;
+    let ThreadItem::CommandExecution {
+        id: completed_id,
+        process_id: completed_process_id,
+        status: completed_status,
+        exit_code,
+        ..
+    } = completed_command
+    else {
+        unreachable!("loop ensures we break on command execution items");
+    };
+    assert_eq!(completed_id, "uexec-1");
+    assert_eq!(completed_status, CommandExecutionStatus::Completed);
+    assert_eq!(exit_code, Some(0));
+    assert_eq!(
+        completed_process_id.as_deref(),
+        Some(started_process_id.as_str())
+    );
+
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+
+    Ok(())
+}
+
 // Helper to create a config.toml pointing at the mock model server.
 fn create_config_toml(
     codex_home: &Path,

codex-rs/app-server/tests/suite/v2/turn_start.rs

@@ -87,6 +87,9 @@ uuid = { workspace = true, features = ["serde", "v4", "v5"] }
 which = { workspace = true }
 wildmatch = { workspace = true }
 
+[features]
+deterministic_process_ids = []
+
 
 [target.'cfg(target_os = "linux")'.dependencies]
 landlock = { workspace = true }
@@ -115,6 +118,7 @@ keyring = { workspace = true, features = ["sync-secret-service"] }
 assert_cmd = { workspace = true }
 assert_matches = { workspace = true }
 codex-arg0 = { workspace = true }
+codex-core = { path = ".", features = ["deterministic_process_ids"] }
 core_test_support = { workspace = true }
 ctor = { workspace = true }
 escargot = { workspace = true }

codex-rs/core/Cargo.toml

@@ -81,6 +81,7 @@ impl SessionTask for UserShellCommandTask {
                 turn_context.as_ref(),
                 EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
                     call_id: call_id.clone(),
+                    process_id: None,
                     turn_id: turn_context.sub_id.clone(),
                     command: command.clone(),
                     cwd: cwd.clone(),
@@ -139,6 +140,7 @@ impl SessionTask for UserShellCommandTask {
                         turn_context.as_ref(),
                         EventMsg::ExecCommandEnd(ExecCommandEndEvent {
                             call_id,
+                            process_id: None,
                             turn_id: turn_context.sub_id.clone(),
                             command: command.clone(),
                             cwd: cwd.clone(),
@@ -161,6 +163,7 @@ impl SessionTask for UserShellCommandTask {
                         turn_context.as_ref(),
                         EventMsg::ExecCommandEnd(ExecCommandEndEvent {
                             call_id: call_id.clone(),
+                            process_id: None,
                             turn_id: turn_context.sub_id.clone(),
                             command: command.clone(),
                             cwd: cwd.clone(),
@@ -205,6 +208,7 @@ impl SessionTask for UserShellCommandTask {
                         turn_context.as_ref(),
                         EventMsg::ExecCommandEnd(ExecCommandEndEvent {
                             call_id,
+                            process_id: None,
                             turn_id: turn_context.sub_id.clone(),
                             command,
                             cwd,