diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml
index 3bd538d9..a00c31ee 100644
--- a/.github/workflows/test.yaml
+++ b/.github/workflows/test.yaml
@@ -72,6 +72,12 @@ jobs:
         name: Install libjpeg on Ubuntu
         run: sudo apt-get install -y libjpeg-dev
 
+      - name: Install setsid on macOS
+        if: runner.os == 'macOS'
+        run: |
+          brew install util-linux
+          echo "$(brew --prefix util-linux)/bin" >> "$GITHUB_PATH"
+
       - name: Install uv
         if: matrix.toxenv == 'oldestdeps'
         run: python -m pip install uv
diff --git a/.gitignore b/.gitignore
index 670257ea..289e11e9 100644
--- a/.gitignore
+++ b/.gitignore
@@ -2,6 +2,7 @@ __pycache__/
 src/__pycache__/
 *.egg-info/
 .coverage
+.coverage.*
 coverage.*
 .duct/
 .idea
diff --git a/CONTRIBUTING.rst b/CONTRIBUTING.rst
index 3f338e13..2b0c211b 100644
--- a/CONTRIBUTING.rst
+++ b/CONTRIBUTING.rst
@@ -59,3 +59,9 @@ It is also possible to use pytest directly, but you will need to override the op
 `tox.ini` with `addopts`. Invocation in this way can be faster while developing::
 
     python -m pytest -o addopts="" test/test_thething.py::specific_test
+
+Some tests need the ``setsid`` command and are skipped where it is missing.
+macOS does not ship it; to run those tests there, install util-linux and put it on ``PATH``::
+
+    brew install util-linux
+    export PATH="$(brew --prefix util-linux)/bin:$PATH"
diff --git a/pyproject.toml b/pyproject.toml
index 16377161..39016b42 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -77,9 +77,6 @@ filterwarnings = [
     # matplotlib 3.5 uses deprecated pyparsing API (enablePackrat -> enable_packrat)
     "ignore:'enablePackrat' deprecated:DeprecationWarning:matplotlib._mathtext",
 ]
-markers = [
-    "flaky: mark a test as being unreliable",
-]
 norecursedirs = ["test/data"]
 
 [tool.setuptools.packages.find]
@@ -117,6 +114,7 @@ src_paths = ["src"]
 [tool.coverage.run]
 branch = true
 parallel = true
+patch = ["subprocess", "execv"]
 
 [tool.coverage.paths]
 source = [
diff --git a/src/con_duct/_duct_main.py b/src/con_duct/_duct_main.py
index e2954927..89d3d8f2 100644
--- a/src/con_duct/_duct_main.py
+++ b/src/con_duct/_duct_main.py
@@ -9,7 +9,7 @@
 from typing import IO, TextIO
 from con_duct._models import LogPaths, Outputs, RecordTypes, SessionMode
 from con_duct._output import TailPipe, prepare_outputs, remove_files, safe_close_files
-from con_duct._signals import SigIntHandler
+from con_duct._signals import SigIntHandler, start_thread
 from con_duct._tracker import Report, monitor_process
 
 __version__ = version("con-duct")
@@ -65,6 +65,11 @@ def execute(
             sample_interval,
         )
 
+    # Install before any command exists so no SIGINT can kill duct and orphan
+    # the command; until attach() the handler only counts them.
+    sigint_handler = SigIntHandler()
+    signal.signal(signal.SIGINT, sigint_handler)
+
     log_paths = LogPaths.create(output_prefix, pid=os.getpid())
     try:
         log_paths.prepare_paths(clobber, capture_outputs)
@@ -99,6 +104,12 @@ def execute(
     )
     files_to_close.append(report.usage_file)
 
+    if sigint_handler.sigcount:
+        safe_close_files(files_to_close)
+        remove_files(log_paths, assert_empty=True)
+        lgr.info("Received SIGINT before starting %r, not starting it", command)
+        return 128 + signal.SIGINT
+
     report.start_time = time.time()
     try:
         report.process = process = subprocess.Popen(
@@ -118,7 +129,7 @@ def execute(
         lgr.error("%s: command not found", command)
         return 127  # seems what zsh and bash return then
 
-    signal.signal(signal.SIGINT, SigIntHandler(process.pid))
+    sigint_handler.attach(process.pid)
     lgr.info("duct %s is executing %r...", __version__, full_command)
     lgr.info("Log files will be written to %s", log_paths.prefix)
     try:
@@ -145,15 +156,15 @@ def execute(
         monitoring_thread = threading.Thread(
             target=monitor_process, args=monitoring_args
         )
-        monitoring_thread.start()
+        start_thread(monitoring_thread)
     else:
         monitoring_thread = None
 
     if record_types.has_system_summary():
         env_thread = threading.Thread(target=report.collect_environment)
-        env_thread.start()
+        start_thread(env_thread)
         sys_info_thread = threading.Thread(target=report.get_system_info)
-        sys_info_thread.start()
+        start_thread(sys_info_thread)
     else:
         env_thread, sys_info_thread = None, None
 
diff --git a/src/con_duct/_output.py b/src/con_duct/_output.py
index d82bd658..ed90b30f 100644
--- a/src/con_duct/_output.py
+++ b/src/con_duct/_output.py
@@ -9,6 +9,7 @@
 import time
 from typing import IO, Any, TextIO
 from con_duct._models import LogPaths, Outputs
+from con_duct._signals import start_thread
 
 
 class TailPipe:
@@ -27,7 +28,7 @@ def start(self) -> None:
         self.stop_event = threading.Event()
         self.infile = open(self.file_path, "rb")
         self.thread = threading.Thread(target=self._tail, daemon=True)
-        self.thread.start()
+        start_thread(self.thread)
 
     def fileno(self) -> int:
         assert self.infile is not None
diff --git a/src/con_duct/_signals.py b/src/con_duct/_signals.py
index 48d66c51..2a7ccbc8 100644
--- a/src/con_duct/_signals.py
+++ b/src/con_duct/_signals.py
@@ -1,41 +1,138 @@
 """Signal handlers for con-duct."""
 
 from __future__ import annotations
+from collections.abc import Iterator
+from contextlib import contextmanager
 import logging
 import os
 import signal
+import threading
 from types import FrameType
 from typing import Optional
 
 lgr = logging.getLogger("con-duct")
 
 
+@contextmanager
+def _sigint_blocked() -> Iterator[None]:
+    """Block SIGINT in the calling thread, restoring its mask on exit.
+
+    A SIGINT that arrives meanwhile stays pending and is handled once the
+    mask is restored.
+    """
+    old_mask = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGINT})
+    try:
+        yield
+    finally:
+        signal.pthread_sigmask(signal.SIG_SETMASK, old_mask)
+
+
+def start_thread(thread: threading.Thread) -> None:
+    """Start a thread that the kernel will never hand a SIGINT to.
+
+    A signal sent to the process is delivered to any one thread that does
+    not block it, but Python runs the handler only in the main thread. If
+    another thread takes a SIGINT, the main thread stays asleep in its wait
+    for the command and the Ctrl-C is not forwarded until the command exits.
+    A new thread inherits its creator's signal mask, so block SIGINT just
+    while starting it. Every thread duct starts must be started this way.
+
+    Parameters
+    ----------
+    thread : threading.Thread
+        The thread to start
+    """
+    with _sigint_blocked():
+        thread.start()
+
+
 class SigIntHandler:
     """
     Handler of SIGINT signals received by the process running duct.
     """
 
-    def __init__(self, pid: int) -> None:
-        """
+    def __init__(self) -> None:
+        # Installed before the command starts; SIGINTs are only counted
+        # until attach() says which process to forward them to.
+        self.pid: Optional[int] = None
+        self.sigcount: int = 0
+
+    def attach(self, pid: int) -> None:
+        """Start forwarding SIGINTs to the command.
+
         Parameters
         ----------
         pid : int
-            The PID of the process monitored by duct
+            The PID of the process monitored by duct. A SIGINT that arrived
+            while it was being started is acted on now.
         """
-        self.pid: int = pid
-        self.sigcount: int = 0
+        # A SIGINT after publishing the PID must not be forwarded both by
+        # __call__ and by the pending-signal check below.
+        with _sigint_blocked():
+            self.pid = pid
+            if self.sigcount:
+                self._act()
 
     def __call__(self, _sig: int, _frame: Optional[FrameType]) -> None:
         self.sigcount += 1
+        if self.pid is not None:
+            self._act()
+
+    def _act(self) -> None:
+        try:
+            self._forward()
+        except ProcessLookupError:
+            # The command has already exited. Python runs this handler in
+            # the main thread at its next bytecode, which can be only after
+            # the wait for the command returns.
+            _log(
+                logging.WARNING,
+                "Received SIGINT, but the command has already exited",
+            )
+
+    def _forward(self) -> None:
+        assert self.pid is not None
+        # Act before logging: the handler can interrupt duct mid-write to
+        # stderr, and a failed log write must not stop the signal.
         if self.sigcount == 1:
-            lgr.info("Received SIGINT, passing to command")
             os.kill(self.pid, signal.SIGINT)
+            _log(logging.INFO, "Received SIGINT, passing to command")
         elif self.sigcount == 2:
-            lgr.info("Received second SIGINT, again passing to command")
             os.kill(self.pid, signal.SIGINT)
+            _log(logging.INFO, "Received second SIGINT, again passing to command")
         elif self.sigcount == 3:
-            lgr.warning("Received third SIGINT, forcefully killing command process")
             os.kill(self.pid, signal.SIGKILL)
+            _log(
+                logging.WARNING,
+                "Received third SIGINT, forcefully killing command process",
+            )
         elif self.sigcount >= 4:
-            lgr.critical("Exiting duct, skipping cleanup")
+            _log(logging.CRITICAL, "Exiting duct, skipping cleanup")
             os._exit(1)
+
+
+def _log(level: int, msg: str) -> None:
+    """Log from a signal handler without ever raising.
+
+    If the signal arrived while duct was writing to stderr, logging raises
+    ``RuntimeError: reentrant call`` (even its error report does): the
+    buffered writer is held by the write the handler interrupted, which
+    cannot finish until the handler returns. Write the message straight to
+    file descriptor 2 instead, if the log level lets the message through.
+    It does not go through the log format, and it may land in the middle of
+    the log line that was being written, so it starts on a new line and says
+    why it looks different.
+    """
+    try:
+        lgr.log(level, msg)
+    except RuntimeError:
+        if not lgr.isEnabledFor(level):
+            return
+        try:
+            line = (
+                f"\ncon-duct [{logging.getLevelName(level)}] (written directly: "
+                f"the signal interrupted a log write): {msg}\n"
+            )
+            os.write(2, line.encode())
+        except OSError:
+            pass
diff --git a/test/data/signal_ignorer.py b/test/data/signal_ignorer.py
index 0f6fe066..3d1191fa 100755
--- a/test/data/signal_ignorer.py
+++ b/test/data/signal_ignorer.py
@@ -1,12 +1,25 @@
 #!/usr/bin/env python3
+"""Ignore SIGINT, reporting progress through the file given as argv[1].
+
+Writes "ready" once the handler is installed, then one "SIGINT" line per
+SIGINT received, so a test can wait for each step instead of sleeping.
+"""
+
 import signal
+import sys
 import time
 from types import FrameType
 from typing import Optional
 
 
+def report(line: str) -> None:
+    with open(sys.argv[1], "a") as f:
+        f.write(line + "\n")
+
+
 def handle_signal(sig: int, _frame: Optional[FrameType]) -> None:
     print(f"Received {sig}")
+    report("SIGINT")
 
 
 if __name__ == "__main__":
@@ -14,6 +27,7 @@ def handle_signal(sig: int, _frame: Optional[FrameType]) -> None:
     signal.siginterrupt(
         signal.SIGINT, False
     )  # Restart interrupted system calls so we can test multiple SIGINTS
+    report("ready")
     t0 = time.time()
-    while time.time() - t0 < 10:
+    while time.time() - t0 < 60:
         time.sleep(0.01)
diff --git a/test/data/spawn_children.sh b/test/data/spawn_children.sh
index eab11e1b..c7a2cec2 100755
--- a/test/data/spawn_children.sh
+++ b/test/data/spawn_children.sh
@@ -1,28 +1,80 @@
 #!/bin/bash
+# Spawn children in various ways, then exit once duct has had its chance to
+# see them, so the test does not depend on how fast duct samples.
+# Usage: spawn_children.sh MODE NCHILDREN USAGE_FILE STARTED_DIR
+# Each child records its pid as a file in STARTED_DIR, so the test can tell
+# exactly which pids to look for in the usage file.
 
 mode=$1
 nchildren=$2
-sleeptime=$3
+usage_file=$3
+started=$4
+until_sampled="$(dirname "$0")/until_sampled.sh"
 
-# Ensure at least one background process for `wait` to track, we want this to take longer than the children.
-long_sleep=$(awk "BEGIN {print $sleeptime * 1.5}")
-sleep "$long_sleep" &
+# Each child records its pid, then becomes until_sampled.sh (exec keeps the
+# pid), so it stays alive until duct has recorded it.
+child=(sh -c 'touch "$1/$$"; exec "$2" "$3"' _ "$started" "$until_sampled" "$usage_file")
+
+# Give up after 60 s rather than hang if duct never samples.
+deadline=$((SECONDS + 60))
+wait_until() {
+    until "$@"; do
+        if [ "$SECONDS" -ge "$deadline" ]; then
+            echo "spawn_children: timed out waiting for: $*" >&2
+            exit 1
+        fi
+        sleep 0.05
+    done
+}
+# $(( )) strips the padding macOS wc puts around its count
+n_usage_lines() {
+    if [ -f "$usage_file" ]; then echo $(($(wc -l < "$usage_file"))); else echo 0; fi
+}
+
+all_started() { [ $(($(find "$started" -type f | wc -l))) -ge "$nchildren" ]; }
+
+case "$mode" in
+    setsid)
+        # Outside duct's session, so duct must not record these. The process
+        # that calls setsid was forked inside duct's session and only then
+        # left it, so duct may have sampled it. It therefore does not record
+        # itself: it forks the child (the `; true` keeps sh from exec'ing it
+        # in place), and that child was never in duct's session.
+        for _ in $(seq 1 "$nchildren"); do
+            setsid sh -c '"$@"; true' _ "${child[@]}" &
+        done
+        wait_until all_started
+        # Two more reports, so duct has sampled while the children were alive
+        target=$(($(n_usage_lines) + 2))
+        reports_since_start() { [ "$(n_usage_lines)" -ge "$target" ]; }
+        wait_until reports_since_start
+        for f in "$started"/*; do
+            kill "$(basename "$f")" 2>/dev/null
+        done
+        exit 0
+        ;;
+    subshell|nohup|plain) ;;
+    *)
+        echo "Unknown mode: $mode" >&2
+        exit 1
+        ;;
+esac
 
 for _ in $(seq 1 "$nchildren"); do
     case "$mode" in
         subshell)
-            ( sleep "$sleeptime" & ) ;;
+            ( "${child[@]}" & ) ;;
         nohup)
-            ( nohup sleep "$sleeptime" & disown ) & ;;
-        setsid)
-            setsid sleep "$sleeptime" & ;;
+            ( nohup "${child[@]}" & disown ) & ;;
         plain)
-            sleep "$sleeptime" & ;;
-        *)
-            echo "Unknown mode: $mode" >&2
-            exit 1
-            ;;
+            "${child[@]}" & ;;
     esac
 done
 
-wait
+all_sampled() {
+    for f in "$started"/*; do
+        grep -q "\"$(basename "$f")\"" "$usage_file" 2>/dev/null || return 1
+    done
+}
+wait_until all_started
+wait_until all_sampled
diff --git a/test/data/until_sampled.sh b/test/data/until_sampled.sh
new file mode 100755
index 00000000..7d0275e0
--- /dev/null
+++ b/test/data/until_sampled.sh
@@ -0,0 +1,13 @@
+#!/bin/sh
+# Stay alive until duct has recorded this process in its usage file, so a
+# test can rely on at least one sample however slow the machine is.
+# Usage: until_sampled.sh USAGE_FILE
+usage_file=$1
+deadline=$(($(date +%s) + 60))
+until grep -q "\"$$\"" "$usage_file" 2>/dev/null; do
+    if [ "$(date +%s)" -ge "$deadline" ]; then
+        echo "until_sampled: pid $$ never appeared in $usage_file" >&2
+        exit 1
+    fi
+    sleep 0.05
+done
diff --git a/test/duct_main/test_e2e.py b/test/duct_main/test_e2e.py
index 62aefc39..daff4f0e 100644
--- a/test/duct_main/test_e2e.py
+++ b/test/duct_main/test_e2e.py
@@ -3,13 +3,17 @@
 import os
 from pathlib import Path
 import platform
+import shutil
+import signal
 import subprocess
 import time
 import pytest
+from utils import assert_files, rest_of_stderr, start_duct
 from con_duct._constants import SUFFIXES
 
 SYSTEM = platform.system()
 TEST_SCRIPT_DIR = Path(__file__).parent.parent / "data"
+LOG_FILES = [SUFFIXES[k] for k in ("stdout", "stderr", "info", "usage")]
 # Allow overriding the duct executable for testing external builds (e.g., PyInstaller)
 _DUCT_EXECUTABLES = [
     exe.strip()
@@ -28,48 +32,68 @@ def test_sanity(temp_output_dir: str, duct_cmd: str) -> None:
     subprocess.check_output(command, shell=True)
 
 
-@pytest.mark.flaky(reruns=3)
-@pytest.mark.parametrize("mode", ["plain", "subshell", "nohup", "setsid"])
+@pytest.mark.parametrize(
+    "mode",
+    [
+        "plain",
+        "subshell",
+        "nohup",
+        pytest.param(
+            "setsid",
+            marks=pytest.mark.skipif(
+                shutil.which("setsid") is None, reason="needs the setsid command"
+            ),
+        ),
+    ],
+)
 @pytest.mark.parametrize("num_children", [1, 2, 10])
 def test_spawn_children(
-    temp_output_dir: str, duct_cmd: str, mode: str, num_children: int
+    temp_output_dir: str,
+    tmp_path: Path,
+    duct_cmd: str,
+    mode: str,
+    num_children: int,
 ) -> None:
     duct_prefix = f"{temp_output_dir}log_"
+    usage_path = f"{duct_prefix}{SUFFIXES['usage']}"
+    started = tmp_path / "started"  # each child creates a file named by its pid
+    started.mkdir()
     script_path = TEST_SCRIPT_DIR / "spawn_children.sh"
-    dur = "0.3"
     command = (
-        f"{duct_cmd} -q --s-i 0.001 --r-i 0.01 "
-        f"-p {duct_prefix} {script_path} {mode} {num_children} {dur}"
+        f"{duct_cmd} -q --s-i 0.001 --r-i 0.01 -p {duct_prefix} "
+        f"{script_path} {mode} {num_children} {usage_path} {started}"
     )
+    # Fails (exit 1) if the children were not all sampled within 60 s
     subprocess.check_output(command, shell=True)
 
-    with open(f"{duct_prefix}{SUFFIXES['usage']}") as usage_file:
-        all_samples = [json.loads(line) for line in usage_file]
-
-    # Only count the child sleep processes
-    all_child_pids = set(
-        pid
-        for sample in all_samples
-        for pid, proc in sample["processes"].items()
-        if "sleep" in proc["cmd"]
-    )
-    # Add one pid for the hold-the-door process, see spawn_children.sh line 7
+    with open(usage_path) as usage_file:
+        sampled_pids = {
+            pid for line in usage_file for pid in json.loads(line)["processes"]
+        }
+    child_pids = {f.name for f in started.iterdir()}
+    assert len(child_pids) == num_children
     if mode == "setsid":
-        assert len(all_child_pids) == 1
+        # Started in their own session, which duct does not track
+        assert not child_pids & sampled_pids
     else:
-        assert len(all_child_pids) == num_children + 1
+        assert child_pids <= sampled_pids
 
 
 @pytest.mark.parametrize("session_mode", ["new-session", "current-session"])
 def test_session_modes(temp_output_dir: str, duct_cmd: str, session_mode: str) -> None:
     """Test that both session modes work correctly and collect appropriate data."""
     duct_prefix = f"{temp_output_dir}log_"
-    command = f"{duct_cmd} -q --s-i 0.01 --r-i 0.05 --mode {session_mode} -p {duct_prefix} sleep 0.3"
+    usage_file = Path(f"{duct_prefix}{SUFFIXES['usage']}")
+    info_file = Path(f"{duct_prefix}{SUFFIXES['info']}")
+    # The command exits once duct has sampled it (exit 1 if it never is)
+    script = TEST_SCRIPT_DIR / "until_sampled.sh"
+    command = (
+        f"{duct_cmd} -q --s-i 0.01 --r-i 0.05 --mode {session_mode} "
+        f"-p {duct_prefix} {script} {usage_file}"
+    )
     subprocess.check_output(command, shell=True)
 
     # Check that log files were created
-    usage_file = Path(f"{duct_prefix}{SUFFIXES['usage']}")
-    info_file = Path(f"{duct_prefix}{SUFFIXES['info']}")
 
     assert usage_file.exists(), f"Usage file not created for {session_mode} mode"
     assert info_file.exists(), f"Info file not created for {session_mode} mode"
@@ -93,23 +117,21 @@ def test_session_modes(temp_output_dir: str, duct_cmd: str, session_mode: str) -
 
     assert "execution_summary" in info_data
     assert info_data["execution_summary"]["exit_code"] == 0
-    assert "sleep" in info_data["command"]
+    assert "until_sampled.sh" in info_data["command"]
 
 
 def test_session_mode_behavior_difference(temp_output_dir: str, duct_cmd: str) -> None:
     """Test that new-session and current-session modes behave differently."""
 
-    # Start a unique background process in the current session
+    # Start a unique background process in the current session. It must outlive
+    # both duct runs however slow they are (the finally block stops it).
     background_process = subprocess.Popen(
-        ["python", "-c", "print('DUCT_TEST_MARKER'); import time; time.sleep(10)"],
+        ["python", "-c", "print('DUCT_TEST_MARKER'); import time; time.sleep(600)"],
         stdout=subprocess.DEVNULL,
         stderr=subprocess.DEVNULL,
     )
 
     try:
-        # Give background process time to start
-        time.sleep(0.1)
-
         new_session_prefix = f"{temp_output_dir}new_"
         current_session_prefix = f"{temp_output_dir}current_"
 
@@ -212,3 +234,64 @@ def test_logging_levels(temp_output_dir: str, duct_cmd: str) -> None:
     assert (
         result_none.stderr == ""
     ), f"Expected empty stderr, got: {result_none.stderr!r}"
+
+
+# The --fail-time values whose effect does not depend on how long the run
+# took: 0 always keeps a failed command's logs, a negative value never does
+FAIL_TIMES = [0, -1]
+
+
+@pytest.mark.parametrize("fail_time", FAIL_TIMES)
+def test_signal_int(temp_output_dir: str, duct_cmd: str, fail_time: int) -> None:
+    args = ["-p", temp_output_dir, f"--fail-time={fail_time}"]
+    proc = start_duct(duct_cmd, [*args, "sleep", "60"])
+    os.kill(proc.pid, signal.SIGINT)
+    proc.communicate()
+
+    # duct forwards SIGINT to the command and exits with the command's code
+    assert proc.returncode == 128 + signal.SIGINT
+
+    if fail_time < 0:
+        assert_files(temp_output_dir, LOG_FILES, exists=False)
+    else:
+        with open(os.path.join(temp_output_dir, SUFFIXES["info"])) as info:
+            info_data = json.loads(info.read())
+        assert info_data["execution_summary"]["exit_code"] == 128 + signal.SIGINT
+
+
+def _wait_for_lines(path: Path, n: int, proc: subprocess.Popen[str]) -> None:
+    """Wait until *path* has *n* lines, failing if duct exits first."""
+    while not path.exists() or len(path.read_text().splitlines()) < n:
+        assert proc.poll() is None, (
+            f"duct exited with {proc.returncode} before {path.name} had {n} lines; "
+            f"its stderr:\n{rest_of_stderr(proc)}"
+        )
+        time.sleep(0.01)
+
+
+@pytest.mark.parametrize("fail_time", FAIL_TIMES)
+def test_signal_kill(
+    temp_output_dir: str, tmp_path: Path, duct_cmd: str, fail_time: int
+) -> None:
+    progress = tmp_path / "signal_ignorer.progress"
+    args = ["-p", temp_output_dir, f"--fail-time={fail_time}"]
+    script = TEST_SCRIPT_DIR / "signal_ignorer.py"
+    proc = start_duct(duct_cmd, [*args, str(script), str(progress)])
+
+    _wait_for_lines(progress, 1, proc)  # "ready": the command ignores SIGINT now
+    # duct forwards the first two; each must arrive before the next is sent,
+    # since pending SIGINTs merge into one
+    for received in (2, 3):
+        os.kill(proc.pid, signal.SIGINT)
+        _wait_for_lines(progress, received, proc)
+    os.kill(proc.pid, signal.SIGINT)  # third: duct kills the command
+    proc.communicate()
+
+    assert proc.returncode == 128 + signal.SIGKILL
+
+    if fail_time < 0:
+        assert_files(temp_output_dir, LOG_FILES, exists=False)
+    else:
+        with open(os.path.join(temp_output_dir, SUFFIXES["info"])) as info:
+            info_data = json.loads(info.read())
+        assert info_data["execution_summary"]["exit_code"] == 128 + signal.SIGKILL
diff --git a/test/duct_main/test_execution.py b/test/duct_main/test_execution.py
index 72745c97..4273265b 100644
--- a/test/duct_main/test_execution.py
+++ b/test/duct_main/test_execution.py
@@ -1,13 +1,11 @@
 from __future__ import annotations
 import json
 import logging
-import multiprocessing
 import os
 from pathlib import Path
-import signal
 import subprocess
 import sys
-from time import sleep, time
+from time import time
 import pytest
 from utils import assert_files, run_duct_command
 from con_duct import _duct_main
@@ -75,18 +73,15 @@ def test_sanity_green(temp_output_dir: str) -> None:
     assert_expected_files(temp_output_dir)
 
 
-@pytest.mark.flaky(reruns=5)
-def test_execution_summary(
-    request: pytest.FixtureRequest, temp_output_dir: str
-) -> None:
-    # Scale sleep time on retries to handle slow CI runners (PyPy, Mac)
-    attempt = getattr(request.node, "execution_count", 1)
-    sleep_time = 0.1 * attempt
+def test_execution_summary(temp_output_dir: str) -> None:
+    # The command exits once duct has sampled it (exit 1 if it never is)
+    script = str(TEST_SCRIPT_DIR / "until_sampled.sh")
+    usage_file = os.path.join(temp_output_dir, SUFFIXES["usage"])
     assert (
         run_duct_command(
-            ["sleep", str(sleep_time)],
-            sample_interval=0.05,  # small enough to ensure we collect at least 1 sample
-            report_interval=sleep_time,
+            [script, usage_file],
+            sample_interval=0.05,
+            report_interval=0.1,
             output_prefix=temp_output_dir,
         )
         == 0
@@ -122,6 +117,28 @@ def test_sanity_red(
     assert_expected_files(temp_output_dir)
 
 
+@pytest.mark.parametrize(
+    "fail_time,command,logs_kept",
+    [
+        # Failed sooner than the threshold: logs are removed
+        (3, "exit 1", False),  # 3 is the default --fail-time
+        (60, "exit 1", False),
+        # Failed after running longer than the threshold: logs are kept
+        (1, "sleep 1.2; exit 1", True),
+    ],
+)
+def test_fail_time_threshold(
+    temp_output_dir: str, fail_time: int, command: str, logs_kept: bool
+) -> None:
+    assert (
+        run_duct_command(
+            ["sh", "-c", command], output_prefix=temp_output_dir, fail_time=fail_time
+        )
+        == 1
+    )
+    assert_expected_files(temp_output_dir, exists=logs_kept)
+
+
 def test_outputs_full(temp_output_dir: str) -> None:
     script_path = str(TEST_SCRIPT_DIR / "test_script.py")
     assert (
@@ -228,91 +245,6 @@ def test_execute_unknown_command(
     assert_expected_files(temp_output_dir, exists=False)
 
 
-def _runner_for_signal_int(temp_output_dir: str, fail_time: float | None) -> int:
-    kws = {}
-    if fail_time is not None:
-        kws["fail_time"] = fail_time
-    return run_duct_command(
-        ["sleep", "60.74016230000801"], output_prefix=temp_output_dir, **kws
-    )
-
-
-@pytest.mark.flaky(reruns=5)
-@pytest.mark.parametrize("fail_time", [None, 0, 10, -1, -3.14])
-def test_signal_int(
-    request: pytest.FixtureRequest, temp_output_dir: str, fail_time: float | None
-) -> None:
-    # Scale wait time on retries to handle slow CI runners (PyPy, Mac)
-    attempt = getattr(request.node, "execution_count", 1)
-    wait_time = 0.2 * attempt
-    proc = multiprocessing.Process(
-        target=_runner_for_signal_int, args=(temp_output_dir, fail_time)
-    )
-    proc.start()
-    sleep(wait_time)
-    assert proc.pid is not None, "Process PID should not be None"  # for mypy
-    os.kill(proc.pid, signal.SIGINT)
-    proc.join()
-
-    # Once the command has been killed, duct should exit gracefully with exit code 0
-    assert proc.exitcode == 0
-
-    if fail_time is None or fail_time != 0:
-        assert_expected_files(temp_output_dir, exists=False)
-    else:
-        # proc exit code should Cannot retrieve the exit code from the thread, it is written to the file
-        with open(os.path.join(temp_output_dir, SUFFIXES["info"])) as info:
-            info_data = json.loads(info.read())
-
-        command_exit_code = info_data["execution_summary"]["exit_code"]
-        # SIGINT
-        assert command_exit_code == 128 + 2
-
-
-def _runner_for_signal_kill(temp_output_dir: str, fail_time: float | None) -> int:
-    script_path = str(TEST_SCRIPT_DIR / "signal_ignorer.py")
-    kws = {}
-    if fail_time is not None:
-        kws["fail_time"] = fail_time
-    return run_duct_command([script_path], output_prefix=temp_output_dir, **kws)
-
-
-@pytest.mark.flaky(reruns=5)
-@pytest.mark.parametrize("fail_time", [None, 0, 10, -1, -3.14])
-def test_signal_kill(
-    request: pytest.FixtureRequest, temp_output_dir: str, fail_time: float | None
-) -> None:
-    # Scale wait time on retries to handle slow CI runners (PyPy, Mac)
-    attempt = getattr(request.node, "execution_count", 1)
-    wait_time = 0.2 * attempt
-    proc = multiprocessing.Process(
-        target=_runner_for_signal_kill, args=(temp_output_dir, fail_time)
-    )
-    proc.start()
-    sleep(wait_time)
-    assert proc.pid is not None, "Process PID should not be None"  # for mypy
-    os.kill(proc.pid, signal.SIGINT)
-    sleep(wait_time)
-    os.kill(proc.pid, signal.SIGINT)
-    sleep(wait_time)
-    os.kill(proc.pid, signal.SIGINT)
-    proc.join()
-
-    # Once the command has been killed, duct should exit gracefully with exit code 0
-    assert proc.exitcode == 0
-
-    if fail_time is None or fail_time != 0:
-        assert_expected_files(temp_output_dir, exists=False)
-    else:
-        # Cannot retrieve the command exit code from the thread, get from duct log
-        with open(os.path.join(temp_output_dir, SUFFIXES["info"])) as info:
-            info_data = json.loads(info.read())
-
-        command_exit_code = info_data["execution_summary"]["exit_code"]
-        # SIGKILL
-        assert command_exit_code == 128 + 9
-
-
 def test_duct_as_executable(temp_output_dir: str) -> None:
     ps_command = [
         sys.executable,
diff --git a/test/duct_main/test_signals.py b/test/duct_main/test_signals.py
new file mode 100644
index 00000000..bdc84a11
--- /dev/null
+++ b/test/duct_main/test_signals.py
@@ -0,0 +1,209 @@
+from __future__ import annotations
+from collections.abc import Generator
+import logging
+import os
+from pathlib import Path
+import signal
+import threading
+from typing import Any
+import pytest
+from utils import run_duct_command
+from con_duct import _signals
+from con_duct._signals import SigIntHandler, start_thread
+from con_duct._tracker import Report
+
+
+@pytest.fixture
+def sent(monkeypatch: pytest.MonkeyPatch) -> list[tuple[int, int]]:
+    """Record os.kill calls made by the handler instead of sending them."""
+    calls: list[tuple[int, int]] = []
+    monkeypatch.setattr(
+        "con_duct._signals.os.kill", lambda pid, sig: calls.append((pid, sig))
+    )
+    return calls
+
+
+@pytest.fixture
+def restore_sigint() -> Generator[None, None, None]:
+    """execute() installs its own SIGINT handler; put pytest's back."""
+    orig = signal.getsignal(signal.SIGINT)
+    yield
+    signal.signal(signal.SIGINT, orig)
+
+
+@pytest.fixture
+def stderr_locked(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Make logging fail as it does when the handler interrupts a stderr write."""
+
+    def reentrant(*_args: Any) -> None:
+        raise RuntimeError("reentrant call inside <_io.BufferedWriter name='<stderr>'>")
+
+    monkeypatch.setattr(_signals.lgr, "log", reentrant)
+
+
+@pytest.mark.parametrize(
+    "presses,expected",
+    [
+        (1, [signal.SIGINT]),
+        (2, [signal.SIGINT, signal.SIGINT]),
+        (3, [signal.SIGINT, signal.SIGINT, signal.SIGKILL]),
+    ],
+)
+def test_forwards_after_attach(
+    sent: list[tuple[int, int]], presses: int, expected: list[int]
+) -> None:
+    handler = SigIntHandler()
+    handler.attach(1234)
+    for _ in range(presses):
+        handler(signal.SIGINT, None)
+    assert sent == [(1234, sig) for sig in expected]
+
+
+def test_counts_but_does_not_forward_before_attach(
+    sent: list[tuple[int, int]],
+) -> None:
+    handler = SigIntHandler()
+    handler(signal.SIGINT, None)
+    assert handler.sigcount == 1
+    assert sent == []
+
+
+@pytest.mark.parametrize(
+    "presses,expected", [(1, signal.SIGINT), (2, signal.SIGINT), (3, signal.SIGKILL)]
+)
+def test_attach_acts_on_earlier_sigints(
+    sent: list[tuple[int, int]], presses: int, expected: int
+) -> None:
+    handler = SigIntHandler()
+    for _ in range(presses):
+        handler(signal.SIGINT, None)
+    handler.attach(1234)
+    # Only the latest escalation level is applied, once
+    assert sent == [(1234, expected)]
+
+
+@pytest.mark.usefixtures("restore_sigint")
+def test_sigint_during_attach_is_forwarded_once(sent: list[tuple[int, int]]) -> None:
+    class InterruptOnAttach(SigIntHandler):
+        def __setattr__(self, name: str, value: Any) -> None:
+            super().__setattr__(name, value)
+            if name == "pid" and value is not None:
+                # Deliver a real SIGINT immediately after publishing the PID.
+                signal.raise_signal(signal.SIGINT)
+
+    handler = InterruptOnAttach()
+    signal.signal(signal.SIGINT, handler)
+    handler.attach(1234)
+
+    assert handler.sigcount == 1
+    assert sent == [(1234, signal.SIGINT)]
+
+
+@pytest.mark.usefixtures("stderr_locked")
+def test_forwards_even_if_logging_fails(sent: list[tuple[int, int]]) -> None:
+    handler = SigIntHandler()
+    handler.attach(1234)
+    handler(signal.SIGINT, None)
+    assert sent == [(1234, signal.SIGINT)]
+
+
+@pytest.mark.usefixtures("stderr_locked")
+def test_log_falls_back_to_raw_stderr_write(
+    capfd: pytest.CaptureFixture[str], caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.INFO, logger="con-duct")
+    _signals._log(logging.INFO, "Received SIGINT, passing to command")
+    assert capfd.readouterr().err == (
+        "\ncon-duct [INFO] (written directly: the signal interrupted a log write): "
+        "Received SIGINT, passing to command\n"
+    )
+
+
+@pytest.mark.usefixtures("stderr_locked")
+def test_log_fallback_respects_log_level(
+    capfd: pytest.CaptureFixture[str], caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.WARNING, logger="con-duct")
+    _signals._log(logging.INFO, "Received SIGINT, passing to command")
+    assert capfd.readouterr().err == ""
+
+
+@pytest.mark.usefixtures("stderr_locked")
+def test_log_fallback_never_raises(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    def closed(_fd: int, _data: bytes) -> int:
+        raise OSError(9, "Bad file descriptor")
+
+    caplog.set_level(logging.INFO, logger="con-duct")
+    monkeypatch.setattr("con_duct._signals.os.write", closed)
+    _signals._log(logging.INFO, "Received SIGINT, passing to command")  # must not raise
+
+
+def test_command_already_exited(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    def no_such_process(_pid: int, _sig: int) -> None:
+        raise ProcessLookupError(3, "No such process")
+
+    monkeypatch.setattr("con_duct._signals.os.kill", no_such_process)
+    handler = SigIntHandler()
+    handler.attach(1234)
+    handler(signal.SIGINT, None)  # must not raise
+    assert [(r.levelname, r.message) for r in caplog.records] == [
+        ("WARNING", "Received SIGINT, but the command has already exited")
+    ]
+
+
+def _sigint_blocked() -> bool:
+    """Whether the calling thread blocks SIGINT."""
+    return signal.SIGINT in signal.pthread_sigmask(signal.SIG_BLOCK, [])
+
+
+def test_start_thread_blocks_sigint_only_in_the_new_thread() -> None:
+    blocked_in_thread: list[bool] = []
+    thread = threading.Thread(
+        target=lambda: blocked_in_thread.append(_sigint_blocked())
+    )
+    start_thread(thread)
+    thread.join()
+    assert blocked_in_thread == [True]
+    assert not _sigint_blocked()
+
+
+@pytest.mark.usefixtures("restore_sigint")
+def test_every_duct_thread_starts_with_sigint_blocked(
+    temp_output_dir: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # A thread inherits the mask its creator has at the moment it is started
+    blocked_at_start: list[bool] = []
+    real_start = threading.Thread.start
+
+    def recording_start(self: threading.Thread) -> None:
+        blocked_at_start.append(_sigint_blocked())
+        real_start(self)
+
+    monkeypatch.setattr(threading.Thread, "start", recording_start)
+    assert run_duct_command(["true"], output_prefix=temp_output_dir) == 0
+    assert blocked_at_start and all(blocked_at_start)
+
+
+@pytest.mark.usefixtures("restore_sigint")
+def test_sigint_before_command_starts_does_not_start_it(
+    temp_output_dir: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    real_report = Report
+
+    def report_then_sigint(*args: Any, **kwargs: Any) -> Any:
+        # Report is built after duct's handler is installed and before Popen
+        report = real_report(*args, **kwargs)
+        os.kill(os.getpid(), signal.SIGINT)
+        return report
+
+    monkeypatch.setattr("con_duct._duct_main.Report", report_then_sigint)
+    marker = tmp_path / "command-ran"
+    rc = run_duct_command(["touch", str(marker)], output_prefix=temp_output_dir)
+
+    assert rc == 128 + signal.SIGINT
+    assert not marker.exists()
+    assert list(Path(temp_output_dir).iterdir()) == []
diff --git a/test/utils.py b/test/utils.py
index d5ec9bef..ad928512 100644
--- a/test/utils.py
+++ b/test/utils.py
@@ -1,8 +1,16 @@
 from __future__ import annotations
 from io import BytesIO
 from pathlib import Path
+import re
+import shlex
+import subprocess
 from typing import Any
 
+# duct logs this right after attaching its SIGINT handler to the command
+# (_duct_main.execute).
+# Anchored on the log format so the wrapped command's echoed stderr can't match.
+DUCT_READY = re.compile(r"\] con-duct: duct \S+ is executing ")
+
 
 def run_duct_command(cli_args: list[str], **kwargs: Any) -> int:
     """Helper to run duct with test-friendly defaults.
@@ -40,6 +48,56 @@ def run_duct_command(cli_args: list[str], **kwargs: Any) -> int:
     return duct_execute(command=command, command_args=command_args, **defaults)  # type: ignore[arg-type]
 
 
+def start_duct(duct_cmd: str, args: list[str]) -> subprocess.Popen[str]:
+    """Start duct as a subprocess and return once it is ready for signals.
+
+    Reads duct's stderr until the log line emitted just after the SIGINT
+    handler is attached to the command. The caller should finish with ``communicate()``,
+    which drains the rest of stderr and waits for duct to exit.
+
+    That line is logged at INFO, so duct is started with ``--log-level=INFO``
+    whatever DUCT_LOG_LEVEL or a .env file says; ``args`` must not lower it.
+
+    Args:
+        duct_cmd: How to invoke duct, e.g. "duct" or "con-duct run"
+        args: Arguments for duct, including the command to run
+
+    Returns:
+        The running duct process, with its SIGINT handler installed
+    """
+    proc = subprocess.Popen(
+        [*shlex.split(duct_cmd), "--log-level=INFO", *args],
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    assert proc.stderr is not None  # for mypy
+    seen = []
+    for line in proc.stderr:
+        if DUCT_READY.search(line):
+            return proc
+        seen.append(line)
+    raise RuntimeError(
+        f"duct exited with {proc.wait()} before logging that it is executing; "
+        f"its stderr:\n{''.join(seen)}"
+    )
+
+
+def rest_of_stderr(proc: subprocess.Popen[str]) -> str:
+    """What duct wrote to stderr since ``start_duct`` returned, for failure messages.
+
+    Reads to end of file, so only call it once duct has exited.
+
+    Args:
+        proc: A duct process from ``start_duct``
+
+    Returns:
+        The unread remainder of duct's stderr
+    """
+    assert proc.stderr is not None  # for mypy
+    rest: str = proc.stderr.read()
+    return rest
+
+
 class MockStream:
     """Mocks stderr or stdout"""
 
diff --git a/tox.ini b/tox.ini
index e6710d69..a145033b 100644
--- a/tox.ini
+++ b/tox.ini
@@ -8,7 +8,7 @@ minversion = 3.3.0
 deps =
     pytest
     pytest-cov
-    pytest-rerunfailures
+    coverage>=7.10  # for [tool.coverage.run] patch in pyproject.toml
     -e .[all]
 commands =
     pytest {posargs} test
@@ -37,7 +37,7 @@ description = Test with minimal dependencies (no optional extras)
 deps =
     pytest
     pytest-cov
-    pytest-rerunfailures
+    coverage>=7.10  # for [tool.coverage.run] patch in pyproject.toml
     -e .
 commands =
     pytest {posargs} test
@@ -49,7 +49,7 @@ allowlist_externals = uv
 deps =
     pytest
     pytest-cov
-    pytest-rerunfailures
+    coverage>=7.10  # for [tool.coverage.run] patch in pyproject.toml
 commands =
     # numpy<2 required because matplotlib 3.5 was compiled against numpy 1.x
     # numpy>=1.21.3 is first with Python 3.10 wheels