Align profiler bindings and capability checks with Python 3.15 (#9340)

* Align offline profiler argument binding with Python 3.15

Accept the canonical keyword names for binary sample writes and native
context exits. Resolve the argument layout before numeric callbacks,
then preserve float-subclass payloads while accepting numeric protocols
for required and optional profile statistics.

AI-assisted source extraction, implementation, and source review with
OpenAI Codex. Runtime validation remains pending.

Assisted-by: OpenAI Codex:model version unavailable

* Restore the offline profiler capability flag

Preserve the original mega branch's zero-valued process_vm_readv
capability export. The native live unwinder remains unsupported; this
restores the public feature query without claiming live inspection support
or changing canonical test helpers.

AI-assisted source preservation and review with OpenAI Codex.

Assisted-by: OpenAI Codex:model version unavailable

* Restore finite-EOF table bounds for offline profiles

Match CPython's EOF-wide table count and decoding windows while retaining
checked offsets, allocation limits, cursor bounds, and replay checks.
This restores the original mega's finite-file contract without changing
canonical tests or the already validated argument-binding implementation.

Assisted-by: OpenAI Codex:model version unavailable

* Preserve symtable warning exceptions and original filenames (#9282)

Keep warning-hook exceptions untouched instead of rewriting SyntaxError
subclasses. Thread the original Python filename through parser warning
emission and fresh warning-as-error SyntaxError construction, preserving
fsdecoded bytes paths and lone surrogates.

Assisted-by: OpenAI Codex:model-version-unavailable

* build(deps): align the Cranelift 0.136 dependency family (#9341)

* build(deps): bump the wasmtime group across 1 directory with 2 updates

Bumps the wasmtime group with 2 updates in the / directory: [cranelift](https://github.com/bytecodealliance/wasmtime) and [cranelift-module](https://github.com/bytecodealliance/wasmtime).


Updates `cranelift` from 0.135.2 to 0.136.3
- [Release notes](https://github.com/bytecodealliance/wasmtime/releases)
- [Changelog](https://github.com/bytecodealliance/wasmtime/blob/main/RELEASES.md)
- [Commits](https://github.com/bytecodealliance/wasmtime/commits)

Updates `cranelift-module` from 0.135.3 to 0.136.3
- [Release notes](https://github.com/bytecodealliance/wasmtime/releases)
- [Changelog](https://github.com/bytecodealliance/wasmtime/blob/main/RELEASES.md)
- [Commits](https://github.com/bytecodealliance/wasmtime/commits)

---
updated-dependencies:
- dependency-name: cranelift
  dependency-version: 0.136.3
  dependency-type: direct:production
  update-type: version-update:semver-minor
  dependency-group: wasmtime
- dependency-name: cranelift-module
  dependency-version: 0.136.3
  dependency-type: direct:production
  update-type: version-update:semver-minor
  dependency-group: wasmtime
...

Signed-off-by: dependabot[bot] <support@github.com>

* Align cranelift-jit with the 0.136 dependency family

Update cranelift-jit alongside cranelift and cranelift-module so the JIT shares their LibCall, FuncId, and Module trait identities. Regenerate only the affected dependency lock entries, preserving the original dependency PR changes.

AI assistance covered the dependency diagnosis, alignment, and static review.

Assisted-by: OpenAI Codex:model-version-unavailable

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* Support selective canonical function migration

Add hash-pinned function selection to the official Lib migration tool,
preserving canonical source bytes and rejecting ambiguous dependencies.

Assisted-by: OpenAI Codex:model version unavailable

* Detect CPython runtimes for canonical inspection feature checks

Add native runtime-cookie detection and import the exact Python 3.15
inspection helpers and tests through the official Lib migration tool.
Preserve the offline profiler argument and finite-EOF corrections.

Assisted-by: OpenAI Codex:model version unavailable

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Cargo.lock

@@ -3859,11 +3859,13 @@ dependencies = [
  "libffi",
  "libloading 0.9.0",
  "mac_address",
+ "mach2",
  "memchr",
  "memmap2",
  "nix",
  "num-traits",
  "num_cpus",
+ "object",
  "parking_lot",
  "paste",
  "pkcs8 0.11.0",

Cargo.toml

@@ -259,6 +259,7 @@ log = "0.4.30"
 lz4_flex = "0.14"
 nix = { version = "0.31", features = ["fs", "user", "process", "term", "time", "signal", "ioctl", "socket", "sched", "zerocopy", "dir", "hostname", "net", "poll"] }
 mac_address = "1.1.3"
+mach2 = "0.4.3"
 malachite-bigint = "0.14.0"
 malachite-q = "0.14.0"
 malachite-base = "0.14.0"
@@ -271,6 +272,7 @@ num-integer = "0.1.46"
 num-traits = "0.2"
 num_cpus = "1.17.0"
 num_enum = { version = "0.7", default-features = false }
+object = { version = "0.37.3", default-features = false, features = ["read_core", "elf", "macho", "pe", "std", "unaligned"] }
 openssl = "0.10.80"
 openssl-sys = "0.9.110"
 optional = "0.5"

Lib/test/support/__init__.py

@@ -22,6 +22,7 @@ import warnings
 
 
 __all__ = [
+    "has_remote_subprocess_debugging", "requires_remote_subprocess_debugging",
     "requires_root_user", "requires_non_root_user",
     # globals
     "PIPE_MAX_SIZE", "verbose", "max_memuse", "use_resources", "failfast",
@@ -657,6 +658,92 @@ def requires_working_socket(*, module=False):
     else:
         return unittest.skipUnless(has_socket_support, msg)
 
+@functools.cache
+def has_remote_subprocess_debugging():
+    """Check if we have permissions to debug subprocesses remotely.
+
+    Returns True if we have permissions, False if we don't.
+    Checks for:
+    - Platform support (Linux, macOS, Windows only)
+    - On Linux: process_vm_readv support
+    - _remote_debugging module availability
+    - Actual subprocess debugging permissions (e.g., macOS entitlements)
+    Result is cached.
+    """
+    # Check platform support
+    if sys.platform not in ("linux", "darwin", "win32"):
+        return False
+
+    try:
+        import _remote_debugging
+    except ImportError:
+        return False
+
+    # On Linux, check for process_vm_readv support
+    if sys.platform == "linux":
+        if not getattr(_remote_debugging, "PROCESS_VM_READV_SUPPORTED", False):
+            return False
+
+    # First check if we can read our own process
+    if not _remote_debugging.is_python_process(os.getpid()):
+        return False
+
+    # Check subprocess access - debugging child processes may require
+    # additional permissions depending on platform security settings
+    import socket
+    import subprocess
+
+    # Create a socket for child to signal readiness
+    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    server.bind(("127.0.0.1", 0))
+    server.listen(1)
+    port = server.getsockname()[1]
+
+    # Child connects to signal it's ready, then waits for parent to close
+    child_code = f"""
+import socket
+s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+s.connect(("127.0.0.1", {port}))
+s.recv(1)  # Wait for parent to signal done
+"""
+    proc = subprocess.Popen(
+        [sys.executable, "-c", child_code],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    try:
+        server.settimeout(5.0)
+        conn, _ = server.accept()
+        # Child is ready, test if we can probe it
+        result = _remote_debugging.is_python_process(proc.pid)
+        # Check if subprocess is still alive after probing
+        if proc.poll() is not None:
+            return False
+        conn.close()  # Signal child to exit
+        return result
+    except (socket.timeout, OSError):
+        return False
+    finally:
+        server.close()
+        proc.kill()
+        proc.wait()
+
+
+def requires_remote_subprocess_debugging():
+    """Skip tests that require remote subprocess debugging permissions.
+
+    This also implies subprocess support, so no need to use both
+    @requires_subprocess() and @requires_remote_subprocess_debugging().
+    """
+    if not has_subprocess_support:
+        return unittest.skip("requires subprocess support")
+    return unittest.skipUnless(
+        has_remote_subprocess_debugging(),
+        "requires remote subprocess debugging permissions"
+    )
+
+
 # Does strftime() support glibc extension like '%4Y'?
 has_strftime_extensions = False
 if sys.platform != "win32":

crates/host_env/Cargo.toml

@@ -77,8 +77,12 @@ libloading = { workspace = true, optional = true }
 libffi = { workspace = true, features = ["system"], optional = true }
 
 [target.'cfg(target_os = "macos")'.dependencies]
+mach2 = { workspace = true }
 system-configuration = { workspace = true }
 
+[target.'cfg(any(all(target_os = "linux", not(target_env = "uclibc")), target_os = "macos", windows))'.dependencies]
+object = { workspace = true }
+
 [target.'cfg(windows)'.dependencies]
 memchr.workspace = true
 junction = { workspace = true }
@@ -97,6 +101,7 @@ windows-sys = { workspace = true, features = [
     "Win32_System_Com",
     "Win32_System_Console",
     "Win32_System_Diagnostics_Debug",
+    "Win32_System_Diagnostics_ToolHelp",
     "Win32_System_Environment",
     "Win32_System_EventLog",
     "Win32_System_Ole",

crates/host_env/src/lib.rs

@@ -36,6 +36,7 @@ pub mod locale;
 #[cfg(feature = "native-certs")]
 pub mod native_certs;
 pub mod readline;
+pub mod remote_debugging;
 #[cfg(feature = "ssl")]
 pub mod ssl;
 

crates/host_env/src/remote_debugging/common.rs

@@ -0,0 +1,62 @@
+use std::io;
+
+pub(super) const COOKIE: &[u8; 8] = b"xdebugpy";
+
+pub(super) trait RemoteRead {
+    fn read_exact(&mut self, address: u64, destination: &mut [u8]) -> io::Result<()>;
+}
+
+pub(super) fn has_cookie(reader: &mut impl RemoteRead, address: u64) -> io::Result<bool> {
+    if address == 0 {
+        return Ok(false);
+    }
+    let mut bytes = [0; COOKIE.len()];
+    match reader.read_exact(address, &mut bytes) {
+        Ok(()) => Ok(&bytes == COOKIE),
+        Err(error) if stops_probe(&error) => Err(error),
+        Err(_) => Ok(false),
+    }
+}
+
+pub(super) fn stops_probe(error: &io::Error) -> bool {
+    matches!(
+        error.kind(),
+        io::ErrorKind::PermissionDenied | io::ErrorKind::OutOfMemory
+    )
+}
+
+pub(super) fn invalid_data(message: &'static str) -> io::Error {
+    io::Error::new(io::ErrorKind::InvalidData, message)
+}
+
+// These are resource limits, not claims about an unwinder's capabilities.
+// Exhaustion is a failed probe, just like inaccessible or missing metadata.
+pub(super) struct ReadBudget {
+    bytes: usize,
+    calls: usize,
+}
+
+impl ReadBudget {
+    pub(super) fn new() -> Self {
+        Self {
+            bytes: 256 * 1024 * 1024,
+            calls: 65_536,
+        }
+    }
+
+    pub(super) fn consume(&mut self, bytes: usize) -> io::Result<()> {
+        self.bytes = self.bytes.checked_sub(bytes).ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::OutOfMemory,
+                "runtime probe byte budget exhausted",
+            )
+        })?;
+        self.calls = self.calls.checked_sub(1).ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::OutOfMemory,
+                "runtime probe read budget exhausted",
+            )
+        })?;
+        Ok(())
+    }
+}

crates/host_env/src/remote_debugging/linux.rs

@@ -0,0 +1,190 @@
+use super::{COOKIE, ReadBudget, RemoteRead, has_cookie, invalid_data, object_file, stops_probe};
+use std::{
+    ffi::OsStr,
+    fs::{File, OpenOptions},
+    io::{self, Read},
+    os::unix::{ffi::OsStrExt, fs::FileExt},
+    path::Path,
+};
+
+const MAX_MAPS_BYTES: u64 = 8 * 1024 * 1024;
+const SCAN_CHUNK: usize = 1024 * 1024;
+
+struct Process {
+    pid: i32,
+    memory: Option<File>,
+    budget: ReadBudget,
+}
+
+impl RemoteRead for Process {
+    fn read_exact(&mut self, address: u64, destination: &mut [u8]) -> io::Result<()> {
+        let address = usize::try_from(address)
+            .map_err(|_| invalid_data("remote address exceeds native pointer width"))?;
+        address
+            .checked_add(destination.len())
+            .ok_or_else(|| invalid_data("remote read address overflow"))?;
+        let mut completed = 0;
+        while completed < destination.len() {
+            let remaining = &mut destination[completed..];
+            self.budget.consume(remaining.len())?;
+            let current = address + completed;
+            let count = if let Some(memory) = &self.memory {
+                memory.read_at(remaining, current as u64)?
+            } else {
+                let local = libc::iovec {
+                    iov_base: remaining.as_mut_ptr().cast(),
+                    iov_len: remaining.len(),
+                };
+                let remote = libc::iovec {
+                    iov_base: current as *mut libc::c_void,
+                    iov_len: remaining.len(),
+                };
+                // SAFETY: local names the writable destination slice. The
+                // remote pointer is only an address for the kernel, never a
+                // pointer dereferenced by this process.
+                let result = unsafe { libc::process_vm_readv(self.pid, &local, 1, &remote, 1, 0) };
+                if result < 0 {
+                    let error = io::Error::last_os_error();
+                    if error.raw_os_error() == Some(libc::ENOSYS) {
+                        // Match CPython's fallback only for a missing syscall.
+                        // Permission denials must not select another route.
+                        self.memory = Some(
+                            OpenOptions::new()
+                                .read(true)
+                                .write(true)
+                                .open(format!("/proc/{}/mem", self.pid))?,
+                        );
+                        continue;
+                    }
+                    return Err(error);
+                }
+                result as usize
+            };
+            if count == 0 || count > remaining.len() {
+                return Err(io::ErrorKind::UnexpectedEof.into());
+            }
+            completed += count;
+        }
+        Ok(())
+    }
+}
+
+struct Mapping<'a> {
+    start: u64,
+    end: u64,
+    writable: bool,
+    path: &'a [u8],
+}
+
+fn token<'a>(remaining: &mut &'a [u8]) -> Option<&'a [u8]> {
+    let source = (*remaining).trim_ascii_start();
+    if source.is_empty() {
+        return None;
+    }
+    let end = source
+        .iter()
+        .position(u8::is_ascii_whitespace)
+        .unwrap_or(source.len());
+    let (value, rest) = source.split_at(end);
+    *remaining = rest;
+    Some(value)
+}
+
+fn mapping(line: &[u8]) -> Option<Mapping<'_>> {
+    let mut fields = line;
+    let range = token(&mut fields)?;
+    let permissions = token(&mut fields)?;
+    for _ in 0..3 {
+        token(&mut fields)?;
+    }
+    let path = fields.trim_ascii_start();
+    if path.is_empty() || (path.starts_with(b"[") && path.ends_with(b"]")) {
+        return None;
+    }
+    let dash = range.iter().position(|byte| *byte == b'-')?;
+    let parse = |bytes| u64::from_str_radix(core::str::from_utf8(bytes).ok()?, 16).ok();
+    let start = parse(&range[..dash])?;
+    let end = parse(&range[dash + 1..])?;
+    if start == 0 || end <= start {
+        return None;
+    }
+    Some(Mapping {
+        start,
+        end,
+        writable: permissions.starts_with(b"rw"),
+        path,
+    })
+}
+
+fn scan_deleted(process: &mut Process, mapping: &Mapping<'_>) -> io::Result<Option<u64>> {
+    let mut buffer = vec![0; SCAN_CHUNK];
+    let mut address = mapping.start;
+    while address < mapping.end {
+        let wanted = usize::try_from((mapping.end - address).min(SCAN_CHUNK as u64))
+            .map_err(|_| invalid_data("mapping scan size overflow"))?;
+        match process.read_exact(address, &mut buffer[..wanted]) {
+            Ok(()) => {
+                if let Some(offset) = buffer[..wanted]
+                    .windows(COOKIE.len())
+                    .position(|bytes| bytes == COOKIE)
+                {
+                    return Ok(Some(address + offset as u64));
+                }
+            }
+            Err(error) if stops_probe(&error) => return Err(error),
+            Err(_) => {}
+        }
+        let overlap = COOKIE.len() - 1;
+        if wanted <= overlap {
+            break;
+        }
+        address = address
+            .checked_add((wanted - overlap) as u64)
+            .ok_or_else(|| invalid_data("mapping scan address overflow"))?;
+    }
+    Ok(None)
+}
+
+pub(super) fn probe(pid: i32) -> io::Result<bool> {
+    let mut maps = Vec::new();
+    File::open(format!("/proc/{pid}/maps"))?
+        .take(MAX_MAPS_BYTES + 1)
+        .read_to_end(&mut maps)?;
+    if maps.len() as u64 > MAX_MAPS_BYTES {
+        return Err(invalid_data("process map exceeds its size bound"));
+    }
+    let mut process = Process {
+        pid,
+        memory: None,
+        budget: ReadBudget::new(),
+    };
+    for line in maps.split(|byte| *byte == b'\n') {
+        process.budget.consume(0)?;
+        let Some(mapping) = mapping(line) else {
+            continue;
+        };
+        let path = Path::new(OsStr::from_bytes(mapping.path));
+        let Some(name) = path.file_name() else {
+            continue;
+        };
+        if !name.as_bytes().windows(6).any(|bytes| bytes == b"python") {
+            continue;
+        }
+        let candidate = if mapping.path.ends_with(b" (deleted)") {
+            if !mapping.writable {
+                continue;
+            }
+            scan_deleted(&mut process, &mapping)
+        } else {
+            object_file::read(path, &mut process.budget).and_then(|bytes| {
+                object_file::section_address(&bytes, mapping.start, object::BinaryFormat::Elf)
+            })
+        };
+        match candidate {
+            Ok(Some(address)) if has_cookie(&mut process, address)? => return Ok(true),
+            Err(error) if stops_probe(&error) => return Err(error),
+            _ => {}
+        }
+    }
+    Ok(false)
+}

crates/host_env/src/remote_debugging/macos.rs

@@ -0,0 +1,281 @@
+use super::{ReadBudget, RemoteRead, has_cookie, invalid_data, object_file, stops_probe};
+use core::mem::size_of;
+use mach2::{
+    kern_return::{
+        KERN_INVALID_ADDRESS, KERN_NO_ACCESS, KERN_PROTECTION_FAILURE, KERN_RESOURCE_SHORTAGE,
+        KERN_SUCCESS, kern_return_t,
+    },
+    mach_port::mach_port_deallocate,
+    port::{MACH_PORT_DEAD, MACH_PORT_NULL, mach_port_t},
+    traps::{mach_task_self, task_for_pid},
+    vm::{mach_vm_read_overwrite, mach_vm_region},
+    vm_prot::{VM_PROT_EXECUTE, VM_PROT_READ},
+    vm_region::{VM_REGION_BASIC_INFO_64, vm_region_basic_info_64},
+};
+use object::BinaryFormat;
+use std::{ffi::OsStr, io, os::unix::ffi::OsStrExt, path::Path};
+
+const MAX_FAT_ARCHES: usize = 64;
+
+struct OwnedPort(mach_port_t);
+
+impl Drop for OwnedPort {
+    fn drop(&mut self) {
+        if self.0 != MACH_PORT_NULL && self.0 != MACH_PORT_DEAD {
+            // task_for_pid and mach_vm_region each return an owned send right.
+            // mach_task_self is borrowed and must not itself be deallocated.
+            unsafe { mach_port_deallocate(mach_task_self(), self.0) };
+        }
+    }
+}
+
+struct Process {
+    task: OwnedPort,
+    budget: ReadBudget,
+}
+
+fn mach_error(code: kern_return_t) -> io::Error {
+    // Mach error codes carry the detailed code in their low fourteen bits.
+    let kind = match code & 0x3fff {
+        KERN_PROTECTION_FAILURE | KERN_NO_ACCESS => io::ErrorKind::PermissionDenied,
+        KERN_RESOURCE_SHORTAGE => io::ErrorKind::OutOfMemory,
+        KERN_INVALID_ADDRESS => io::ErrorKind::UnexpectedEof,
+        _ => io::ErrorKind::Other,
+    };
+    io::Error::new(kind, format!("Mach runtime probe failed: {code}"))
+}
+
+impl RemoteRead for Process {
+    fn read_exact(&mut self, address: u64, destination: &mut [u8]) -> io::Result<()> {
+        if destination.is_empty() {
+            return Ok(());
+        }
+        address
+            .checked_add(destination.len() as u64 - 1)
+            .ok_or_else(|| invalid_data("remote read address overflow"))?;
+        self.budget.consume(destination.len())?;
+        let mut count = 0;
+        // Only the local destination is a dereferenceable pointer. The remote
+        // address remains an integer interpreted by the kernel.
+        let result = unsafe {
+            mach_vm_read_overwrite(
+                self.task.0,
+                address,
+                destination.len() as u64,
+                destination.as_mut_ptr() as usize as u64,
+                &mut count,
+            )
+        };
+        if result != KERN_SUCCESS {
+            return Err(mach_error(result));
+        }
+        if count != destination.len() as u64 {
+            return Err(io::ErrorKind::UnexpectedEof.into());
+        }
+        Ok(())
+    }
+}
+
+fn word<const N: usize>(bytes: &[u8], offset: usize) -> io::Result<[u8; N]> {
+    let end = offset
+        .checked_add(N)
+        .ok_or_else(|| invalid_data("Mach-O field offset overflow"))?;
+    bytes
+        .get(offset..end)
+        .and_then(|value| value.try_into().ok())
+        .ok_or_else(|| invalid_data("truncated Mach-O field"))
+}
+
+fn read_u32(bytes: &[u8], offset: usize, little: bool) -> io::Result<u32> {
+    let value = word(bytes, offset)?;
+    Ok(if little {
+        u32::from_le_bytes(value)
+    } else {
+        u32::from_be_bytes(value)
+    })
+}
+
+fn sysctl_int(name: &core::ffi::CStr) -> io::Result<i32> {
+    let mut value = 0i32;
+    let mut length = size_of::<i32>();
+    let result = unsafe {
+        libc::sysctlbyname(
+            name.as_ptr(),
+            (&mut value as *mut i32).cast(),
+            &mut length,
+            core::ptr::null_mut(),
+            0,
+        )
+    };
+    if result != 0 {
+        return Err(io::Error::last_os_error());
+    }
+    if length != size_of::<i32>() {
+        return Err(invalid_data("invalid CPU sysctl size"));
+    }
+    Ok(value)
+}
+
+fn mach_slice(bytes: &[u8]) -> io::Result<Option<&[u8]>> {
+    let magic = u32::from_be_bytes(word(bytes, 0)?);
+    let little = match magic {
+        object::macho::FAT_MAGIC => false,
+        object::macho::FAT_CIGAM => true,
+        _ => return Ok(Some(bytes)),
+    };
+    let count = usize::try_from(read_u32(bytes, 4, little)?)
+        .map_err(|_| invalid_data("Mach-O architecture count overflow"))?;
+    if count > MAX_FAT_ARCHES {
+        return Err(invalid_data("Mach-O architecture count exceeds its bound"));
+    }
+    let entry_size = 20;
+    let table_end = count
+        .checked_mul(entry_size)
+        .and_then(|size| size.checked_add(8))
+        .ok_or_else(|| invalid_data("Mach-O architecture table overflow"))?;
+    if table_end > bytes.len() {
+        return Err(invalid_data("truncated Mach-O architecture table"));
+    }
+    // Match the pinned CPython locator's host-CPU selection, rather than guessing
+    // from the inspector's build target (which may itself run under translation).
+    let cpu = sysctl_int(c"hw.cputype")? as u32;
+    let capable = sysctl_int(c"hw.cpu64bit_capable")?;
+    let cpu = cpu
+        | if capable != 0 {
+            object::macho::CPU_ARCH_ABI64
+        } else {
+            0
+        };
+    for index in 0..count {
+        // The bounded, validated architecture table makes these offsets safe.
+        let entry = 8 + index * entry_size;
+        if read_u32(bytes, entry, little)? != cpu {
+            continue;
+        }
+        let offset = u64::from(read_u32(bytes, entry + 8, little)?);
+        let length = u64::from(read_u32(bytes, entry + 12, little)?);
+        let end = offset
+            .checked_add(length)
+            .ok_or_else(|| invalid_data("Mach-O architecture range overflow"))?;
+        let offset = usize::try_from(offset)
+            .map_err(|_| invalid_data("Mach-O architecture offset overflow"))?;
+        let end =
+            usize::try_from(end).map_err(|_| invalid_data("Mach-O architecture end overflow"))?;
+        if offset < table_end || length == 0 {
+            return Err(invalid_data("invalid Mach-O architecture range"));
+        }
+        return bytes
+            .get(offset..end)
+            .map(Some)
+            .ok_or_else(|| invalid_data("truncated Mach-O architecture"));
+    }
+    Ok(None)
+}
+
+fn candidate(process: &mut Process, path: &Path, base: u64) -> io::Result<bool> {
+    let bytes = object_file::read(path, &mut process.budget)?;
+    let Some(bytes) = mach_slice(&bytes)? else {
+        return Ok(false);
+    };
+    let Some(address) = object_file::section_address(bytes, base, BinaryFormat::MachO)? else {
+        return Ok(false);
+    };
+    has_cookie(process, address)
+}
+
+fn search(process: &mut Process, pid: i32, prefix: &[u8]) -> io::Result<bool> {
+    let mut address = 0;
+    loop {
+        process.budget.consume(0)?;
+        let requested = address;
+        let mut size = 0;
+        let mut info = vm_region_basic_info_64::default();
+        let mut count = vm_region_basic_info_64::count();
+        let mut object = MACH_PORT_NULL;
+        let result = unsafe {
+            mach_vm_region(
+                process.task.0,
+                &mut address,
+                &mut size,
+                VM_REGION_BASIC_INFO_64,
+                (&mut info as *mut vm_region_basic_info_64).cast(),
+                &mut count,
+                &mut object,
+            )
+        };
+        // Even irrelevant regions may return an owned memory-object port.
+        let _object = OwnedPort(object);
+        if result == KERN_INVALID_ADDRESS {
+            return Ok(false);
+        }
+        if result != KERN_SUCCESS {
+            return Err(mach_error(result));
+        }
+        if count < vm_region_basic_info_64::count() || address < requested || size == 0 {
+            return Err(invalid_data("invalid Mach memory region"));
+        }
+        let next = address
+            .checked_add(size)
+            .ok_or_else(|| invalid_data("Mach memory region address overflow"))?;
+        if info.protection & (VM_PROT_READ | VM_PROT_EXECUTE) == (VM_PROT_READ | VM_PROT_EXECUTE) {
+            let mut path = [0u8; libc::MAXPATHLEN as usize + 1];
+            process.budget.consume(0)?;
+            let length = unsafe {
+                // A zero return can describe an anonymous mapping; clear errno
+                // first so an older failure cannot be mistaken for a denial.
+                *libc::__error() = 0;
+                libc::proc_regionfilename(
+                    pid,
+                    address,
+                    path.as_mut_ptr().cast(),
+                    libc::MAXPATHLEN as u32,
+                )
+            };
+            if length <= 0 {
+                let error = io::Error::last_os_error();
+                if stops_probe(&error) {
+                    return Err(error);
+                }
+            } else if (length as usize) < path.len()
+                && let Some(end) = path.iter().position(|&byte| byte == 0)
+                && end < libc::MAXPATHLEN as usize
+            {
+                let path = &path[..end];
+                let name = path.rsplit(|&byte| byte == b'/').next().unwrap_or(path);
+                if name.starts_with(prefix) {
+                    match candidate(process, Path::new(OsStr::from_bytes(path)), address) {
+                        Ok(true) => return Ok(true),
+                        Err(error) if stops_probe(&error) => return Err(error),
+                        Ok(false) | Err(_) => {}
+                    }
+                }
+            }
+        }
+        address = next;
+    }
+}
+
+pub(super) fn probe(pid: i32) -> io::Result<bool> {
+    let mut task = MACH_PORT_NULL;
+    let result = unsafe { task_for_pid(mach_task_self(), pid, &mut task) };
+    let task = OwnedPort(task);
+    if result != KERN_SUCCESS || task.0 == MACH_PORT_NULL || task.0 == MACH_PORT_DEAD {
+        // CPython classifies every task_for_pid failure as a permission failure.
+        return Err(io::Error::new(
+            io::ErrorKind::PermissionDenied,
+            format!("cannot acquire task port for PID {pid}: {result}"),
+        ));
+    }
+    let mut process = Process {
+        task,
+        budget: ReadBudget::new(),
+    };
+    for prefix in [b"libpython".as_slice(), b"python", b"Python"] {
+        match search(&mut process, pid, prefix) {
+            Ok(true) => return Ok(true),
+            Err(error) if stops_probe(&error) => return Err(error),
+            Ok(false) | Err(_) => {}
+        }
+    }
+    Ok(false)
+}

crates/host_env/src/remote_debugging/mod.rs

@@ -0,0 +1,56 @@
+//! Read-only discovery of CPython's exported runtime metadata.
+//!
+//! A successful cookie probe identifies a runtime; it does not validate its
+//! debug offsets or promise that stack unwinding is supported.
+
+#[cfg(all(target_os = "linux", not(target_env = "uclibc")))]
+mod linux;
+#[cfg(target_os = "macos")]
+mod macos;
+#[cfg(windows)]
+mod windows;
+
+#[cfg(any(
+    all(target_os = "linux", not(target_env = "uclibc")),
+    target_os = "macos",
+    windows
+))]
+mod object_file;
+
+#[cfg(any(
+    all(target_os = "linux", not(target_env = "uclibc")),
+    target_os = "macos",
+    windows
+))]
+mod common;
+
+#[cfg(all(target_os = "linux", not(target_env = "uclibc")))]
+use common::COOKIE;
+#[cfg(any(
+    all(target_os = "linux", not(target_env = "uclibc")),
+    target_os = "macos",
+    windows
+))]
+use common::{ReadBudget, RemoteRead, has_cookie, invalid_data, stops_probe};
+
+pub const PROCESS_VM_READV_SUPPORTED: u8 =
+    cfg!(all(target_os = "linux", not(target_env = "uclibc"))) as u8;
+
+pub fn is_python_process(pid: i32) -> bool {
+    #[cfg(all(target_os = "linux", not(target_env = "uclibc")))]
+    let result = linux::probe(pid);
+    #[cfg(target_os = "macos")]
+    let result = macos::probe(pid);
+    #[cfg(windows)]
+    let result = windows::probe(pid);
+    #[cfg(not(any(
+        all(target_os = "linux", not(target_env = "uclibc")),
+        target_os = "macos",
+        windows
+    )))]
+    let result: Result<bool, std::io::Error> = {
+        let _ = pid;
+        Err(std::io::ErrorKind::Unsupported.into())
+    };
+    result.unwrap_or(false)
+}

crates/host_env/src/remote_debugging/object_file.rs

@@ -0,0 +1,94 @@
+use super::{ReadBudget, invalid_data};
+use object::{BinaryFormat, Object, ObjectSection, ObjectSegment};
+use std::{fs::OpenOptions, io, io::Read, path::Path};
+
+const MAX_OBJECT_BYTES: u64 = 256 * 1024 * 1024;
+
+pub(super) fn read(path: &Path, budget: &mut ReadBudget) -> io::Result<Vec<u8>> {
+    budget.consume(0)?;
+    let mut options = OpenOptions::new();
+    options.read(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt;
+        // A mapped pathname can be replaced with a FIFO while being inspected.
+        options.custom_flags(libc::O_NONBLOCK);
+    }
+    let file = options.open(path)?;
+    let metadata = file.metadata()?;
+    if !metadata.is_file() || metadata.len() > MAX_OBJECT_BYTES {
+        return Err(invalid_data(
+            "runtime candidate is not a bounded regular file",
+        ));
+    }
+    let mut bytes = Vec::new();
+    bytes
+        .try_reserve_exact(metadata.len() as usize)
+        .map_err(|_| io::Error::from(io::ErrorKind::OutOfMemory))?;
+    let result = file.take(MAX_OBJECT_BYTES + 1).read_to_end(&mut bytes);
+    // A failed read may already have appended bytes. Account for them before
+    // propagating the error, as well as before rejecting a grown file.
+    budget.consume(bytes.len())?;
+    result?;
+    if bytes.len() as u64 > MAX_OBJECT_BYTES {
+        return Err(invalid_data("runtime candidate grew beyond its size bound"));
+    }
+    Ok(bytes)
+}
+
+pub(super) fn section_address(
+    bytes: &[u8],
+    mapping_base: u64,
+    format: BinaryFormat,
+) -> io::Result<Option<u64>> {
+    let file = object::File::parse(bytes).map_err(|_| invalid_data("invalid object file"))?;
+    if file.format() != format {
+        return Ok(None);
+    }
+    let name = match format {
+        BinaryFormat::Elf => b".PyRuntime".as_slice(),
+        BinaryFormat::MachO => b"PyRuntime".as_slice(),
+        BinaryFormat::Pe => b"PyRuntim".as_slice(),
+        _ => return Ok(None),
+    };
+    let section = if format == BinaryFormat::MachO {
+        file.sections().find(|section| {
+            section.name_bytes().ok() == Some(name)
+                && section.segment_name_bytes().ok() == Some(Some(b"__DATA".as_slice()))
+        })
+    } else {
+        file.section_by_name_bytes(name)
+    };
+    let Some(section) = section else {
+        return Ok(None);
+    };
+    let image_base = match format {
+        BinaryFormat::Elf => {
+            // The ELF reader's segment iterator contains only PT_LOAD entries.
+            let first = file
+                .segments()
+                .next()
+                .ok_or_else(|| invalid_data("ELF image has no load segment"))?;
+            let alignment = first.align();
+            if alignment == 0 || !alignment.is_power_of_two() {
+                return Err(invalid_data("invalid ELF load alignment"));
+            }
+            first.address() - first.address() % alignment
+        }
+        BinaryFormat::MachO => file
+            .segments()
+            .find(|segment| segment.name_bytes().ok() == Some(Some(b"__TEXT".as_slice())))
+            .ok_or_else(|| invalid_data("Mach-O image has no text segment"))?
+            .address(),
+        BinaryFormat::Pe => file.relative_address_base(),
+        _ => unreachable!(),
+    };
+    let relative = section
+        .address()
+        .checked_sub(image_base)
+        .ok_or_else(|| invalid_data("runtime section precedes its image base"))?;
+    let address = mapping_base
+        .checked_add(relative)
+        .ok_or_else(|| invalid_data("runtime section address overflow"))?;
+    Ok((address != 0).then_some(address))
+}

crates/host_env/src/remote_debugging/windows.rs

@@ -0,0 +1,178 @@
+use super::{ReadBudget, RemoteRead, has_cookie, invalid_data, object_file, stops_probe};
+use core::mem::size_of;
+use object::BinaryFormat;
+use std::{ffi::OsString, io, os::windows::ffi::OsStringExt, path::PathBuf};
+use windows_sys::Win32::{
+    Foundation::{
+        CloseHandle, ERROR_BAD_LENGTH, ERROR_NO_MORE_FILES, GetLastError, HANDLE,
+        INVALID_HANDLE_VALUE,
+    },
+    System::{
+        Diagnostics::{
+            Debug::ReadProcessMemory,
+            ToolHelp::{
+                CreateToolhelp32Snapshot, MODULEENTRY32W, Module32FirstW, Module32NextW,
+                TH32CS_SNAPMODULE,
+            },
+        },
+        Threading::{
+            OpenProcess, PROCESS_QUERY_INFORMATION, PROCESS_SUSPEND_RESUME, PROCESS_VM_OPERATION,
+            PROCESS_VM_READ, PROCESS_VM_WRITE,
+        },
+    },
+};
+
+const MAX_SNAPSHOT_ATTEMPTS: usize = 16;
+const MAX_MODULES: usize = 16_384;
+
+struct OwnedHandle(HANDLE);
+
+impl Drop for OwnedHandle {
+    fn drop(&mut self) {
+        // Only successfully acquired process and snapshot handles are owned here.
+        unsafe { CloseHandle(self.0) };
+    }
+}
+
+struct Process {
+    handle: OwnedHandle,
+    budget: ReadBudget,
+}
+
+impl RemoteRead for Process {
+    fn read_exact(&mut self, address: u64, mut destination: &mut [u8]) -> io::Result<()> {
+        if destination.is_empty() {
+            return Ok(());
+        }
+        let end = address
+            .checked_add(destination.len() as u64 - 1)
+            .ok_or_else(|| invalid_data("remote read address overflow"))?;
+        usize::try_from(end).map_err(|_| invalid_data("remote address exceeds pointer width"))?;
+        let mut address = usize::try_from(address)
+            .map_err(|_| invalid_data("remote address exceeds pointer width"))?;
+        while !destination.is_empty() {
+            self.budget.consume(destination.len())?;
+            let mut count = 0;
+            // The address is passed to the OS, never dereferenced in this process.
+            let success = unsafe {
+                ReadProcessMemory(
+                    self.handle.0,
+                    address as *const _,
+                    destination.as_mut_ptr().cast(),
+                    destination.len(),
+                    &mut count,
+                )
+            };
+            if success == 0 {
+                // In particular, an ERROR_PARTIAL_COPY failure is not a successful
+                // short read. CPython only continues after successful reads.
+                return Err(io::Error::last_os_error());
+            }
+            if count == 0 {
+                return Err(io::ErrorKind::UnexpectedEof.into());
+            }
+            if count > destination.len() {
+                return Err(invalid_data("remote read exceeded its destination"));
+            }
+            destination = &mut destination[count..];
+            if !destination.is_empty() {
+                address = address
+                    .checked_add(count)
+                    .ok_or_else(|| invalid_data("remote read address overflow"))?;
+            }
+        }
+        Ok(())
+    }
+}
+
+fn snapshot(pid: u32, budget: &mut ReadBudget) -> io::Result<OwnedHandle> {
+    for _ in 0..MAX_SNAPSHOT_ATTEMPTS {
+        budget.consume(0)?;
+        // Keep CPython's module selection flags, including its cross-bitness limits.
+        let handle = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPMODULE, pid) };
+        if handle != INVALID_HANDLE_VALUE {
+            return Ok(OwnedHandle(handle));
+        }
+        let error = unsafe { GetLastError() };
+        if error != ERROR_BAD_LENGTH {
+            return Err(io::Error::from_raw_os_error(error as i32));
+        }
+    }
+    Err(io::Error::from_raw_os_error(ERROR_BAD_LENGTH as i32))
+}
+
+fn terminated(value: &[u16]) -> io::Result<&[u16]> {
+    let end = value
+        .iter()
+        .position(|&unit| unit == 0)
+        .ok_or_else(|| invalid_data("unterminated module name or path"))?;
+    Ok(&value[..end])
+}
+
+fn candidate(process: &mut Process, module: &MODULEENTRY32W) -> io::Result<bool> {
+    const PYTHON: &[u16] = &[112, 121, 116, 104, 111, 110];
+    let name = terminated(&module.szModule)?;
+    if !name.windows(PYTHON.len()).any(|part| part == PYTHON) {
+        return Ok(false);
+    }
+    let path = PathBuf::from(OsString::from_wide(terminated(&module.szExePath)?));
+    let bytes = object_file::read(&path, &mut process.budget)?;
+    let base = module.modBaseAddr as usize as u64;
+    let Some(address) = object_file::section_address(&bytes, base, BinaryFormat::Pe)? else {
+        return Ok(false);
+    };
+    has_cookie(process, address)
+}
+
+pub(super) fn probe(pid: i32) -> io::Result<bool> {
+    // CPython requests these rights even for this read-only operation. Reducing
+    // them could make detection succeed where CPython's handle initialization fails.
+    // The DWORD conversion also preserves the Windows API's signed-PID behavior.
+    let handle = unsafe {
+        OpenProcess(
+            PROCESS_VM_READ
+                | PROCESS_VM_WRITE
+                | PROCESS_VM_OPERATION
+                | PROCESS_QUERY_INFORMATION
+                | PROCESS_SUSPEND_RESUME,
+            0,
+            pid as u32,
+        )
+    };
+    if handle.is_null() {
+        return Err(io::Error::last_os_error());
+    }
+    let mut process = Process {
+        handle: OwnedHandle(handle),
+        budget: ReadBudget::new(),
+    };
+    let snapshot = snapshot(pid as u32, &mut process.budget)?;
+    let mut module = MODULEENTRY32W {
+        dwSize: size_of::<MODULEENTRY32W>() as u32,
+        ..Default::default()
+    };
+    process.budget.consume(0)?;
+    let mut success = unsafe { Module32FirstW(snapshot.0, &mut module) };
+    for _ in 0..MAX_MODULES {
+        if success == 0 {
+            let error = unsafe { GetLastError() };
+            return if error == ERROR_NO_MORE_FILES {
+                Ok(false)
+            } else {
+                Err(io::Error::from_raw_os_error(error as i32))
+            };
+        }
+        match candidate(&mut process, &module) {
+            Ok(true) => return Ok(true),
+            Err(error) if stops_probe(&error) => return Err(error),
+            Ok(false) | Err(_) => {}
+        }
+        process.budget.consume(0)?;
+        module.dwSize = size_of::<MODULEENTRY32W>() as u32;
+        success = unsafe { Module32NextW(snapshot.0, &mut module) };
+    }
+    Err(io::Error::new(
+        io::ErrorKind::OutOfMemory,
+        "runtime probe module budget exhausted",
+    ))
+}

crates/stdlib/src/_remote_debugging.rs

@@ -7,7 +7,7 @@ mod _remote_debugging {
     use super::binary;
     use crate::vm::{
         AsObject, Py, PyObject, PyObjectRef, PyRef, PyResult, VirtualMachine,
-        builtins::{PyBaseExceptionRef, PyDictRef, PyList, PyTuple, PyType},
+        builtins::{PyBaseExceptionRef, PyDictRef, PyFloat, PyList, PyTuple, PyType},
         common::{lock::PyMutex, rc::PyRc},
         convert::{IntoPyException, ToPyObject},
         function::{FsPath, FuncArgs},
@@ -117,6 +117,10 @@ mod _remote_debugging {
     #[pyclass(with(PyStructSequence))]
     impl InterpreterInfo {}
 
+    #[pyattr]
+    const PROCESS_VM_READV_SUPPORTED: u8 =
+        rustpython_host_env::remote_debugging::PROCESS_VM_READV_SUPPORTED;
+
     #[pyattr]
     const THREAD_STATUS_HAS_GIL: u8 = 1;
     #[pyattr]
@@ -130,6 +134,22 @@ mod _remote_debugging {
     #[pyattr]
     const THREAD_STATUS_MAIN_THREAD: u8 = 32;
 
+    #[derive(FromArgs)]
+    struct IsPythonProcessArgs {
+        #[pyarg(any)]
+        pid: PyObjectRef,
+    }
+
+    /// Check if a process is a Python process.
+    #[pyfunction]
+    fn is_python_process(args: IsPythonProcessArgs, vm: &VirtualMachine) -> PyResult<bool> {
+        // Finish argument binding before invoking a user-defined __index__.
+        let pid: i32 = args.pid.try_into_value(vm)?;
+        Ok(rustpython_host_env::remote_debugging::is_python_process(
+            pid,
+        ))
+    }
+
     #[pyfunction]
     fn zstd_available() -> bool {
         false
@@ -293,20 +313,36 @@ mod _remote_debugging {
 
     #[derive(FromArgs)]
     struct BinaryExitArgs {
-        #[pyarg(positional, optional)]
+        #[pyarg(any, optional)]
         exc_type: Option<PyObjectRef>,
-        #[pyarg(positional, optional)]
+        #[pyarg(any, optional, name = "exc_val")]
         _exc_value: Option<PyObjectRef>,
-        #[pyarg(positional, optional)]
+        #[pyarg(any, optional, name = "exc_tb")]
         _exc_tb: Option<PyObjectRef>,
     }
 
+    #[derive(FromArgs)]
+    struct SampleArgs {
+        #[pyarg(any)]
+        stack_frames: PyObjectRef,
+        #[pyarg(any)]
+        timestamp_us: PyObjectRef,
+    }
+
+    fn profile_float(obj: &PyObject, vm: &VirtualMachine) -> PyResult<f64> {
+        // PyFloat_AsDouble reads float subclasses without calling __float__.
+        Ok(match obj.downcast_ref::<PyFloat>() {
+            Some(value) => value.to_f64(),
+            None => obj.try_float(vm)?.to_f64(),
+        })
+    }
+
     #[derive(FromArgs)]
     struct SetStatsArgs {
         #[pyarg(any)]
-        duration_sec: f64,
+        duration_sec: PyObjectRef,
         #[pyarg(any)]
-        sample_rate: f64,
+        sample_rate: PyObjectRef,
         #[pyarg(any, optional)]
         error_rate: Option<PyObjectRef>,
         #[pyarg(any, optional)]
@@ -316,19 +352,19 @@ mod _remote_debugging {
     fn optional_float(value: Option<PyObjectRef>, vm: &VirtualMachine) -> PyResult<Option<f64>> {
         value
             .filter(|value| !vm.is_none(value))
-            .map(|value| value.try_float(vm).map(|value| value.to_f64()))
+            .map(|value| profile_float(&value, vm))
             .transpose()
     }
 
     #[pyclass(with(Constructor, Initializer))]
     impl BinaryWriter {
         #[pymethod]
-        fn write_sample(
-            zelf: &Py<Self>,
-            stack_frames: PyObjectRef,
-            timestamp_us: u64,
-            vm: &VirtualMachine,
-        ) -> PyResult<()> {
+        fn write_sample(zelf: &Py<Self>, args: SampleArgs, vm: &VirtualMachine) -> PyResult<()> {
+            let SampleArgs {
+                stack_frames,
+                timestamp_us,
+            } = args;
+            let timestamp_us = timestamp_us.try_into_value::<u64>(vm)?;
             if zelf.state.lock().writer.is_none() {
                 return Err(vm.new_value_error("Writer is closed"));
             }
@@ -378,13 +414,15 @@ mod _remote_debugging {
 
         #[pymethod]
         fn set_stats(zelf: &Py<Self>, args: SetStatsArgs, vm: &VirtualMachine) -> PyResult<()> {
+            let duration_sec = profile_float(&args.duration_sec, vm)?;
+            let sample_rate = profile_float(&args.sample_rate, vm)?;
             let is_open = zelf.state.lock().writer.is_some();
             if !is_open {
                 return Err(vm.new_value_error("Writer is closed"));
             }
             let profile = binary::ProfileStats {
-                duration_sec: args.duration_sec,
-                sample_rate: args.sample_rate,
+                duration_sec,
+                sample_rate,
                 error_rate: optional_float(args.error_rate, vm)?,
                 missed_samples: optional_float(args.missed_samples, vm)?,
             };

crates/stdlib/src/_remote_debugging/binary.rs

@@ -792,7 +792,7 @@ impl Reader {
         let mut footer = Cursor::new(&data[footer_start..]);
         let string_count = endian.u32(footer.fixed()?);
         let frame_count = endian.u32(footer.fixed()?);
-        let (profile_stats, tables_end) = parse_profile_stats(&data, endian, footer_start)?;
+        let (profile_stats, _) = parse_profile_stats(&data, endian, footer_start)?;
         for (name, offset) in [("string", string_offset), ("frame", frame_offset)] {
             if offset > data.len() as u64 {
                 return Err(invalid(format!(
@@ -815,15 +815,14 @@ impl Reader {
             .map_err(|_| Error::Overflow("File is too large to read"))?;
         let frame_offset = usize::try_from(frame_offset)
             .map_err(|_| Error::Overflow("File is too large to read"))?;
-        if frame_offset > tables_end {
-            return Err(invalid("Frame table overlaps file trailer"));
-        }
         if compression_type == 1 {
             return Err(Error::Runtime(
                 "File uses zstd compression but zstd support not compiled in",
             ));
         }
-        let mut string_data = Cursor::new(&data[string_offset..frame_offset]);
+        // CPython bounds table entries by the remaining file bytes, including
+        // subsequent sections. Keep the count and cursor checks anchored to EOF.
+        let mut string_data = Cursor::new(&data[string_offset..]);
         validate_count("string", string_count, string_data.remaining(), 1)?;
         let mut strings = Vec::new();
         reserve(&mut strings, string_count as usize)?;
@@ -836,7 +835,7 @@ impl Reader {
                 .map_err(|_| invalid("String table overflow"))?;
             strings.push(decode_utf8_replacing(bytes)?);
         }
-        let mut frame_data = Cursor::new(&data[frame_offset..tables_end]);
+        let mut frame_data = Cursor::new(&data[frame_offset..]);
         validate_count("frame", frame_count, frame_data.remaining(), 7)?;
         let mut frames = Vec::new();
         reserve(&mut frames, frame_count as usize)?;

scripts/update_lib/cmd_migrate.py

@@ -19,6 +19,12 @@ Non-marker adaptations require reviewed blocks on each migration, for example
 Use a host Python that parses the source version (Python 3.15 for 3.15 syntax),
 as required by the existing marker extraction and application steps.
 
+For additive updates, --assignments or --functions selects contiguous canonical
+top-level definitions before an unchanged anchor. Both input SHA256 pins are
+required. --exports copies only complete selected __all__ lines. Function
+dependencies are checked without importing or executing either module; unchanged
+imports and simple assignments are supported, not arbitrary existing helpers.
+
 Canonical removals are a separate, explicit operation. JSON source metadata
 pins path (Lib/test/test_*), commit and tree; removals maps explicit relative
 filenames to sha256 and upstream_commit. The caller must first verify and
@@ -43,6 +49,7 @@ import pathlib
 import re
 import shutil
 import subprocess
+import symtable
 import sys
 
 sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
@@ -57,6 +64,68 @@ def _preserve_block(old, new, canonical, local):
     return new.replace(canonical, local, 1)
 
 
+def _single_assignment(tree, name):
+    matches = [
+        node
+        for node in ast.walk(tree)
+        if isinstance(node, ast.Name)
+        and isinstance(node.ctx, ast.Store)
+        and node.id == name
+    ]
+    nodes = [
+        node
+        for node in tree.body
+        if isinstance(node, ast.Assign)
+        and len(node.targets) == 1
+        and isinstance(node.targets[0], ast.Name)
+        and node.targets[0].id == name
+    ]
+    rebound = any(
+        isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
+        and node.name == name
+        or isinstance(node, ast.alias)
+        and (node.asname or node.name.split(".")[0]) == name
+        for node in ast.walk(tree)
+    )
+    if len(nodes) != 1 or matches != [nodes[0].targets[0]] or rebound:
+        raise ValueError(f"Expected one top-level assignment to {name}")
+    return nodes[0]
+
+
+def _selected_export_addition(src, dst, source_lines, lines, exports):
+    src_all, dst_all = (
+        _single_assignment(src, "__all__"),
+        _single_assignment(dst, "__all__"),
+    )
+    if not isinstance(src_all.value, ast.List) or not isinstance(
+        dst_all.value, ast.List
+    ):
+        raise ValueError("__all__ must be a literal list")
+    if any(
+        not isinstance(node, ast.Constant) or not isinstance(node.value, str)
+        for node in src_all.value.elts + dst_all.value.elts
+    ):
+        raise ValueError("__all__ entries must be literal strings")
+    src_values, dst_values = (
+        ast.literal_eval(src_all.value),
+        ast.literal_eval(dst_all.value),
+    )
+    if any(src_values.count(name) != 1 or name in dst_values for name in exports):
+        raise ValueError(
+            "Exports must occur once in canonical __all__ and be absent locally"
+        )
+    export_nodes = [node for node in src_all.value.elts if node.value in exports]
+    export_lines = sorted({node.lineno - 1 for node in export_nodes})
+    fragment = "".join(source_lines[line] for line in export_lines)
+    if ast.literal_eval("[\n" + fragment + "]") != exports:
+        raise ValueError(
+            "Canonical export lines include unselected entries or differ in order"
+        )
+    if lines[dst_all.lineno - 1].strip() != "__all__ = [":
+        raise ValueError("Expected a multiline destination __all__ list")
+    return dst_all.lineno, [fragment]
+
+
 def copy_selected_assignments(source, destination, names, exports):
     """Copy contiguous canonical assignments and export lines, failing closed.
 
@@ -74,34 +143,7 @@ def copy_selected_assignments(source, destination, names, exports):
         raise ValueError("Exports must be unique selected assignment names")
     src, dst = ast.parse(source), ast.parse(destination)
 
-    def assignment(tree, name):
-        matches = [
-            node
-            for node in ast.walk(tree)
-            if isinstance(node, ast.Name)
-            and isinstance(node.ctx, ast.Store)
-            and node.id == name
-        ]
-        nodes = [
-            node
-            for node in tree.body
-            if isinstance(node, ast.Assign)
-            and len(node.targets) == 1
-            and isinstance(node.targets[0], ast.Name)
-            and node.targets[0].id == name
-        ]
-        rebound = any(
-            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
-            and node.name == name
-            or isinstance(node, ast.alias)
-            and (node.asname or node.name.split(".")[0]) == name
-            for node in ast.walk(tree)
-        )
-        if len(nodes) != 1 or matches != [nodes[0].targets[0]] or rebound:
-            raise ValueError(f"Expected one top-level assignment to {name}")
-        return nodes[0]
-
-    selected = [assignment(src, name) for name in names]
+    selected = [_single_assignment(src, name) for name in names]
     start = src.body.index(selected[0])
     if src.body[start : start + len(names)] != selected or start + len(names) == len(
         src.body
@@ -223,34 +265,9 @@ def copy_selected_assignments(source, destination, names, exports):
         )
     ]
     if exports:
-        src_all, dst_all = assignment(src, "__all__"), assignment(dst, "__all__")
-        if not isinstance(src_all.value, ast.List) or not isinstance(
-            dst_all.value, ast.List
-        ):
-            raise ValueError("__all__ must be a literal list")
-        if any(
-            not isinstance(node, ast.Constant) or not isinstance(node.value, str)
-            for node in src_all.value.elts + dst_all.value.elts
-        ):
-            raise ValueError("__all__ entries must be literal strings")
-        src_values, dst_values = (
-            ast.literal_eval(src_all.value),
-            ast.literal_eval(dst_all.value),
+        additions.append(
+            _selected_export_addition(src, dst, source_lines, lines, exports)
         )
-        if any(src_values.count(name) != 1 or name in dst_values for name in exports):
-            raise ValueError(
-                "Exports must occur once in canonical __all__ and be absent locally"
-            )
-        export_nodes = [node for node in src_all.value.elts if node.value in exports]
-        export_lines = sorted({node.lineno - 1 for node in export_nodes})
-        fragment = "".join(source_lines[line] for line in export_lines)
-        if ast.literal_eval("[\n" + fragment + "]") != exports:
-            raise ValueError(
-                "Canonical export lines include unselected entries or differ in order"
-            )
-        if lines[dst_all.lineno - 1].strip() != "__all__ = [":
-            raise ValueError("Expected a multiline destination __all__ list")
-        additions.append((dst_all.lineno, [fragment]))
     for offset, fragment in sorted(additions, reverse=True):
         lines[offset:offset] = fragment
     result = "".join(lines)
@@ -258,10 +275,291 @@ def copy_selected_assignments(source, destination, names, exports):
     return result
 
 
-def migrate_selected_assignments(path, names, exports, source_hash, destination_hash):
+def _module_bindings(tree):
+    """Collect possible module bindings without mistaking function locals for them."""
+    bindings = collections.defaultdict(list)
+
+    def visit(node):
+        children = ast.iter_child_nodes(node)
+        names = []
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+            names = [node.name]
+            # A declaration in a nested scope can rebind a module dependency.
+            for child in ast.walk(node):
+                if isinstance(child, ast.Global):
+                    for name in child.names:
+                        bindings[name].append(child)
+            children = list(node.decorator_list)
+            if isinstance(node, ast.ClassDef):
+                children += node.bases + node.keywords
+            else:
+                children += node.args.defaults + [
+                    value for value in node.args.kw_defaults if value is not None
+                ]
+        elif isinstance(node, ast.Lambda):
+            children = node.args.defaults + [
+                value for value in node.args.kw_defaults if value is not None
+            ]
+        elif isinstance(node, ast.Name) and isinstance(node.ctx, (ast.Store, ast.Del)):
+            names = [node.id]
+        elif isinstance(node, ast.alias):
+            names = [node.asname or node.name.split(".")[0]]
+        elif isinstance(node, (ast.ExceptHandler, ast.MatchAs, ast.MatchStar)):
+            names = [node.name] if node.name is not None else []
+        elif isinstance(node, ast.MatchMapping) and node.rest is not None:
+            names = [node.rest]
+        for name in names:
+            bindings[name].append(node)
+        for child in children:
+            visit(child)
+
+    visit(tree)
+    return bindings
+
+
+def _function_dependencies(fragment):
+    """Find definition-time and deferred global reads without executing source."""
+    table = symtable.symtable(fragment, "<canonical function>", "exec")
+    if any(
+        symbol.is_assigned() and not symbol.is_namespace()
+        for symbol in table.get_symbols()
+    ):
+        raise ValueError("Function headers cannot bind additional module names")
+    immediate = {
+        symbol.get_name() for symbol in table.get_symbols() if symbol.is_referenced()
+    }
+    deferred = set()
+
+    def visit(scope):
+        for symbol in scope.get_symbols():
+            if symbol.is_global():
+                if symbol.is_assigned() or symbol.is_imported():
+                    raise ValueError("Selected functions cannot rebind module globals")
+                if symbol.is_referenced():
+                    deferred.add(symbol.get_name())
+        for child in scope.get_children():
+            visit(child)
+
+    for child in table.get_children():
+        visit(child)
+    if (immediate | deferred) & {"eval", "exec", "globals", "locals", "vars"}:
+        raise ValueError("Dynamic namespace dependencies require a full migration")
+    return immediate, deferred
+
+
+def copy_selected_functions(source, destination, names, exports):
+    """Add contiguous complete functions, preserving unrelated destination bytes.
+
+    Global dependencies must be builtins, selected functions, or unambiguously
+    bound identical imports/simple assignments (checked transitively). Existing
+    helper functions and dynamic namespace dependencies require a full migration.
+    Scoped expressions in definition-time headers and classes anywhere inside
+    selected functions also require a full migration. Symbol-table analysis
+    distinguishes body locals, closures and comprehensions from module
+    dependencies; it never imports or calls the selected functions.
+    """
+    if (
+        not names
+        or len(set(names)) != len(names)
+        or not all(name.isidentifier() for name in names)
+    ):
+        raise ValueError("Function names must be unique identifiers")
+    if len(set(exports)) != len(exports) or not set(exports) <= set(names):
+        raise ValueError("Exports must be unique selected function names")
+    src, dst = ast.parse(source), ast.parse(destination)
+    src_bindings, dst_bindings = _module_bindings(src), _module_bindings(dst)
+    selected = []
+    for name in names:
+        matches = src_bindings.get(name, [])
+        if (
+            len(matches) != 1
+            or not isinstance(matches[0], (ast.FunctionDef, ast.AsyncFunctionDef))
+            or matches[0] not in src.body
+            or getattr(matches[0], "type_params", ())
+        ):
+            raise ValueError(f"Expected one top-level non-generic function: {name}")
+        if dst_bindings.get(name):
+            raise ValueError(
+                f"A selected function already exists in the destination: {name}"
+            )
+        node = matches[0]
+        headers = (
+            node.decorator_list
+            + node.args.defaults
+            + [value for value in node.args.kw_defaults if value is not None]
+        )
+        arguments = node.args.posonlyargs + node.args.args + node.args.kwonlyargs
+        arguments += [arg for arg in (node.args.vararg, node.args.kwarg) if arg]
+        headers += [arg.annotation for arg in arguments if arg.annotation is not None]
+        if node.returns is not None:
+            headers.append(node.returns)
+        if any(
+            isinstance(
+                child,
+                (
+                    ast.Lambda,
+                    ast.ListComp,
+                    ast.SetComp,
+                    ast.DictComp,
+                    ast.GeneratorExp,
+                ),
+            )
+            for header in headers
+            for child in ast.walk(header)
+        ):
+            raise ValueError(
+                "Scoped expressions in function headers require a full migration"
+            )
+        if any(isinstance(child, ast.ClassDef) for child in ast.walk(node)):
+            raise ValueError("Classes in selected functions require a full migration")
+        selected.append(node)
+    start = src.body.index(selected[0])
+    if src.body[start : start + len(names)] != selected or start + len(names) == len(
+        src.body
+    ):
+        raise ValueError("Functions must be contiguous and followed by an anchor")
+    first_line = min(
+        [selected[0].lineno, *(node.lineno for node in selected[0].decorator_list)]
+    )
+    anchor = src.body[start + len(names)]
+    matches = [node for node in dst.body if ast.dump(node) == ast.dump(anchor)]
+    if len(matches) != 1:
+        raise ValueError("Expected one unchanged destination anchor")
+    for tree, node in ((src, anchor), (dst, matches[0])):
+        index = tree.body.index(node)
+        if (
+            getattr(node, "decorator_list", ())
+            or node.col_offset
+            or index
+            and tree.body[index - 1].end_lineno >= node.lineno
+            or index + 1 < len(tree.body)
+            and tree.body[index + 1].lineno <= node.end_lineno
+        ):
+            raise ValueError(
+                "Anchor must be undecorated and occupy complete source lines"
+            )
+    if start and src.body[start - 1].end_lineno >= first_line:
+        raise ValueError("Selected functions must occupy complete source lines")
+
+    source_lines, lines = source.splitlines(True), destination.splitlines(True)
+    futures = [
+        [
+            node
+            for node in tree.body
+            if isinstance(node, ast.ImportFrom) and node.module == "__future__"
+        ]
+        for tree in (src, dst)
+    ]
+    if [ast.dump(node) for node in futures[0]] != [
+        ast.dump(node) for node in futures[1]
+    ]:
+        raise ValueError("Source and destination future imports differ")
+    prefix = "".join(
+        "".join(source_lines[node.lineno - 1 : node.end_lineno]) for node in futures[0]
+    )
+    if "*" in src_bindings or "*" in dst_bindings:
+        raise ValueError("Wildcard import makes dependency bindings ambiguous")
+
+    def dependency(name, source_boundary, destination_boundary):
+        if name in {"eval", "exec", "globals", "locals", "vars"}:
+            raise ValueError("Dynamic namespace dependencies require a full migration")
+        declarations = []
+        for tree, bindings, boundary in (
+            (src, src_bindings, source_boundary),
+            (dst, dst_bindings, destination_boundary),
+        ):
+            bound = bindings.get(name, [])
+            if not bound and name in vars(builtins):
+                declarations.append(None)
+                continue
+            if len(bound) != 1:
+                raise ValueError(f"Missing or ambiguous dependency binding: {name}")
+            candidates = [
+                node
+                for node in tree.body
+                if isinstance(node, (ast.Import, ast.ImportFrom))
+                and bound[0] in node.names
+                or isinstance(node, ast.Assign)
+                and len(node.targets) == 1
+                and node.targets[0] is bound[0]
+            ]
+            if len(candidates) != 1 or candidates[0].lineno >= boundary:
+                raise ValueError(
+                    f"Expected an earlier import or simple assignment: {name}"
+                )
+            declarations.append(candidates[0])
+        left, right = declarations
+        if left is None and right is None:
+            return
+        if left is None or right is None or ast.dump(left) != ast.dump(right):
+            raise ValueError(f"Changed dependency binding: {name}")
+        if isinstance(left, ast.Assign):
+            for child in ast.walk(left.value):
+                if isinstance(
+                    child,
+                    (
+                        ast.Lambda,
+                        ast.NamedExpr,
+                        ast.ListComp,
+                        ast.SetComp,
+                        ast.DictComp,
+                        ast.GeneratorExp,
+                    ),
+                ):
+                    raise ValueError(f"Unsupported dependency expression: {name}")
+                if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load):
+                    dependency(child.id, left.lineno, right.lineno)
+
+    earlier = set()
+    for node in selected:
+        line = min([node.lineno, *(item.lineno for item in node.decorator_list)])
+        immediate, deferred = _function_dependencies(
+            prefix + "".join(source_lines[line - 1 : node.end_lineno])
+        )
+        if (immediate & set(names)) - earlier:
+            raise ValueError("Definition-time dependency is not yet bound")
+        for name in (immediate | deferred) - set(names):
+            dependency(name, first_line, matches[0].lineno)
+        earlier.add(node.name)
+
+    # Keep a following anchor comment attached to its original statement. Only
+    # canonical function bytes and their trailing whitespace/comments are added.
+    end = anchor.lineno - 1
+    comment = next(
+        (
+            i
+            for i in range(selected[-1].end_lineno, end)
+            if source_lines[i].startswith("#")
+        ),
+        end,
+    )
+    anchor_comments = source_lines[comment:end]
+    offset = matches[0].lineno - 1 - len(anchor_comments)
+    if offset < 0 or lines[offset : matches[0].lineno - 1] != anchor_comments:
+        raise ValueError("Anchor's leading comments differ")
+    additions = [(offset, source_lines[first_line - 1 : comment])]
+    if exports:
+        for tree, bindings in ((src, src_bindings), (dst, dst_bindings)):
+            if bindings.get("__all__") != [
+                _single_assignment(tree, "__all__").targets[0]
+            ]:
+                raise ValueError("Ambiguous __all__ binding")
+        additions.append(
+            _selected_export_addition(src, dst, source_lines, lines, exports)
+        )
+    for offset, fragment in sorted(additions, reverse=True):
+        lines[offset:offset] = fragment
+    result = "".join(lines)
+    ast.parse(result)
+    return result
+
+
+def _migrate_selected(
+    path, names, exports, source_hash, destination_hash, kind, copier
+):
     target = parse_lib_path(path)
     if target.parts[:1] != ("Lib",) or ".." in target.parts or target.suffix != ".py":
-        raise ValueError("Selective assignments require an explicit Lib Python file")
+        raise ValueError("Selective imports require an explicit Lib Python file")
     source, destination = path.read_bytes(), target.read_bytes()
     for label, contents, expected in (
         ("source", source, source_hash),
@@ -269,16 +567,16 @@ def migrate_selected_assignments(path, names, exports, source_hash, destination_
     ):
         if hashlib.sha256(contents).hexdigest() != expected:
             raise ValueError(f"{label} SHA256 does not match the pinned input")
-    result = copy_selected_assignments(
+    result = copier(
         source.decode("utf-8"), destination.decode("utf-8"), names, exports
     ).encode("utf-8")
     if target.read_bytes() != destination:
-        raise ValueError("Destination changed during assignment preparation")
+        raise ValueError("Destination changed during selective import preparation")
     target.write_bytes(result)
     print(
         json.dumps(
             {
-                "assignments": names,
+                kind: names,
                 "exports": exports,
                 "source_sha256": source_hash,
                 "destination_before_sha256": destination_hash,
@@ -289,6 +587,30 @@ def migrate_selected_assignments(path, names, exports, source_hash, destination_
     )
 
 
+def migrate_selected_assignments(path, names, exports, source_hash, destination_hash):
+    return _migrate_selected(
+        path,
+        names,
+        exports,
+        source_hash,
+        destination_hash,
+        "assignments",
+        copy_selected_assignments,
+    )
+
+
+def migrate_selected_functions(path, names, exports, source_hash, destination_hash):
+    return _migrate_selected(
+        path,
+        names,
+        exports,
+        source_hash,
+        destination_hash,
+        "functions",
+        copy_selected_functions,
+    )
+
+
 def patch_single_content(
     src_path: pathlib.Path,
     lib_path: pathlib.Path,
@@ -781,6 +1103,11 @@ def main(argv: list[str] | None = None) -> int:
         nargs="+",
         help="Copy selected canonical assignments",
     )
+    parser.add_argument(
+        "--functions",
+        nargs="+",
+        help="Copy complete selected canonical functions without executing them",
+    )
     parser.add_argument("--exports", nargs="+", default=[])
     parser.add_argument("--source-sha256")
     parser.add_argument("--destination-sha256")
@@ -792,6 +1119,7 @@ def main(argv: list[str] | None = None) -> int:
                 args.legacy_map is not None
                 or args.patches is not None
                 or args.assignments
+                or args.functions
                 or args.exports
                 or args.source_sha256
                 or args.destination_sha256
@@ -808,15 +1136,25 @@ def main(argv: list[str] | None = None) -> int:
             return 0
         if args.dry_run:
             raise ValueError("--dry-run requires --canonical-removals")
-        if args.assignments:
-            if args.legacy_map is not None or args.patches is not None:
+        if args.assignments or args.functions:
+            if (
+                args.legacy_map is not None
+                or args.patches is not None
+                or args.assignments
+                and args.functions
+            ):
                 raise ValueError(
-                    "Selective imports cannot combine with --legacy-map or --patches"
+                    "Selective imports cannot combine with other migration modes"
                 )
             try:
-                migrate_selected_assignments(
+                migrate = (
+                    migrate_selected_functions
+                    if args.functions
+                    else migrate_selected_assignments
+                )
+                migrate(
                     args.path,
-                    args.assignments,
+                    args.functions or args.assignments,
                     args.exports,
                     args.source_sha256,
                     args.destination_sha256,
@@ -825,7 +1163,7 @@ def main(argv: list[str] | None = None) -> int:
                 raise ValueError(str(exc)) from exc
             return 0
         if args.exports or args.source_sha256 or args.destination_sha256:
-            raise ValueError("Selective options require --assignments")
+            raise ValueError("Selective options require --assignments or --functions")
         if args.patches is not None and (
             args.path.is_dir() or args.legacy_map is not None
         ):

scripts/update_lib/tests/test_selected_functions.py

@@ -0,0 +1,320 @@
+"""Source-only selective function migration, without importing copied helpers."""
+
+import contextlib
+import hashlib
+import io
+import json
+import pathlib
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from update_lib.cmd_migrate import copy_selected_functions, main
+
+
+class TestSelectedFunctions(unittest.TestCase):
+    names = ["available", "requires_available"]
+    exports = names
+    imports = "import functools\nimport os\nimport sys\nimport unittest\n\n"
+    export_line = '    "available", "requires_available",\n'
+    dependencies = 'IS_MOBILE = sys.platform == "mobile"\nAVAILABLE = not IS_MOBILE\n\n'
+    block = (
+        "@functools.cache\n"
+        "def available(limit=1):\n"
+        "    import errno\n"
+        "    try:\n"
+        '        return AVAILABLE and bool(errno.EINVAL) and os.name != "" and limit > 0\n'
+        "    except ValueError as error:\n"
+        "        return bool(str(error))\n\n\n"
+        "def requires_available():\n"
+        '    return unittest.skipUnless(available(), "needs availability")\n\n\n'
+    )
+    anchor = "# Keep this comment with STATUS.\nSTATUS = 1\n"
+    source = (
+        imports
+        + '__all__ = [\n    "existing",\n'
+        + export_line
+        + "]\n\n"
+        + dependencies
+        + block
+        + anchor
+    )
+    destination = (
+        imports
+        + '__all__ = [\n    "existing",\n]\n\n'
+        + dependencies
+        + 'def existing(sys):\n    return "keep π"\n\n'
+        + anchor
+    )
+
+    def copy(self, source=None, destination=None, names=None, exports=None):
+        return copy_selected_functions(
+            self.source if source is None else source,
+            self.destination if destination is None else destination,
+            self.names if names is None else names,
+            self.exports if exports is None else exports,
+        )
+
+    def test_exact_functions_exports_and_unrelated_bytes(self):
+        self.assertEqual(
+            self.copy(),
+            self.destination.replace(
+                "__all__ = [\n", "__all__ = [\n" + self.export_line
+            ).replace(self.anchor, self.block + self.anchor),
+        )
+        self.assertEqual(self.copy().count(self.anchor), 1)
+
+    def test_destination_line_endings_and_local_adaptations_survive(self):
+        source = self.source.replace(self.anchor, "STATUS = 1\n")
+        destination = self.destination.replace(self.anchor, "STATUS = 1\n").replace(
+            "\n", "\r\n"
+        )
+        self.assertEqual(
+            self.copy(source, destination),
+            destination.replace(
+                "__all__ = [\r\n", "__all__ = [\r\n" + self.export_line
+            ).replace("STATUS = 1\r\n", self.block + "STATUS = 1\r\n"),
+        )
+
+    def test_local_imports_closures_and_comprehensions_are_not_global_dependencies(
+        self,
+    ):
+        block = (
+            "@functools.cache\n"
+            "def available(items=()):\n"
+            "    def inner(value):\n"
+            "        return value in items\n"
+            "    return [inner(value) for value in items]\n\n\n"
+            "async def requires_available():\n"
+            "    import errno as local\n"
+            "    return local.EINVAL\n\n\n"
+        )
+        self.assertIn(block, self.copy(self.source.replace(self.block, block)))
+        with self.assertRaises(ValueError):
+            self.copy(
+                self.source.replace(
+                    self.block, block.replace("value in items", "MISSING", 1)
+                )
+            )
+
+    def test_multiple_decorators_defaults_and_body_are_copied_without_execution(self):
+        block = self.block.replace(
+            "@functools.cache\ndef available(limit=1):",
+            "@functools.cache\n@functools.lru_cache(\n    maxsize=None\n)\n"
+            "def available(limit=os.name):",
+        ).replace(
+            "    import errno\n",
+            "    raise RuntimeError('must not run')\n    import errno\n",
+        )
+        self.assertIn(block, self.copy(self.source.replace(self.block, block)))
+
+    def test_selection_and_existing_bindings_fail_closed(self):
+        for source, destination, names in (
+            (self.source, self.destination, []),
+            (self.source, self.destination, self.names * 2),
+            (self.source, self.destination, list(reversed(self.names))),
+            (self.source, self.destination, ["missing"]),
+            (self.source + "available = None\n", self.destination, self.names),
+            (self.source, self.destination + "del available\n", self.names),
+            (self.source, self.destination + "import other as available\n", self.names),
+            (
+                self.source,
+                self.destination + "if condition:\n    available = None\n",
+                self.names,
+            ),
+            (
+                self.source.replace(
+                    "def requires_available", "UNSELECTED = 1\ndef requires_available"
+                ),
+                self.destination,
+                self.names,
+            ),
+            (self.source.replace(self.anchor, ""), self.destination, self.names),
+        ):
+            with self.subTest(names=names), self.assertRaises(ValueError):
+                self.copy(source, destination, names)
+
+    def test_missing_changed_late_and_rebound_dependencies_fail_closed(self):
+        for destination in (
+            self.destination.replace("import functools\n", ""),
+            self.destination.replace(
+                "import functools\n", "import other as functools\n"
+            ),
+            self.destination.replace("import os\n", "") + "import os\n",
+            self.destination.replace(
+                'sys.platform == "mobile"', 'sys.platform == "other"'
+            ),
+            self.destination + "AVAILABLE = True\n",
+            self.destination + "del sys\n",
+            self.destination + "bool = None\n",
+            self.destination + "def rebind():\n    global sys\n    sys = None\n",
+            self.destination + "from unknown import *\n",
+        ):
+            with self.subTest(destination=destination), self.assertRaises(ValueError):
+                self.copy(destination=destination)
+        for source in (
+            self.source + "del sys\n",
+            self.source + "from unknown import *\n",
+            self.source.replace("AVAILABLE = not IS_MOBILE", "AVAILABLE = MISSING"),
+            self.source.replace("limit=1", "limit=requires_available()"),
+        ):
+            with self.subTest(source=source), self.assertRaises(ValueError):
+                self.copy(source=source)
+
+    def test_unsupported_namespace_changes_and_future_mismatch_fail_closed(self):
+        for statement in (
+            "global AVAILABLE\n    AVAILABLE = True",
+            "return globals()",
+            "return eval('AVAILABLE')",
+        ):
+            with self.subTest(statement=statement), self.assertRaises(ValueError):
+                self.copy(
+                    self.source.replace(
+                        "    import errno", "    " + statement + "\n    import errno"
+                    )
+                )
+        with self.assertRaises(ValueError):
+            self.copy("from __future__ import annotations\n" + self.source)
+        with self.assertRaises(ValueError):
+            self.copy(self.source.replace("limit=1", "limit=(EXTRA := 1)"))
+
+    def test_anchor_ambiguity_partial_lines_and_changed_comments_fail_closed(self):
+        for destination in (
+            self.destination.replace("STATUS = 1", "STATUS = 2"),
+            self.destination + "STATUS = 1\n",
+            self.destination.replace("STATUS = 1", "STATUS = 1; other = 2"),
+            self.destination.replace("STATUS = 1", "other = 2; STATUS = 1"),
+            self.destination.replace(
+                "# Keep this comment with STATUS.", "# Changed comment."
+            ),
+        ):
+            with self.subTest(destination=destination), self.assertRaises(ValueError):
+                self.copy(destination=destination)
+        replacement = "@functools.cache\ndef anchor(): pass\n"
+        with self.assertRaises(ValueError):
+            self.copy(
+                self.source.replace("STATUS = 1\n", replacement),
+                self.destination.replace("STATUS = 1\n", replacement),
+            )
+
+    def test_exports_require_complete_canonical_lines(self):
+        for exports in (
+            self.exports[:1],
+            list(reversed(self.exports)),
+            self.exports * 2,
+            ["other"],
+        ):
+            with self.subTest(exports=exports), self.assertRaises(ValueError):
+                self.copy(exports=exports)
+        with self.assertRaises(ValueError):
+            self.copy(destination=self.destination + "del __all__\n")
+        self.assertNotIn(self.export_line, self.copy(exports=[]))
+
+    def test_cli_hash_pins_receipt_and_concurrent_destination_guard(self):
+        with tempfile.TemporaryDirectory() as temp, contextlib.chdir(temp):
+            source = pathlib.Path("canonical/Lib/example.py")
+            destination = pathlib.Path("Lib/example.py")
+            source.parent.mkdir(parents=True)
+            destination.parent.mkdir(parents=True)
+            source.write_text(self.source)
+            destination.write_text(self.destination)
+            args = [
+                str(source),
+                "--functions",
+                *self.names,
+                "--exports",
+                *self.exports,
+                "--source-sha256",
+                hashlib.sha256(source.read_bytes()).hexdigest(),
+                "--destination-sha256",
+                "0" * 64,
+            ]
+            with contextlib.redirect_stderr(io.StringIO()):
+                self.assertEqual(main(args), 1)
+            self.assertEqual(destination.read_text(), self.destination)
+            args[-1] = hashlib.sha256(destination.read_bytes()).hexdigest()
+            with contextlib.redirect_stdout(io.StringIO()) as output:
+                self.assertEqual(main(args), 0)
+            receipt = json.loads(output.getvalue())
+            self.assertEqual(receipt["functions"], self.names)
+            self.assertEqual(destination.read_text(), self.copy())
+            destination.write_text(self.destination)
+
+            def concurrent_change(*unused):
+                destination.write_text("# changed meanwhile\n")
+                return self.copy()
+
+            with patch(
+                "update_lib.cmd_migrate.copy_selected_functions", concurrent_change
+            ):
+                with contextlib.redirect_stderr(io.StringIO()) as errors:
+                    self.assertEqual(main(args), 1)
+            self.assertIn("Destination changed", errors.getvalue())
+            self.assertEqual(destination.read_text(), "# changed meanwhile\n")
+
+    def test_cli_modes_are_exclusive_before_reading_inputs(self):
+        for options in (
+            ["--assignments", "value"],
+            ["--patches", "missing.json"],
+            ["--legacy-map", "missing.json"],
+            ["--canonical-removals", "missing.json"],
+        ):
+            with (
+                self.subTest(options=options),
+                contextlib.redirect_stderr(io.StringIO()) as errors,
+            ):
+                self.assertEqual(
+                    main(
+                        ["missing/Lib/example.py", "--functions", *self.names, *options]
+                    ),
+                    1,
+                )
+            self.assertIn("cannot combine", errors.getvalue())
+
+    def test_scoped_header_expressions_fail_closed(self):
+        for header in (
+            "def available(limit=(lambda: requires_available())()):",
+            "def available(limit=[requires_available() for _ in (0,)]):",
+            "def available(limit={requires_available() for _ in (0,)}):",
+            "def available(limit={0: requires_available() for _ in (0,)}):",
+            "def available(limit=(requires_available() for _ in (0,))):",
+            "def available(*, limit=(lambda: requires_available())()):",
+            "def available(limit: (lambda: requires_available())() = 1):",
+        ):
+            with (
+                self.subTest(header=header),
+                self.assertRaisesRegex(
+                    ValueError, "Scoped expressions in function headers"
+                ),
+            ):
+                self.copy(self.source.replace("def available(limit=1):", header))
+        for decorator in (
+            "@(lambda: requires_available())()",
+            "@[requires_available for _ in (0,)][0]",
+        ):
+            with (
+                self.subTest(decorator=decorator),
+                self.assertRaisesRegex(
+                    ValueError, "Scoped expressions in function headers"
+                ),
+            ):
+                self.copy(self.source.replace("@functools.cache", decorator))
+
+    def test_nested_classes_fail_closed(self):
+        for body in (
+            "    class Inner:\n        os = os\n",
+            "    def nested():\n        class Inner:\n            os = os\n",
+        ):
+            with (
+                self.subTest(body=body),
+                self.assertRaisesRegex(ValueError, "Classes in selected functions"),
+            ):
+                self.copy(
+                    self.source.replace(
+                        "    import errno\n", body + "    import errno\n"
+                    )
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()