Fork-per-mutant inherits the parent's dead state; async tests are recorded as timeouts instead of their verdicts #578

# mylib/core.py
import asyncio
from concurrent.futures import ThreadPoolExecutor

_pool = ThreadPoolExecutor(max_workers=1)


async def double(x: int) -> int:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(_pool, lambda: x * 2)

# tests/test_core.py
from mylib.core import double


async def test_double():
    assert await double(0) == 0

import os
from concurrent.futures import ThreadPoolExecutor, TimeoutError

pool = ThreadPoolExecutor(max_workers=1)
print("parent:", pool.submit(lambda: 1 * 2).result(timeout=5))

if os.fork() == 0:
    fut = pool.submit(lambda: 1 * 2)
    print("child threads:", len(pool._threads), "alive:", [t.is_alive() for t in pool._threads], flush=True)
    try:
        print("child:", fut.result(timeout=3), flush=True)
    except TimeoutError:
        print("child: still queued after 3s", flush=True)
    os._exit(0)
os.wait()

# forkprobe.py -- run with PYTHONPATH=. python forkprobe.py, against the files above
import faulthandler, os, signal, sys, time
import pytest

args = ["-q", "-p", "no:cacheprovider", "tests"]
print("=== parent run ===", flush=True)
print("parent exit:", pytest.main(args), flush=True)

pid = os.fork()
if pid == 0:
    dump = open("child-trace.txt", "w")
    faulthandler.register(signal.SIGUSR1, file=dump)
    os._exit(int(pytest.main(args)))

deadline = time.time() + 20
while time.time() < deadline:
    done, status = os.waitpid(pid, os.WNOHANG)
    if done:
        print("=== child exited:", os.waitstatus_to_exitcode(status), flush=True)
        sys.exit(0)
    time.sleep(0.2)
print("=== child still running after 20s; SIGUSR1 traceback follows ===", flush=True)
os.kill(pid, signal.SIGUSR1)
time.sleep(1)
os.kill(pid, signal.SIGKILL)
os.waitpid(pid, 0)
sys.exit(1)