Add opt-in protocol discovery for local stdio MCP servers (#57210)

## Why

Stdio servers without `CODEX_MCP_PROTOCOL_VERSION` use the legacy handshake, even when they support MCP `2026-07-28`.

## What changed

- Add the disabled-by-default `mcp_connection_sharing` feature to enable discovery for eligible local stdio servers.
- Recover from failed or late discovery with a fresh handshake when needed, waiting for the previous helper to release resources before restarting.
- Retain a working legacy connection when an upgrade attempt fails for the same account, except on authentication errors.
- Keep tool catalogs separate by protocol and invalidate legacy catalogs when discovery negotiates the modern protocol.

## Testing

Add regression coverage for failed and late discovery, legacy fallback, protocol-specific catalog reuse, message limits, and Windows descendant cleanup.

GitOrigin-RevId: 36c675ef3b81d32fccf06ccbddda2312cfe602c9

codex-rs/utils/pty/src/process_group.rs

@@ -288,6 +288,33 @@ pub fn kill_process_group(_process_group_id: u32) -> io::Result<()> {
     Ok(())
 }
 
+/// Wait up to `grace` for a signaled group's members to release their resources.
+/// After that, allow the caller to proceed. An unreaped zombie or a recycled group ID
+/// can remain visible after the original processes have released their resources.
+#[cfg(unix)]
+pub fn wait_for_process_group_exit(process_group_id: u32, grace: std::time::Duration) {
+    let deadline = std::time::Instant::now() + grace;
+    loop {
+        #[cfg(target_os = "macos")]
+        let exists = signal_process_group_with_member_fallback(
+            process_group_id,
+            /*signal*/ 0,
+            signal_process_group_id,
+            signal_process_id,
+        );
+        #[cfg(not(target_os = "macos"))]
+        let exists = signal_process_group_id(process_group_id as libc::pid_t, /*signal*/ 0);
+        if matches!(exists, Ok(false)) {
+            break;
+        }
+        let remaining = deadline.saturating_duration_since(std::time::Instant::now());
+        if remaining.is_zero() {
+            break;
+        }
+        std::thread::sleep(remaining.min(std::time::Duration::from_millis(10)));
+    }
+}
+
 #[cfg(unix)]
 /// Kill the process group for a tokio child (best-effort).
 pub fn kill_child_process_group(child: &mut Child) -> io::Result<()> {

codex-rs/utils/pty/src/win/job.rs

@@ -1,29 +1,44 @@
 use filedescriptor::OwnedHandle;
+use std::collections::HashMap;
 use std::io;
 use std::os::windows::io::AsRawHandle;
 use std::os::windows::io::FromRawHandle;
+use std::os::windows::io::OwnedHandle as ProcessHandle;
 use std::os::windows::io::RawHandle;
 use std::sync::Mutex;
+use std::time::Duration;
+use std::time::Instant;
 use tokio::process::Child;
 use tokio::process::Command;
 use winapi::shared::ntdef::NT_SUCCESS;
 use winapi::shared::ntdef::NTSTATUS;
+use winapi::shared::winerror::ERROR_INVALID_PARAMETER;
+use winapi::shared::winerror::ERROR_MORE_DATA;
+use winapi::shared::winerror::WAIT_TIMEOUT;
+use winapi::um::jobapi::IsProcessInJob;
 use winapi::um::jobapi2::AssignProcessToJobObject;
 use winapi::um::jobapi2::CreateJobObjectW;
+use winapi::um::jobapi2::QueryInformationJobObject;
 use winapi::um::jobapi2::SetInformationJobObject;
 use winapi::um::jobapi2::TerminateJobObject;
 use winapi::um::processthreadsapi::OpenProcess;
 use winapi::um::processthreadsapi::TerminateProcess;
+use winapi::um::synchapi::WaitForSingleObject;
 use winapi::um::winbase::CREATE_NO_WINDOW;
 use winapi::um::winbase::CREATE_SUSPENDED;
+use winapi::um::winbase::WAIT_OBJECT_0;
 use winapi::um::winnt::HANDLE;
 use winapi::um::winnt::JOB_OBJECT_LIMIT_BREAKAWAY_OK;
 use winapi::um::winnt::JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
+use winapi::um::winnt::JOBOBJECT_BASIC_PROCESS_ID_LIST;
 use winapi::um::winnt::JOBOBJECT_EXTENDED_LIMIT_INFORMATION;
+use winapi::um::winnt::JobObjectBasicProcessIdList;
 use winapi::um::winnt::JobObjectExtendedLimitInformation;
+use winapi::um::winnt::PROCESS_QUERY_LIMITED_INFORMATION;
 use winapi::um::winnt::PROCESS_SET_QUOTA;
 use winapi::um::winnt::PROCESS_SUSPEND_RESUME;
 use winapi::um::winnt::PROCESS_TERMINATE;
+use winapi::um::winnt::SYNCHRONIZE;
 
 #[link(name = "ntdll")]
 unsafe extern "system" {
@@ -39,6 +54,19 @@ pub struct JobObject {
     preserve_descendants: Mutex<bool>,
 }
 
+// A failed restart must still stop any remaining members.
+struct TerminateOnDrop<'a>(Option<&'a JobObject>);
+
+impl Drop for TerminateOnDrop<'_> {
+    fn drop(&mut self) {
+        if let Some(job) = self.0
+            && let Err(error) = job.terminate()
+        {
+            log::warn!("Failed to terminate Windows job after an interrupted restart: {error}");
+        }
+    }
+}
+
 impl JobObject {
     /// Creates a Job Object configured to terminate all members when its last handle closes.
     pub fn create() -> io::Result<Self> {
@@ -67,19 +95,23 @@ impl JobObject {
     }
 
     /// Captures an owned process handle before its numeric identifier can be reused.
-    pub fn open_process_handle(process_id: u32) -> io::Result<std::os::windows::io::OwnedHandle> {
+    pub fn open_process_handle(process_id: u32) -> io::Result<ProcessHandle> {
+        Self::open_process(process_id, PROCESS_TERMINATE)
+    }
+
+    fn open_process(process_id: u32, access: u32) -> io::Result<ProcessHandle> {
         let handle = unsafe {
-            OpenProcess(PROCESS_TERMINATE, /*bInheritHandle*/ 0, process_id)
+            OpenProcess(access, /*bInheritHandle*/ 0, process_id)
         };
         if handle.is_null() {
             return Err(io::Error::last_os_error());
         }
 
-        Ok(unsafe { std::os::windows::io::OwnedHandle::from_raw_handle(handle.cast()) })
+        Ok(unsafe { ProcessHandle::from_raw_handle(handle.cast()) })
     }
 
     /// Terminates the exact process identified by a previously captured handle.
-    pub fn terminate_process_handle(handle: &std::os::windows::io::OwnedHandle) -> io::Result<()> {
+    pub fn terminate_process_handle(handle: &ProcessHandle) -> io::Result<()> {
         let terminated = unsafe {
             TerminateProcess(handle.as_raw_handle().cast(), /*uExitCode*/ 1)
         };
@@ -132,22 +164,15 @@ impl JobObject {
             .kill_on_drop(true);
     }
 
-    /// Assigns and resumes a suspended child, returning whether assignment succeeded.
+    /// Assigns and resumes a suspended child, returning its handle if containment succeeded.
     ///
     /// Nested jobs can reject assignment. Such a child is resumed without
     /// containment so callers can preserve their existing compatibility fallback.
-    pub fn assign_and_resume_process(&self, process_id: u32) -> io::Result<bool> {
-        let process = unsafe {
-            OpenProcess(
-                PROCESS_SET_QUOTA | PROCESS_TERMINATE | PROCESS_SUSPEND_RESUME,
-                /*bInheritHandle*/ 0,
-                process_id,
-            )
-        };
-        if process.is_null() {
-            return Err(io::Error::last_os_error());
-        }
-        let process = unsafe { OwnedHandle::from_raw_handle(process.cast()) };
+    pub fn assign_and_resume_process(&self, process_id: u32) -> io::Result<Option<ProcessHandle>> {
+        let process = Self::open_process(
+            process_id,
+            PROCESS_SET_QUOTA | PROCESS_TERMINATE | PROCESS_SUSPEND_RESUME | SYNCHRONIZE,
+        )?;
         let assignment = self.assign_process(process.as_raw_handle());
 
         let status = unsafe { NtResumeProcess(process.as_raw_handle().cast()) };
@@ -161,12 +186,12 @@ impl JobObject {
         }
 
         match assignment {
-            Ok(()) => Ok(true),
+            Ok(()) => Ok(Some(process)),
             Err(error) => {
                 log::warn!(
                     "Windows process job assignment unavailable for pid {process_id}: {error}"
                 );
-                Ok(false)
+                Ok(None)
             }
         }
     }
@@ -251,6 +276,113 @@ impl JobObject {
             Ok(())
         }
     }
+
+    /// Stops a root from [`Self::assign_and_resume_process`] and the remaining members of its job.
+    pub fn terminate_and_wait(&self, root: &ProcessHandle, grace: Duration) -> io::Result<()> {
+        let mut cleanup = TerminateOnDrop(Some(self));
+        let deadline = Instant::now() + grace;
+        Self::request_stop(root)?;
+        Self::wait_for_exit(root, deadline)?;
+        // TerminateJobObject can empty the job list before members release their handles.
+        // Stop by process handle, then rescan for children born during the previous pass.
+        let mut list = vec![0usize; 16];
+        let mut stopped = HashMap::new();
+        loop {
+            if Instant::now() >= deadline {
+                return Err(io::ErrorKind::TimedOut.into());
+            }
+            let pids = self.member_ids(&mut list)?;
+            if pids.iter().all(|pid| stopped.contains_key(pid)) {
+                self.terminate()?;
+                cleanup.0 = None;
+                return Ok(());
+            }
+            for &pid in pids {
+                if stopped.contains_key(&pid) {
+                    continue;
+                }
+                let member = match Self::open_process(
+                    pid as u32,
+                    PROCESS_TERMINATE | PROCESS_QUERY_LIMITED_INFORMATION | SYNCHRONIZE,
+                ) {
+                    Ok(member) => member,
+                    Err(error) if error.raw_os_error() == Some(ERROR_INVALID_PARAMETER as i32) => {
+                        continue;
+                    }
+                    Err(error) => return Err(error),
+                };
+                // A listed PID can be reused before OpenProcess. Only signal confirmed job members.
+                let mut in_job = 0;
+                if unsafe {
+                    IsProcessInJob(
+                        member.as_raw_handle().cast(),
+                        self.handle.as_raw_handle().cast(),
+                        &mut in_job,
+                    )
+                } == 0
+                {
+                    return Err(io::Error::last_os_error());
+                }
+                if in_job != 0 {
+                    Self::request_stop(&member)?;
+                    stopped.insert(pid, member);
+                }
+            }
+            for member in stopped.values() {
+                Self::wait_for_exit(member, deadline)?;
+            }
+        }
+    }
+
+    fn member_ids<'a>(&self, list: &'a mut Vec<usize>) -> io::Result<&'a [usize]> {
+        const HEADER_WORDS: usize =
+            std::mem::offset_of!(JOBOBJECT_BASIC_PROCESS_ID_LIST, ProcessIdList)
+                / std::mem::size_of::<usize>();
+        loop {
+            let listed = unsafe {
+                QueryInformationJobObject(
+                    self.handle.as_raw_handle().cast(),
+                    JobObjectBasicProcessIdList,
+                    list.as_mut_ptr().cast(),
+                    u32::try_from(std::mem::size_of_val(list.as_slice()))
+                        .map_err(io::Error::other)?,
+                    std::ptr::null_mut(),
+                )
+            };
+            if listed != 0 {
+                let header = unsafe { &*list.as_ptr().cast::<JOBOBJECT_BASIC_PROCESS_ID_LIST>() };
+                if header.NumberOfProcessIdsInList == header.NumberOfAssignedProcesses {
+                    return Ok(&list[HEADER_WORDS..][..header.NumberOfProcessIdsInList as usize]);
+                }
+            } else {
+                let error = io::Error::last_os_error();
+                if error.raw_os_error() != Some(ERROR_MORE_DATA as i32) {
+                    return Err(error);
+                }
+            }
+            list.resize(list.len() * 2, 0);
+        }
+    }
+
+    fn request_stop(process: &ProcessHandle) -> io::Result<()> {
+        // Windows also returns access denied when the process has already exited. Its handle
+        // must still signal before a replacement can start.
+        match Self::terminate_process_handle(process) {
+            Err(error) if error.kind() == io::ErrorKind::PermissionDenied => Ok(()),
+            result => result,
+        }
+    }
+
+    fn wait_for_exit(process: &ProcessHandle, deadline: Instant) -> io::Result<()> {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        // The largest DWORD denotes INFINITE, so cap this bounded wait below it.
+        let millis = remaining.as_millis().min(u128::from(u32::MAX - 1)) as u32;
+        match unsafe { WaitForSingleObject(process.as_raw_handle().cast(), millis) } {
+            WAIT_OBJECT_0 => Ok(()),
+            WAIT_TIMEOUT => Err(io::ErrorKind::TimedOut.into()),
+            _ => Err(io::Error::last_os_error()),
+        }
+    }
 }
 
 impl AsRawHandle for JobObject {

codex-rs/utils/pty/src/windows_tests.rs

@@ -8,6 +8,7 @@ use crate::spawn_pty_process;
 use pretty_assertions::assert_eq;
 use std::collections::HashMap;
 use std::os::windows::io::AsRawHandle;
+use std::os::windows::io::BorrowedHandle;
 use std::os::windows::io::FromRawHandle;
 use std::os::windows::io::OwnedHandle;
 use std::os::windows::process::CommandExt;
@@ -20,7 +21,10 @@ use tokio::io::BufReader;
 use tokio::process::Command;
 use winapi::um::jobapi::IsProcessInJob;
 use winapi::um::processthreadsapi::OpenProcess;
+use winapi::um::synchapi::WaitForSingleObject;
+use winapi::um::winbase::WAIT_OBJECT_0;
 use winapi::um::winnt::PROCESS_QUERY_LIMITED_INFORMATION;
+use winapi::um::winnt::SYNCHRONIZE;
 
 const READY_MARKER: &str = "__CODEX_CHILD_READY__";
 const VALUE_MARKER: &str = "__CODEX_CHILD_VALUE__";
@@ -67,7 +71,10 @@ async fn piped_child_has_no_console_with_or_without_job_containment() -> anyhow:
             let pid = child
                 .id()
                 .ok_or_else(|| anyhow::anyhow!("missing child pid"))?;
-            anyhow::ensure!(job.assign_and_resume_process(pid)?, "job assignment failed");
+            anyhow::ensure!(
+                job.assign_and_resume_process(pid)?.is_some(),
+                "job assignment failed"
+            );
         }
         child
             .stdin
@@ -289,7 +296,13 @@ async fn contained_spawn_owns_immediate_descendant() -> anyhow::Result<()> {
     tokio::time::timeout(Duration::from_secs(10), stdout.read_line(&mut child_pid)).await??;
     let child_pid: u32 = child_pid.trim().parse()?;
 
-    let process = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, child_pid) };
+    let process = unsafe {
+        OpenProcess(
+            PROCESS_QUERY_LIMITED_INFORMATION | SYNCHRONIZE,
+            /*bInheritHandle*/ 0,
+            child_pid,
+        )
+    };
     anyhow::ensure!(!process.is_null(), "failed to open immediate child process");
     let process = unsafe { OwnedHandle::from_raw_handle(process.cast()) };
     let mut in_job = 0;
@@ -303,7 +316,15 @@ async fn contained_spawn_owns_immediate_descendant() -> anyhow::Result<()> {
     anyhow::ensure!(checked != 0, "failed to inspect child Job Object");
     anyhow::ensure!(in_job != 0, "immediate child escaped its Job Object");
 
-    job.terminate()?;
+    let root_handle = root
+        .raw_handle()
+        .ok_or_else(|| anyhow::anyhow!("missing contained root handle"))?;
+    let root_handle = unsafe { BorrowedHandle::borrow_raw(root_handle) }.try_clone_to_owned()?;
+    job.terminate_and_wait(&root_handle, Duration::from_secs(10))?;
+    let child_status = unsafe {
+        WaitForSingleObject(process.as_raw_handle().cast(), /*dwMilliseconds*/ 0)
+    };
+    assert_eq!(child_status, WAIT_OBJECT_0);
     tokio::time::timeout(Duration::from_secs(10), root.wait()).await??;
     Ok(())
 }
@@ -346,7 +367,9 @@ async fn rejected_job_assignment_resumes_existing_job_member() -> anyhow::Result
         .ok_or_else(|| anyhow::anyhow!("missing suspended process id"))?;
 
     assert!(
-        !rejected_job.assign_and_resume_process(process_id)?,
+        rejected_job
+            .assign_and_resume_process(process_id)?
+            .is_none(),
         "unrelated nested job unexpectedly accepted the process"
     );
     let stdout = root