fix: stop timed-out node --test runs as a whole process tree

`node --test` runs every test file in a child process, and
`subprocess.run(timeout=)` kills only the runner, so a hung test file
outlives its runner with ppid 1. On 16.10.2026 four such orphans of a
hung mutation test grew to 77-456 GB within minutes and forced a
hardware reset of the host.

- tests/group_run.py: `run_in_group` starts a command in a new process
  group and, on timeout or any other exception, kills the group and
  every descendant through `platform_layer.kill_process_tree` before
  re-raising. The seven `subprocess.run([node, "--test", ...],
  timeout=...)` call sites use it with their timeouts, working
  directories and environments unchanged.
- tests/test_group_run.py: a CPU-only spinning `node:test` file proves
  both sides: nothing survives `run_in_group`, while bare
  `subprocess.run` leaves the file's child running on POSIX (the test
  then kills it by its exact PID).
- scripts/safe_test.py: optional `--timeout SECONDS`. Without it the
  launcher behaves as before. With it the command gets a session of its
  own; at the deadline its group and every descendant get SIGTERM,
  SIGKILL ten seconds later, and the launcher exits 124 like GNU
  `timeout`. A SIGINT, SIGTERM or SIGHUP that stops the launcher stops
  the command the same way; Windows force-stops the tree with
  `taskkill /T`.

Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>

scripts/safe_test.py

@@ -10,17 +10,32 @@ command, and exit status is not proof that they are gone, so the launcher keeps
 the new root in place and prints its exact path (`SAFE_TEST_RETAINED`); whoever
 later proves the run's processes gone removes it. Retention is ordinary, not a
 failure: the exit status is the command's own.
+
+`--timeout SECONDS` is a deadline that stops everything the command started, which
+`subprocess.run(timeout=)`, `perl -e 'alarm N; exec ...'` or killing the runner's PID
+do not: `node --test`, for one, keeps each test file in a child that outlives its
+runner. The command gets a session of its own; at the deadline its group and every
+descendant (captured first, whatever group it joined) get SIGTERM, SIGKILL ten
+seconds later, and the launcher exits 124 like GNU `timeout`. A SIGINT, SIGTERM or
+SIGHUP that stops the launcher stops the command the same way. Windows force-stops
+the tree at once with `taskkill /T`.
 """
 from __future__ import annotations
 
 import argparse
 import ast
 import json
+import os
 from pathlib import Path
 import runpy
+import signal
 import subprocess
 import sys
 import tempfile
+import time
+
+_KILL_AFTER_SECONDS = 10  # --timeout: from SIGTERM to SIGKILL
+_TIMEOUT_STATUS = 124  # GNU timeout's status for an expired deadline
 
 
 def _git_ancestor(path: Path):
@@ -31,14 +46,99 @@ def _git_ancestor(path: Path):
     return None
 
 
+def _descendants(root: int) -> list[int]:
+    """Every descendant of ``root`` in one ``ps`` snapshot, whatever process group it joined."""
+    try:
+        listing = subprocess.run(["ps", "-axo", "pid=,ppid="], capture_output=True, text=True,
+                                 timeout=10, check=True).stdout
+    except (OSError, subprocess.SubprocessError):
+        return []
+    children: dict[int, list[int]] = {}
+    for line in listing.splitlines():
+        fields = line.split()
+        if len(fields) == 2 and fields[0].isdigit() and fields[1].isdigit():
+            children.setdefault(int(fields[1]), []).append(int(fields[0]))
+    found: list[int] = []
+    frontier = [root]
+    while frontier:
+        for child in children.get(frontier.pop(), ()):
+            if child not in found:
+                found.append(child)
+                frontier.append(child)
+    return found
+
+
+def _signal_all(group: int, members: list[int], signum: int) -> None:
+    for send, target in ((os.killpg, group), *((os.kill, pid) for pid in members)):
+        try:
+            send(target, signum)
+        except OSError:  # already gone, or no longer ours to signal
+            pass
+
+
+def _exists(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except OSError:
+        pass  # EPERM: it exists, but is not ours
+    return True
+
+
+def _stop(process: subprocess.Popen) -> None:
+    """Stop the command's group and every descendant. The tree is captured before the first
+    signal: a parent that dies first orphans its children, and a test may have moved them
+    into groups of their own (``tests/group_run.py``)."""
+    if os.name == "nt":  # a console tree has no group SIGTERM
+        subprocess.call(["taskkill", "/F", "/T", "/PID", str(process.pid)],
+                        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    else:
+        for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
+            signal.signal(signum, signal.SIG_IGN)  # a second signal must not abandon a half-stopped tree
+        members = _descendants(process.pid)
+        _signal_all(process.pid, members, signal.SIGTERM)
+        grace = time.monotonic() + _KILL_AFTER_SECONDS
+        while time.monotonic() < grace and (process.poll() is None or any(map(_exists, members))):
+            time.sleep(0.1)
+        late = _descendants(process.pid) if process.poll() is None else []
+        _signal_all(process.pid, members + late, signal.SIGKILL)
+    process.wait()
+
+
+def _call_with_deadline(command: list[str], repo: Path, env: dict, seconds: float) -> int:
+    """``subprocess.call`` whose deadline, or a signal that stops the launcher, stops the whole tree."""
+    if os.name == "nt":
+        process = subprocess.Popen(command, cwd=repo, env=env,
+                                   creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
+    else:
+        process = subprocess.Popen(command, cwd=repo, env=env, start_new_session=True)
+        for signum in (signal.SIGTERM, signal.SIGHUP):  # the command's own session no longer gets them
+            signal.signal(signum, lambda received, _frame: sys.exit(128 + received))
+    try:
+        return process.wait(timeout=seconds)
+    except subprocess.TimeoutExpired:
+        print(f"SAFE_TEST_TIMEOUT after {seconds:g}s: stopping the command and every process it started",
+              file=sys.stderr, flush=True)
+        _stop(process)
+        return _TIMEOUT_STATUS
+    except BaseException:  # Ctrl-C, or SIGTERM/SIGHUP turned into SystemExit above
+        _stop(process)
+        raise
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--temp-parent", type=Path)
+    parser.add_argument("--timeout", type=float, metavar="SECONDS",
+                        help="stop the command and every process it started after SECONDS; exit 124")
     parser.add_argument("command", nargs=argparse.REMAINDER)
     args = parser.parse_args()
     command = args.command[1:] if args.command[:1] == ["--"] else args.command
     if not command:
         parser.error("supply a verification command after --")
+    if args.timeout is not None and not args.timeout > 0:
+        parser.error("--timeout must be a positive number of seconds")
     repo = Path(__file__).resolve().parents[1]
     helper = repo / "ouroboros" / "test_environment.py"
     tree = ast.parse(helper.read_text(encoding="utf-8"))
@@ -78,7 +178,9 @@ def main() -> int:
               or key.startswith("OUROBOROS_")}), flush=True)
         # Complete environment replacement reaches the interpreter BEFORE site,
         # plugins, conftest or any application module can import.
-        return subprocess.call(command, cwd=repo, env=env)
+        if args.timeout is None:
+            return subprocess.call(command, cwd=repo, env=env)
+        return _call_with_deadline(command, repo, env, args.timeout)
     finally:
         print(f"SAFE_TEST_RETAINED {root}", file=sys.stderr, flush=True)
 

tests/group_run.py

@@ -0,0 +1,55 @@
+"""``subprocess.run`` for test commands whose timeout must stop every process they started.
+
+``subprocess.run(timeout=...)`` kills only its direct child. ``node --test`` runs each
+test file in a child process of its own, so a timed-out runner leaves that child alive
+with ppid 1; on 16.10.2026 four such orphans of a hung test grew to 77-456 GB within
+minutes and forced a hardware reset. Here the command starts in a new process group, and
+a wait that ends in a timeout or any other exception kills the group and every
+descendant before the exception propagates.
+
+Outer supervisors reach these processes through the process tree, not through their
+own group: ``kill_process_tree`` and ``scripts/safe_test.py --timeout`` both do.
+"""
+from __future__ import annotations
+
+import subprocess
+import time
+
+from ouroboros.platform_layer import (
+    force_kill_pid, kill_process_tree, pid_provably_gone, subprocess_new_group_kwargs,
+)
+
+# After the kill only an escaped descendant can still hold a pipe; never wait on it forever.
+_DRAIN_SECONDS = 10
+
+
+def run_in_group(args, *, timeout: float, cwd=None, env=None) -> subprocess.CompletedProcess:
+    """Run ``args`` capturing UTF-8 text; on timeout kill the whole tree, then re-raise."""
+    with subprocess.Popen(args, cwd=cwd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                          encoding="utf-8", errors="replace", **subprocess_new_group_kwargs()) as process:
+        try:
+            stdout, stderr = process.communicate(timeout=timeout)
+        except subprocess.TimeoutExpired as expired:
+            kill_process_tree(process)
+            try:
+                expired.stdout, expired.stderr = process.communicate(timeout=_DRAIN_SECONDS)
+            except subprocess.TimeoutExpired:
+                pass
+            raise
+        except BaseException:  # an interrupt or pytest-timeout's signal must not orphan the tree either
+            kill_process_tree(process)
+            raise
+    return subprocess.CompletedProcess(args, process.returncode, stdout, stderr)
+
+
+def kill_survivors(pids, *, within: float = 10.0) -> list[int]:
+    """Give each PID ``within`` seconds to be provably gone; kill the rest by exact PID and return them."""
+    deadline = time.monotonic() + within
+    survivors = []
+    for pid in pids:
+        while not pid_provably_gone(pid) and time.monotonic() < deadline:
+            time.sleep(0.05)
+        if not pid_provably_gone(pid):
+            survivors.append(pid)
+            force_kill_pid(pid)
+    return survivors

tests/test_batch1_exact_consumers.py

@@ -235,11 +235,11 @@ def test_history_stamps_replay_evidence_on_a_latest_accounting_wait_checkpoint(t
 
 def test_review_reference_producer_history_and_chat_reconnect_keep_tool_evidence(tmp_path):
     import shutil
-    import subprocess
     from ouroboros.gateway.history import make_chat_history_endpoint
     from ouroboros.tool_call_log import append_call_row
     from ouroboros.tools.plan_review_references import _emit_review_reference
     from ouroboros.utils import append_jsonl
+    from tests.group_run import run_in_group
 
     task_id = "turn-a"
     append_jsonl(tmp_path / "logs/progress.jsonl", {"type": "send_message", "role": "assistant",
@@ -264,9 +264,9 @@ def test_review_reference_producer_history_and_chat_reconnect_keep_tool_evidence
     history.write_bytes(response.body)
     node = shutil.which("node")
     assert node, "Node is required for the real Chat consumer proof"
-    result = subprocess.run([node, "--test", "--test-name-pattern=reference carriers",
+    result = run_in_group([node, "--test", "--test-name-pattern=reference carriers",
         "web/tests/chat_activity_block.test.js"], cwd=pathlib.Path(__file__).resolve().parents[1],
-        env={**os.environ, "OURO_TEST_TOOL_HISTORY": str(history)}, capture_output=True, text=True, timeout=30)
+        env={**os.environ, "OURO_TEST_TOOL_HISTORY": str(history)}, timeout=30)
     assert result.returncode == 0, result.stdout + result.stderr
 
 

tests/test_batch4_launch_completion.py

@@ -320,13 +320,13 @@ def test_enqueue_snapshot_and_first_reservation_keep_original_billing_without_li
 def test_focused_node_control_consumers():
     import pathlib
     import shutil
-    import subprocess
+    from tests.group_run import run_in_group
     root = pathlib.Path(__file__).resolve().parents[1]
     node = shutil.which('node')
     assert node, 'focused frontend verification requires the provided Node runtime'
-    result = subprocess.run([node, '--test', 'web/tests/task_continue.test.js',
-                             'web/tests/chat_activity_block.test.js', 'web/tests/activity_schedule_time.test.js'],
-                            cwd=root, capture_output=True, text=True, timeout=60)
+    result = run_in_group([node, '--test', 'web/tests/task_continue.test.js',
+                           'web/tests/chat_activity_block.test.js', 'web/tests/activity_schedule_time.test.js'],
+                          cwd=root, timeout=60)
     print(result.stdout)
     assert result.returncode == 0, result.stdout + result.stderr
 

tests/test_group_run.py

@@ -0,0 +1,61 @@
+"""A timed-out ``node --test`` run leaves no process behind under ``tests.group_run``.
+
+``node --test`` runs each test file in a child process. Bare ``subprocess.run(timeout=)``
+kills only the runner, so on POSIX that child survives with ppid 1; the falsifying test
+proves the check sees such a survivor, then kills it by its exact PID. The spinning file
+only burns CPU and never allocates.
+"""
+import os
+import shutil
+import subprocess
+import sys
+import time
+
+import pytest
+
+from ouroboros.platform_layer import force_kill_pid, pid_provably_gone
+from tests.group_run import kill_survivors, run_in_group
+
+NODE = shutil.which("node")
+pytestmark = [pytest.mark.serial,  # real processes, killed by PID
+              pytest.mark.skipif(not NODE, reason="node is not installed")]
+
+SPIN = """\
+const fs = require('node:fs');
+const test = require('node:test');
+test('spins without allocating', () => {
+  fs.writeFileSync(process.env.SPIN_PIDS, `${process.pid} ${process.ppid}`);
+  for (;;) {}
+});
+"""
+
+
+def _spin(tmp_path, launch):
+    """Time out a spinning test file; return the PIDs of its file child and of the runner."""
+    spin, pids = tmp_path / "spin.test.cjs", tmp_path / "pids"
+    spin.write_text(SPIN, encoding="utf-8")
+    with pytest.raises(subprocess.TimeoutExpired):
+        launch([NODE, "--test", str(spin)], cwd=tmp_path, env={**os.environ, "SPIN_PIDS": str(pids)}, timeout=3)
+    deadline = time.monotonic() + 5  # a surviving child may still be writing
+    while not (pids.exists() and len(pids.read_text(encoding="utf-8").split()) == 2):
+        assert time.monotonic() < deadline, "node did not start the spinning test file before the timeout"
+        time.sleep(0.05)
+    child, runner = map(int, pids.read_text(encoding="utf-8").split())
+    return child, runner
+
+
+def test_timeout_stops_the_test_file_child_with_its_runner(tmp_path):
+    assert not kill_survivors(_spin(tmp_path, run_in_group))
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="libuv's job object ends node's children with their parent")
+def test_bare_run_timeout_leaves_the_test_file_child_running(tmp_path):
+    child, runner = _spin(tmp_path, lambda args, **kwargs: subprocess.run(args, capture_output=True, **kwargs))
+    try:
+        assert not kill_survivors([runner])
+        time.sleep(0.5)
+        assert not pid_provably_gone(child), "the orphan check no longer sees a survivor"
+    finally:
+        if not pid_provably_gone(child):
+            force_kill_pid(child)
+    assert not kill_survivors([child])

tests/test_main_durable_edges.py

@@ -53,14 +53,14 @@ def _history(root, chat_id=1):
 def _consume_payload(root, payload, pattern):
     import pathlib
     import shutil
-    import subprocess
+    from tests.group_run import run_in_group
     path = root / 'main-consumer-history.json'
     path.write_text(json.dumps(payload))
     node = shutil.which('node')
     assert node, 'Node is required for the actual Main consumer regression'
-    result = subprocess.run([node, '--test', '--test-force-exit', f'--test-name-pattern={pattern}', 'web/tests/main_durable_edges.test.js'],
+    result = run_in_group([node, '--test', '--test-force-exit', f'--test-name-pattern={pattern}', 'web/tests/main_durable_edges.test.js'],
         cwd=pathlib.Path(__file__).resolve().parents[1], env={**os.environ, 'MAIN_REVIEW_HISTORY_PAYLOAD': str(path)},
-        capture_output=True, text=True, timeout=30)
+        timeout=30)
     (root / 'consumer.log').write_text(result.stdout + result.stderr)
     assert result.returncode == 0, result.stdout + result.stderr
 

tests/test_merge_receipt_qualification.py

@@ -7,7 +7,6 @@ import json
 import os
 import pathlib
 import shutil
-import subprocess
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor
 from threading import Event
@@ -21,6 +20,7 @@ from tests import test_pr_merge_receipts as merge_fixture
 from tests.test_pr_merge_receipts import BASE, HEAD, MERGE, TREE, _receipts
 from tests._store_shared import delete_result, read_result, seed_result, unreadable_result
 from tests._cancel_intents_shared import DeliveryBus
+from tests.group_run import run_in_group
 
 world = merge_fixture.world
 
@@ -239,11 +239,10 @@ def test_concurrent_publication_through_outbox_dedup_and_history(world, monkeypa
     fixture.write_text(json.dumps({"live": chats, "history": rows}), encoding="utf-8")
     node = shutil.which("node")
     assert node, "node is required for the real live/replay projection qualification"
-    result = subprocess.run([node, "--test", "--test-name-pattern=canonical receipt revisions",
-                             "web/tests/chat_card_row_placement.test.js"],
-                            cwd=pathlib.Path(__file__).resolve().parents[1],
-                            env={**os.environ, "MERGE_RECEIPT_PROJECTION": str(fixture)},
-                            capture_output=True, text=True, timeout=30)
+    result = run_in_group([node, "--test", "--test-name-pattern=canonical receipt revisions",
+                           "web/tests/chat_card_row_placement.test.js"],
+                          cwd=pathlib.Path(__file__).resolve().parents[1],
+                          env={**os.environ, "MERGE_RECEIPT_PROJECTION": str(fixture)}, timeout=30)
     assert result.returncode == 0, result.stdout + result.stderr
     assert sum(c[:2] == ["pr", "merge"] for c in world.gh.calls) == 1
 
@@ -361,10 +360,9 @@ def bounded_receipt_history(world, monkeypatch):
 def test_cold_bounded_history_supplies_current_receipt_to_new_chat(world, monkeypatch):
     payload = bounded_receipt_history(world, monkeypatch)
     fixture = world.root / "cold-history.json"
-    result = subprocess.run([shutil.which("node"), "--test", "--test-name-pattern=cold bounded receipt",
-                             "web/tests/chat_card_row_placement.test.js"],
-                            env={**os.environ, "COLD_RECEIPT_PROJECTION": str(fixture)},
-                            capture_output=True, text=True, timeout=30)
+    result = run_in_group([shutil.which("node"), "--test", "--test-name-pattern=cold bounded receipt",
+                           "web/tests/chat_card_row_placement.test.js"],
+                          env={**os.environ, "COLD_RECEIPT_PROJECTION": str(fixture)}, timeout=30)
     assert result.returncode == 0, result.stdout + result.stderr
     (receipt,) = [r for r in payload["messages"] if r.get("card_row_id")]
     assert receipt["card_row_revision"] == 3 and "merge: merged" in receipt["text"]

tests/test_project_child_words_placement.py

@@ -23,7 +23,6 @@ import json
 import os
 import pathlib
 import shutil
-import subprocess
 from collections import deque
 from types import SimpleNamespace
 
@@ -37,6 +36,7 @@ from ouroboros.tools.registry import ToolContext, ToolRegistry
 from ouroboros.utils import append_jsonl
 from supervisor import events_chat_delivery as delivery
 from supervisor import message_bus
+from tests.group_run import run_in_group
 
 pytestmark = pytest.mark.serial  # Access is read from the process environment
 
@@ -221,10 +221,9 @@ def test_a_routing_notice_for_a_child_stands_alone_and_the_childs_words_stay_in_
                                    "running_history": notice_read["messages"]}), encoding="utf-8")
     node = shutil.which("node")
     assert node, "node is required for the real chat.js live/replay consumer"
-    result = subprocess.run([node, "--test", "--test-reporter=tap", "--test-name-pattern=child words",
-                             "web/tests/chat_card_row_placement.test.js"],
-                            cwd=REPO, env={**os.environ, "CHILD_WORDS_PROJECTION": str(fixture)},
-                            capture_output=True, text=True, timeout=60)
+    result = run_in_group([node, "--test", "--test-reporter=tap", "--test-name-pattern=child words",
+                           "web/tests/chat_card_row_placement.test.js"],
+                          cwd=REPO, env={**os.environ, "CHILD_WORDS_PROJECTION": str(fixture)}, timeout=60)
     assert result.returncode == 0 and "# pass 7\n" in result.stdout, result.stdout + result.stderr
 
 

tests/test_store_migrating.py

@@ -3,13 +3,13 @@ import asyncio
 import json
 import pathlib
 import shutil
-import subprocess
 from types import SimpleNamespace
 
 import pytest
 
 from ouroboros import store, store_import, task_status
 from ouroboros.gateway import history
+from tests.group_run import run_in_group
 from tests.test_history_tool_carriers import TS, _tool, _write
 
 
@@ -55,7 +55,6 @@ def test_migrating_client_consumers():
     node = shutil.which("node")
     assert node, "focused frontend verification requires Node"
     files = ["chat_history_loading", "dashboard_read_states", "task_card_recovery", "api_client_errors"]
-    result = subprocess.run([node, "--test", *[f"web/tests/{name}.test.js" for name in files]],
-                            cwd=root, capture_output=True, text=True, encoding="utf-8", timeout=60)
+    result = run_in_group([node, "--test", *[f"web/tests/{name}.test.js" for name in files]], cwd=root, timeout=60)
     print(result.stdout)
     assert result.returncode == 0, result.stdout + result.stderr

tests/test_test_environment.py

@@ -1,15 +1,20 @@
 """Real parent/child writes and concurrent runs stay on disposable roots."""
 from concurrent.futures import ThreadPoolExecutor
 import json
+import os
 from pathlib import Path
+import signal
 import subprocess
 import sys
+import time
 
 import pytest
 
+from ouroboros.platform_layer import kill_process_tree
 from ouroboros.preflight_runner import _preflight_env
 from ouroboros.settings_defaults import settings_env_keys
 from ouroboros.test_environment import isolated_environment, settings_keys
+from tests.group_run import kill_survivors, run_in_group
 
 REPO = Path(__file__).resolve().parents[1]
 pytestmark = pytest.mark.serial
@@ -199,6 +204,72 @@ def test_safe_test_retains_its_new_root_and_keeps_the_command_status(tmp_path):
     assert Path(result.stdout.splitlines()[-1]) == root
 
 
+# The command keeps a child in a session of its own, as tests/group_run.py keeps node's runner;
+# a "stubborn" command also ignores SIGTERM, so only the SIGKILL that follows can stop it.
+_TREE = """\
+import os, pathlib, signal, subprocess, sys, time
+own = {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} if os.name == "nt" else {"start_new_session": True}
+child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(600)"], **own)
+if os.name != "nt" and sys.argv[2] == "stubborn":
+    signal.signal(signal.SIGTERM, signal.SIG_IGN)
+pathlib.Path(sys.argv[1]).write_text(f"{os.getpid()} {child.pid}")
+time.sleep(600)
+"""
+_LAUNCHER = [sys.executable, "-I", "-S", str(REPO / "scripts" / "safe_test.py")]
+
+
+def _recorded_tree(path, within=30.0):
+    deadline = time.monotonic() + within
+    while not (path.exists() and len(path.read_text(encoding="utf-8").split()) == 2):
+        assert time.monotonic() < deadline, "the command never recorded its tree"
+        time.sleep(0.05)
+    return [int(pid) for pid in path.read_text(encoding="utf-8").split()]
+
+
+def test_safe_test_timeout_stops_the_whole_tree_and_exits_124(tmp_path):
+    """The deadline reaches a child outside the command's group and a command deaf to SIGTERM."""
+    pids = tmp_path / "pids"
+    result = run_in_group(_LAUNCHER + ["--temp-parent", str(tmp_path), "--timeout", "3", "--",
+                                       sys.executable, "-c", _TREE, str(pids), "stubborn"], cwd=REPO, timeout=120)
+    survivors = kill_survivors(_recorded_tree(pids))
+    assert result.returncode == 124, result.stderr
+    assert "SAFE_TEST_TIMEOUT after 3s" in result.stderr
+    assert sum(line.startswith("SAFE_TEST_RETAINED ") for line in result.stderr.splitlines()) == 1
+    assert not survivors
+
+
+def test_safe_test_timeout_keeps_a_finished_command_status(tmp_path):
+    result = run_in_group(_LAUNCHER + ["--temp-parent", str(tmp_path), "--timeout", "60", "--",
+                                       sys.executable, "-c", "import sys; sys.exit(7)"], cwd=REPO, timeout=120)
+    assert result.returncode == 7, result.stderr
+    assert "SAFE_TEST_TIMEOUT" not in result.stderr and "SAFE_TEST_RETAINED " in result.stderr
+    refused = run_in_group(_LAUNCHER + ["--timeout", "0", "--", sys.executable, "-c", "pass"],
+                           cwd=REPO, timeout=120)
+    assert refused.returncode == 2 and "positive number of seconds" in refused.stderr
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals")
+def test_safe_test_timeout_passes_sigterm_on_to_the_whole_tree(tmp_path):
+    """The command's own session no longer gets what stops the launcher, so the launcher passes it on."""
+    pids = tmp_path / "pids"
+    launcher = subprocess.Popen(_LAUNCHER + ["--temp-parent", str(tmp_path), "--timeout", "600", "--",
+                                             sys.executable, "-c", _TREE, str(pids), "obedient"],
+                                cwd=REPO, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+                                start_new_session=True)
+    try:
+        tree = _recorded_tree(pids)
+        os.kill(launcher.pid, signal.SIGTERM)
+        _, stderr = launcher.communicate(timeout=60)
+    finally:
+        if launcher.poll() is None:
+            kill_process_tree(launcher)
+            launcher.wait()
+    survivors = kill_survivors(tree)
+    assert launcher.returncode == 128 + signal.SIGTERM, stderr
+    assert "SAFE_TEST_RETAINED " in stderr
+    assert not survivors
+
+
 def _live_conftest():
     return next(module for module in list(sys.modules.values())
                 if getattr(module, "__file__", "")