Summary

Closes #1775. Supersedes #2071.

The three TestChildProcessCleanup tests in tests/client/test_stdio.py failed intermittently on Windows/macOS CI. The root cause was a fundamentally timing-dependent test design: watching a file grow on disk as a proxy for "is the child process alive?", with fixed anyio.sleep() durations to "wait long enough" for nested Python interpreters to start. On loaded CI, "long enough" wasn't.

A secondary bug compounded this: when the flaky assertion failed, proc was never terminated — the finally blocks only cleaned up tempfiles. The leaked subprocess was GC'd during a later test, triggering PytestUnraisableExceptionWarning there (observed: test_call_tool — an in-memory test that never touches subprocesses — failing with ResourceWarning: subprocess 5592 is still running).

New design — socket-based, zero sleeps, zero polling

Each subprocess in the tree connects a TCP socket back to a listener owned by the test. Two kernel-guaranteed blocking-I/O signals then tell us everything:

1. await listener.accept() blocks until the subprocess connects → proves it started. No polling for "did the file grow yet?".
2. await stream.receive(1) after _terminate_process_tree() raises anyio.EndOfStream because the kernel closes all file descriptors — including sockets — when a process terminates. This is a direct, OS-level proof that the child is dead. No polling for "did the file stop growing?".

Every synchronization point is a blocking read that unblocks on a kernel event. The anyio.fail_after() wrappers are upper bounds for catastrophic failure (subprocess never started / survived termination), not timing assumptions.

# Old: fixed sleep, indirect proxy, races against CI load
await anyio.sleep(0.5)
initial_size = os.path.getsize(marker_file)
await anyio.sleep(0.3)
assert os.path.getsize(marker_file) > initial_size  # fails: assert 0 > 0

# New: blocking accept, kernel-level EOF signal
with anyio.fail_after(10):
    stream = await sock.accept()           # blocks until child connects
    assert await stream.receive(5) == b"alive"

await _terminate_process_tree(proc)

with anyio.fail_after(5):
    await stream.receive(1)                # raises EndOfStream: child is dead

Secondary fix: process cleanup in finally

Each test now terminates proc in finally if the test body's own termination didn't run (guarded by proc = None after successful in-test termination to avoid noisy double-terminate). anyio.move_on_after(5) ensures cleanup can't mask the original failure.

Also removed

• tempfile import and all file I/O
• escape_path_for_python import (no longer need path escaping for embedded scripts — scripts are now injected via {code!r} repr() literals, which eliminates all nested-quote escaping)
• ~30 lines of boilerplate textwrap.dedent blocks with triply-nested f-string scripts
• The parent_marker check from test 1, which was a no-op: NamedTemporaryFile(delete=False) creates the file, so os.path.exists() was always True

Results

_{AI Disclaimer}