fix(ci): fit the rescue capture into the kill grace of a budget kill

distros.sh stops a distro at its budget with `timeout -k 60`: SIGTERM at
the deadline, SIGKILL 60 s later. dedicated.sh's cleanup trap then ran
the in-container capture under `timeout 1500` (266 s in Baserow#0 of run
40625679270) and the host capture under `timeout 900`, so the SIGKILL
always landed mid-capture, the snapshot was never copied out and the
rescue-diagnostics artifact was lost. The inner `timeout` also sat in its
own process group and outlived the kill, holding the step open.

- distros.sh keeps the grace in one place, KILL_GRACE_SECONDS, and
  exports INFINITO_CI_DISTRO_KILL_EPOCH next to the deadline epoch.
- scripts/tests/deploy/utils/rescue_window.sh decides per trap step:
  past the deadline, the step gets half of the time left until the
  SIGKILL (at least 1 s) and an end epoch; before it, the nominal
  timeout and no end epoch. Inside the trap `$?` is 0 on a budget kill,
  so the deadline epoch is the only reliable signal.
- dedicated.sh runs both captures through `timeout --foreground` with
  that window, keeping them in the trap's process group, and passes the
  end epoch as RESCUE_DEADLINE.
- utils/diagnostics/container.py caps every command at RESCUE_DEADLINE,
  writes `[skipped: past RESCUE_DEADLINE]` for what no longer fits and
  hands the deadline to nested captures. Without the variable nothing
  changes, so the swarm and workspace callers behave as before.

-k 60, the distro budget and every timeout-minutes are unchanged. Drill
tests replay the trap with the real helper and container.py: a budget
kill ends with rc 124 and the snapshot copied out, the old shape ends
with rc 137 and nothing copied, an early failure keeps the full capture.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

scripts/tests/deploy/ci/dedicated.sh

@@ -23,6 +23,8 @@ REPO_ROOT="$(cd "${SCRIPT_DIR}/../../../.." && pwd)"
 # shellcheck source=/dev/null
 source <(grep -E '^INFINITO_(PLAYWRIGHT_REPORTS_BASE_DIR|RESCUE_DIAGNOSTICS_DIR|RESCUE_DIAGNOSTICS_BASE|DNS53_SAMPLER_LOG)=' "${REPO_ROOT}/.env")
 : "${INFINITO_RESCUE_DIAGNOSTICS_BASE:?INFINITO_RESCUE_DIAGNOSTICS_BASE must be set}"
+# shellcheck source=scripts/tests/deploy/utils/rescue_window.sh
+. "${REPO_ROOT}/scripts/tests/deploy/utils/rescue_window.sh"
 
 apps=""
 
@@ -81,10 +83,13 @@ cleanup() {
 
 	local _rescue_host_dir="${INFINITO_RESCUE_DIAGNOSTICS_BASE}/${INFINITO_DISTRO}/${apps}"
 	mkdir -p "${_rescue_host_dir}"
+	local rescue_secs rescue_until
+	rescue_window 1500
 	echo ">>> Capturing rescue diagnostics inside ${INFINITO_CONTAINER} (recursive DiD snapshot) before teardown removes it"
-	timeout 1500 docker exec \
+	timeout --foreground "${rescue_secs}" docker exec \
 		-e "INFINITO_RESCUE_DIAGNOSTICS_DIR=${INFINITO_RESCUE_DIAGNOSTICS_DIR}" \
 		-e "INFINITO_DNS53_SAMPLER_LOG=${INFINITO_DNS53_SAMPLER_LOG}" \
+		-e "RESCUE_DEADLINE=${rescue_until}" \
 		"${INFINITO_CONTAINER}" \
 		python3 /opt/src/infinito/utils/diagnostics/container.py \
 		"${apps}" "compose post-deploy failure" 2>/dev/null || true # nocheck: shell-or-true -- grandfathered: worked in practice; TODO: sharpen to catch only the exact tolerated error
@@ -95,10 +100,12 @@ cleanup() {
 
 	local _did_id
 	_did_id="$(docker inspect --format '{{.Id}}' "${INFINITO_CONTAINER}" 2>/dev/null)" || _did_id=""
+	rescue_window 900
 	echo ">>> Capturing rescue diagnostics on the CI host itself into ${_rescue_host_dir}/ci-host"
 	INFINITO_RESCUE_DIAGNOSTICS_DIR="${_rescue_host_dir}/ci-host" \
 		RESCUE_SEEN="${_did_id}" \
-		timeout 900 python3 "${REPO_ROOT}/utils/diagnostics/container.py" \
+		RESCUE_DEADLINE="${rescue_until}" \
+		timeout --foreground "${rescue_secs}" python3 "${REPO_ROOT}/utils/diagnostics/container.py" \
 		"${apps}" "ci host at compose post-deploy failure" 2>/dev/null || true # nocheck: shell-or-true -- the collector exits 1 by contract so the caller cannot mistake a snapshot for a passing deploy
 
 	bash scripts/tests/deploy/utils/rescue_index.sh "${_rescue_host_dir}"

scripts/tests/deploy/distros.sh

@@ -12,6 +12,10 @@ set -euo pipefail
 #   INFINITO_CI_DISTRO_BUDGET_SECONDS   wall-clock budget for the whole run
 #   GITHUB_STEP_SUMMARY                 optional; the markdown table is appended there
 #
+# Exports:
+#   INFINITO_CI_DISTRO_DEADLINE_EPOCH   epoch the running distro gets the budget SIGTERM at
+#   INFINITO_CI_DISTRO_KILL_EPOCH       epoch the SIGKILL follows at, KILL_GRACE_SECONDS later
+#
 # Exports per iteration:
 #   INFINITO_DISTRO                     the distro under test
 
@@ -26,6 +30,7 @@ source "scripts/meta/env/load.sh"
 : "${INFINITO_CI_DISTRO_BUDGET_SECONDS:?INFINITO_CI_DISTRO_BUDGET_SECONDS is required (declared in default.env)}"
 
 BUDGET_HEADROOM_PERCENT=15
+KILL_GRACE_SECONDS=60
 
 if (($# == 0)); then
 	echo "[ERROR] a per-distro command is required" >&2
@@ -44,6 +49,7 @@ echo "=== Distro execution order: ${distro_arr[*]} ==="
 global_start="$(date +%s)"
 deadline="$((global_start + INFINITO_CI_DISTRO_BUDGET_SECONDS))"
 export INFINITO_CI_DISTRO_DEADLINE_EPOCH="${deadline}"
+export INFINITO_CI_DISTRO_KILL_EPOCH="$((deadline + KILL_GRACE_SECONDS))"
 echo "=== Global time budget: ${INFINITO_CI_DISTRO_BUDGET_SECONDS}s (deadline epoch=${deadline}) ==="
 
 max_seen=0
@@ -129,7 +135,7 @@ for i in "${!distro_arr[@]}"; do
 	distro_start="$(date +%s)"
 
 	set +e
-	timeout -k 60 "${remaining}s" "$@"
+	timeout -k "${KILL_GRACE_SECONDS}" "${remaining}s" "$@"
 	rc=$?
 	set -e
 

scripts/tests/deploy/utils/rescue_window.sh

@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+# Time window of one rescue-trap step, carved out of the distro budget's kill grace.
+#
+# Reads:
+#   INFINITO_CI_DISTRO_DEADLINE_EPOCH  epoch distros.sh sends the budget SIGTERM at
+#   INFINITO_CI_DISTRO_KILL_EPOCH      epoch distros.sh sends the SIGKILL at; unset outside distros.sh
+#
+# Sets in the caller:
+#   rescue_secs   timeout of the step
+#   rescue_until  epoch the step must end by; empty while the step keeps its nominal time
+
+# Param: $1 seconds the step takes when the trap was not entered by the budget SIGTERM
+rescue_window() {
+	local now
+	now="$(date +%s)"
+	rescue_secs="$1"
+	rescue_until=""
+	[ -n "${INFINITO_CI_DISTRO_KILL_EPOCH:-}" ] || return 0 # nocheck: distros.sh exports this epoch at run time; a static default would be a stale timestamp
+	((now >= INFINITO_CI_DISTRO_DEADLINE_EPOCH)) || return 0
+	rescue_secs="$(((INFINITO_CI_DISTRO_KILL_EPOCH - now + 1) / 2))"
+	((rescue_secs > 0)) || rescue_secs=1
+	rescue_until="$((now + rescue_secs))"
+	echo ">>> Budget kill: $((INFINITO_CI_DISTRO_KILL_EPOCH - now))s left until the SIGKILL, ${rescue_secs}s for this rescue step (until epoch ${rescue_until})"
+}

tests/unit/python/scripts/tests/deploy/test_distros.py

@@ -149,6 +149,20 @@ class TestDistrosLoop(unittest.TestCase):
         self.assertIn("❌ failed", self.summary)
         self.assertRegex(self.summary, r"rc=124 after \d+s, with \d+s of the 5s budget")
 
+    def test_the_kill_epoch_trails_the_deadline_by_the_kill_grace(self) -> None:
+        proc, record = self._run(
+            "debian",
+            'echo "$((INFINITO_CI_DISTRO_KILL_EPOCH - INFINITO_CI_DISTRO_DEADLINE_EPOCH))"'
+            ' >> "${RECORD}"',
+            INFINITO_CI_DISTRO_BUDGET_SECONDS="600",
+        )
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        loop = read_text(str(LOOP))
+        grace = re.search(r"^KILL_GRACE_SECONDS=(\d+)$", loop, re.MULTILINE)
+        self.assertIsNotNone(grace, loop)
+        self.assertIn('timeout -k "${KILL_GRACE_SECONDS}"', loop)
+        self.assertEqual(record.split(), [grace.group(1)])
+
     def test_the_job_summary_tables_every_distro_in_execution_order(self) -> None:
         proc, _record = self._run(
             DISTROS,

tests/unit/python/scripts/tests/deploy/utils/test_rescue_window.py

@@ -0,0 +1,206 @@
+"""Budget-kill drills for the rescue window of the compose trap."""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from utils.cache.files import PROJECT_ROOT, read_text
+
+DEPLOY = PROJECT_ROOT / "scripts" / "tests" / "deploy"
+HELPER = DEPLOY / "utils" / "rescue_window.sh"
+INDEX = DEPLOY / "utils" / "rescue_index.sh"
+DEDICATED = DEPLOY / "ci" / "dedicated.sh"
+COLLECTOR = PROJECT_ROOT / "utils" / "diagnostics" / "container.py"
+DISTRO_PREFIX = "INFINITO_CI_DISTRO_"  # nocheck: prefix the drills strip, not a key
+GRACE = 8
+SKIPPED = b"[skipped: past RESCUE_DEADLINE]\n"
+
+WINDOW = """\
+set -euo pipefail
+. "$1"
+now="$(date +%s)"
+if [ -n "$2" ]; then
+	export INFINITO_CI_DISTRO_DEADLINE_EPOCH="$((now + $2))"
+	export INFINITO_CI_DISTRO_KILL_EPOCH="$((now + $3))"
+fi
+rescue_window "$4"
+printf '%s %s %s\\n' "${now}" "${rescue_secs}" "${rescue_until:--}"
+"""
+
+DRIVER = """\
+#!/usr/bin/env bash
+set -euo pipefail
+now="$(date +%s)"
+deadline="$((now + BUDGET))"
+if [ "${EPOCHS}" = true ]; then
+	export INFINITO_CI_DISTRO_DEADLINE_EPOCH="${deadline}"
+	export INFINITO_CI_DISTRO_KILL_EPOCH="$((deadline + GRACE))"
+fi
+echo "kill_epoch=$((deadline + GRACE))"
+timeout -k "${GRACE}" "$((deadline - now))s" bash "${SANDBOX}/stand-in.sh" "$@"
+"""
+
+STAND_IN = """\
+#!/usr/bin/env bash
+set -euo pipefail
+. "${HELPER}"
+cleanup() {
+	local rescue_secs rescue_until
+	rescue_window 1500
+	INFINITO_RESCUE_DIAGNOSTICS_DIR="${SANDBOX}/in-container" \\
+		INFINITO_DNS53_SAMPLER_LOG="${SANDBOX}/dns53.log" \\
+		RESCUE_DEADLINE="${rescue_until}" \\
+		PATH="${SANDBOX}/bin" \\
+		"${TIMEOUT_BIN}" --foreground "${rescue_secs}" "${PYTHON_BIN}" "${COLLECTOR}" app drill >/dev/null 2>&1 || true
+	mkdir -p "${SANDBOX}/host"
+	tar -C "${SANDBOX}/in-container" -cf - . 2>/dev/null | tar -C "${SANDBOX}/host" -xf - 2>/dev/null || true
+	bash "${INDEX}" "${SANDBOX}/host"
+	date +%s >"${SANDBOX}/rescued"
+}
+trap cleanup EXIT
+"$@"
+"""
+
+STUB_JOURNAL = """\
+#!/bin/sh
+echo "RESCUE_DEADLINE=[${RESCUE_DEADLINE}]"
+exec @SLEEP@ "${JOURNAL_SECONDS}"
+"""
+
+
+def _env(**extra: str) -> dict[str, str]:
+    env = {k: v for k, v in os.environ.items() if not k.startswith(DISTRO_PREFIX)}
+    env.update(BASH_ENV="", **extra)
+    return env
+
+
+class RescueWindowTests(unittest.TestCase):
+    def _window(self, deadline: str, kill: str, nominal: int) -> tuple[int, int, str]:
+        proc = subprocess.run(
+            ["bash", "-c", WINDOW, "_", str(HELPER), deadline, kill, str(nominal)],
+            env=_env(),
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        now, secs, until = proc.stdout.splitlines()[-1].split()
+        return int(now), int(secs), until
+
+    def test_outside_distros_sh_a_step_keeps_its_nominal_time(self) -> None:
+        _now, secs, until = self._window("", "", 1500)
+        self.assertEqual((secs, until), (1500, "-"))
+
+    def test_before_the_deadline_a_step_keeps_its_nominal_time(self) -> None:
+        _now, secs, until = self._window("30", "90", 900)
+        self.assertEqual((secs, until), (900, "-"))
+
+    def test_past_the_deadline_a_step_gets_half_the_grace_left(self) -> None:
+        now, secs, until = self._window("-1", "59", 1500)
+        self.assertIn(secs, (29, 30))
+        self.assertLessEqual(abs(int(until) - (now + secs)), 1)
+
+    def test_a_kill_already_due_still_leaves_a_timeout_that_fires(self) -> None:
+        _now, secs, until = self._window("-10", "-1", 1500)
+        self.assertEqual(secs, 1)
+        self.assertNotEqual(until, "-")
+
+
+class DedicatedWiringTests(unittest.TestCase):
+    def test_every_capture_runs_inside_the_window(self) -> None:
+        text = read_text(str(DEDICATED))
+        self.assertIn("utils/rescue_window.sh", text)
+        self.assertIn("rescue_window 1500", text)
+        self.assertIn("rescue_window 900", text)
+        commands = text.replace("\\\n", " ").replace('"', "").splitlines()
+        captures = [c for c in commands if "utils/diagnostics/container.py" in c]
+        self.assertEqual(len(captures), 2, captures)
+        for capture in captures:
+            self.assertIn("timeout --foreground ${rescue_secs}", capture)
+            self.assertIn("RESCUE_DEADLINE=${rescue_until}", capture)
+
+
+class BudgetKillDrillTests(unittest.TestCase):
+    def _drill(
+        self, *deploy: str, budget: int, journal_seconds: str, epochs: bool
+    ) -> tuple[subprocess.CompletedProcess, int, int | None, dict[str, bytes]]:
+        with tempfile.TemporaryDirectory() as tmp:
+            sandbox = Path(tmp)
+            (sandbox / "bin").mkdir()
+            journal = sandbox / "bin" / "journalctl"
+            journal.write_text(STUB_JOURNAL.replace("@SLEEP@", shutil.which("sleep")))
+            journal.chmod(0o755)
+            (sandbox / "stand-in.sh").write_text(STAND_IN)
+            (sandbox / "driver.sh").write_text(DRIVER)
+            proc = subprocess.run(
+                ["bash", str(sandbox / "driver.sh"), *deploy],
+                env=_env(
+                    SANDBOX=str(sandbox),
+                    HELPER=str(HELPER),
+                    INDEX=str(INDEX),
+                    COLLECTOR=str(COLLECTOR),
+                    PYTHON_BIN=sys.executable,
+                    TIMEOUT_BIN=shutil.which("timeout"),
+                    BUDGET=str(budget),
+                    GRACE=str(GRACE),
+                    EPOCHS="true" if epochs else "false",
+                    JOURNAL_SECONDS=journal_seconds,
+                ),
+                capture_output=True,
+                text=True,
+                check=False,
+                timeout=120,
+            )
+            kill_epoch = int(proc.stdout.split("kill_epoch=", 1)[1].split()[0])
+            marker = sandbox / "rescued"
+            rescued = (
+                int(marker.read_text())  # nocheck: cache-read - tempdir fixture
+                if marker.is_file()
+                else None
+            )
+            files = {
+                path.name: path.read_bytes()  # nocheck: cache-read - tempdir fixture
+                for path in (sandbox / "host").glob("app-*/*")
+                if path.is_file()
+            }
+            return proc, kill_epoch, rescued, files
+
+    def test_a_budget_kill_copies_a_partial_snapshot_out_inside_the_grace(self) -> None:
+        proc, kill_epoch, rescued, files = self._drill(
+            "sleep", "1000", budget=1, journal_seconds="30", epochs=True
+        )
+        self.assertEqual(proc.returncode, 124, proc.stdout)
+        self.assertIn("Budget kill:", proc.stdout)
+        self.assertIsNotNone(rescued, proc.stdout)
+        self.assertLess(rescued, kill_epoch)
+        self.assertIn(b"application_id: app", files["meta.txt"])
+        self.assertIn(b"timed out", files["journal.txt"])
+        self.assertEqual(files["journal-errors.txt"], SKIPPED)
+        self.assertIn("Rescue diagnostics:", proc.stdout)
+
+    def test_without_the_window_the_kill_lands_before_the_copy(self) -> None:
+        proc, _kill_epoch, rescued, files = self._drill(
+            "sleep", "1000", budget=1, journal_seconds="30", epochs=False
+        )
+        self.assertEqual(proc.returncode, 137, proc.stdout)
+        self.assertIsNone(rescued)
+        self.assertEqual(files, {})
+
+    def test_a_failure_before_the_deadline_keeps_the_unbounded_capture(self) -> None:
+        proc, _kill_epoch, rescued, files = self._drill(
+            "sh", "-c", "exit 3", budget=60, journal_seconds="0.2", epochs=True
+        )
+        self.assertEqual(proc.returncode, 3, proc.stdout)
+        self.assertNotIn("Budget kill:", proc.stdout)
+        self.assertIsNotNone(rescued)
+        self.assertEqual(files["journal.txt"], b"RESCUE_DEADLINE=[]\n")
+        self.assertNotIn(SKIPPED, files.values())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/python/utils/diagnostics/test_container.py

@@ -10,6 +10,7 @@ import io
 import os
 import subprocess
 import tempfile
+import time
 import unittest
 import unittest.mock as mock
 from pathlib import Path
@@ -620,6 +621,86 @@ class RecurseTests(unittest.TestCase):
             self.assertTrue(any("rm" in c for c in calls))
 
 
+class DeadlineTests(unittest.TestCase):
+    def _run_with(self, deadline: str, **kwargs):
+        mod = _load()
+        with (
+            mock.patch.dict(os.environ, {mod._DEADLINE_ENV: deadline}),
+            mock.patch.object(mod.subprocess, "run", return_value=_cp([])) as spawn,
+        ):
+            result = mod.run(["probe"], **kwargs)
+        return mod, result, spawn
+
+    def test_without_a_deadline_the_nominal_timeout_stands(self):
+        _mod, _result, spawn = self._run_with("", timeout=600)
+        self.assertEqual(spawn.call_args.kwargs["timeout"], 600)
+
+    def test_a_command_is_capped_at_the_deadline(self):
+        _mod, _result, spawn = self._run_with(str(time.time() + 5), timeout=600)
+        self.assertGreater(spawn.call_args.kwargs["timeout"], 0)
+        self.assertLessEqual(spawn.call_args.kwargs["timeout"], 5)
+
+    def test_a_command_past_the_deadline_is_skipped_unrun(self):
+        mod, result, spawn = self._run_with(str(time.time() - 1))
+        spawn.assert_not_called()
+        self.assertEqual(result.returncode, 124)
+        self.assertEqual(result.stderr, mod._SKIPPED)
+
+    def test_a_running_command_is_cut_at_the_deadline(self):
+        mod = _load()
+        with mock.patch.dict(os.environ, {mod._DEADLINE_ENV: str(time.time() + 1)}):
+            started = time.monotonic()
+            result = mod.run(["sleep", "30"])
+        self.assertEqual(result.returncode, 124)
+        self.assertLess(time.monotonic() - started, 10)
+
+    def test_the_deadline_reaches_the_nested_capture(self):
+        mod = _load()
+        calls: list[list[str]] = []
+
+        def fake_run(cmd, **kw):
+            calls.append(cmd)
+            if cmd[-1] == "{{.Names}}":
+                return _cp(cmd, stdout=b"node1\n")
+            if "{{.Id}}" in cmd:
+                return _cp(cmd, stdout=b"d00d1234\n")
+            return _cp(cmd)
+
+        deadline = str(int(time.time()) + 3600)
+        with (
+            tempfile.TemporaryDirectory() as td,
+            mock.patch.dict(os.environ, {mod._DEADLINE_ENV: deadline}),
+            mock.patch.object(mod, "run", side_effect=fake_run),
+        ):
+            mod.recurse(Path(td), "docker", "app", "ctx", 0, [], "S")
+        nested_exec = next(c for c in calls if "python3" in c)
+        self.assertIn(f"{mod._DEADLINE_ENV}={deadline}", nested_exec)
+
+    def test_a_capture_past_the_deadline_still_leaves_a_snapshot(self):
+        mod = _load()
+        with tempfile.TemporaryDirectory() as td:
+            env = {
+                "INFINITO_RESCUE_DIAGNOSTICS_DIR": td,
+                mod._DEADLINE_ENV: str(time.time() - 1),
+            }
+            with (
+                mock.patch.dict(os.environ, env),
+                mock.patch.object(mod, "runtime_bin", return_value="docker"),
+                mock.patch.object(mod.subprocess, "run") as spawn,
+            ):
+                self.assertEqual(mod.main(["rescue.py", "app", "ctx"]), 1)
+            spawn.assert_not_called()
+            snapshot = next(Path(td).glob("app-*"))
+            meta = (
+                snapshot / "meta.txt"
+            ).read_text()  # nocheck: cache-read - tempdir fixture
+            journal = (
+                snapshot / "journal.txt"
+            ).read_bytes()  # nocheck: cache-read - tempdir fixture
+            self.assertIn("application_id: app", meta)
+            self.assertEqual(journal, mod._SKIPPED)
+
+
 class MainTests(unittest.TestCase):
     def test_main_requires_output_dir(self):
         mod = _load()

utils/diagnostics/README.md

@@ -7,6 +7,13 @@ Docker-in-Docker boundary, so one call collects evidence from the
 outermost runner down to the innermost application container. It always
 exits 1.
 
+An optional `RESCUE_DEADLINE` epoch caps every command on every level; once it
+has passed, each remaining capture is written as a
+`[skipped: past RESCUE_DEADLINE]` placeholder. The compose trap in
+[dedicated.sh](../../scripts/tests/deploy/ci/dedicated.sh) sets it through
+[rescue_window.sh](../../scripts/tests/deploy/utils/rescue_window.sh) only when
+the distro budget's SIGTERM has already arrived.
+
 ```mermaid
 flowchart TB
     subgraph entry["🩺 One workflow call on failure()"]

utils/diagnostics/container.py

@@ -9,7 +9,8 @@ container that carries python3 and a container runtime (DiD) and repeats the
 capture there, pulling each nested snapshot back under
 ``containers/<name>/nested/`` - from the outermost caller down to the deepest
 runtime, however deep the nesting goes. ``RESCUE_SEEN`` carries the ids already
-entered, so a runtime that lists itself is cut as a cycle.
+entered, so a runtime that lists itself is cut as a cycle. ``RESCUE_DEADLINE``
+(optional epoch, handed down every level) caps each command and skips it once past.
 
 Every collector is best-effort: a missing source must never abort the capture.
 ``INFINITO_RESCUE_DIAGNOSTICS_DIR`` and ``INFINITO_DNS53_SAMPLER_LOG`` are
@@ -29,6 +30,7 @@ import re
 import shutil
 import subprocess
 import sys
+import time
 from datetime import UTC, datetime
 from pathlib import Path
 
@@ -40,6 +42,8 @@ _NESTED_TIMEOUT = 600
 _TAR_TIMEOUT = 300
 _SELF_IN_CONTAINER = "/tmp/rescue-self.py"  # noqa: S108 - fixed staging path inside the inspected container
 _LOCAL_DUMPS_ENV = "INFINITO_RESCUE_LOCAL_DUMPS_DIR"
+_DEADLINE_ENV = "RESCUE_DEADLINE"
+_SKIPPED = b"[skipped: past RESCUE_DEADLINE]\n"
 _APP_LOG_TAIL = "find /var/log /var/www /opt /srv -xdev -type f -name '*.log' -size -20971520c -mmin -360 -exec sh -c 'echo \"===== $1\"; tail -n 400 \"$1\"' _ {} ';'"
 _PROBE_HOSTS = ("deb.debian.org", "ghcr.io", "repo.packagist.org")
 _DNSMASQ_DIRS = (
@@ -53,9 +57,13 @@ def runtime_bin() -> str | None:
 
 
 def run(
-    cmd: list[str], *, timeout: int = _EXEC_TIMEOUT, stdin: bytes | None = None
+    cmd: list[str], *, timeout: float = _EXEC_TIMEOUT, stdin: bytes | None = None
 ) -> subprocess.CompletedProcess:
     """Best-effort subprocess wrapper: never raises, captures everything."""
+    if os.environ.get(_DEADLINE_ENV):
+        timeout = min(timeout, float(os.environ[_DEADLINE_ENV]) - time.time())
+        if timeout <= 0:
+            return subprocess.CompletedProcess(cmd, 124, b"", _SKIPPED)
     try:
         return subprocess.run(
             cmd, input=stdin, capture_output=True, timeout=timeout, check=False
@@ -160,12 +168,7 @@ def collect_host(out: Path, app_id: str, context: str, stamp: str) -> None:
 
 
 def collect_dnsmasq_introspection(out: Path) -> None:
-    """fd table and rlimits of every live dnsmasq process.
-
-    Run 32118850138 caught dnsmasq alive with zero sockets and nothing in
-    the journal; the socket table alone cannot separate closed listeners
-    from an exhausted fd table or a lost netlink watch, the fd list can.
-    """
+    """fd table and rlimits of every live dnsmasq process."""
     for pid in " ".join(list_lines(["pidof", "dnsmasq"])).split():
         capture(out, f"dnsmasq-{pid}-fd.txt", ["ls", "-l", f"/proc/{pid}/fd"])
         capture(out, f"dnsmasq-{pid}-limits.txt", ["cat", f"/proc/{pid}/limits"])
@@ -191,13 +194,7 @@ def daemon_json_dns() -> list[str]:
 
 
 def collect_resolver_probes(out: Path) -> None:
-    """Resolve each probed name, once through the stack and once per nameserver.
-
-    ``getent`` honours nsswitch and cannot address a server, so on its own it
-    cannot say which of the resolvers in resolv.conf refused. Where the distro
-    ships a query tool, ask every nameserver directly; where it does not, the
-    per-name verdicts still stand on their own.
-    """
+    """Resolve each probed name, once through the stack and once per nameserver."""
     tool = next(
         (
             argv
@@ -428,6 +425,8 @@ def recurse(
                 "-e",
                 "RESCUE_SEEN=" + ",".join([*seen, cid] if cid else seen),
                 "-e",
+                f"{_DEADLINE_ENV}={os.environ.get(_DEADLINE_ENV, '')}",
+                "-e",
                 f"{_LOCAL_DUMPS_ENV}={os.environ.get(_LOCAL_DUMPS_ENV, '')}",
                 "-e",
                 f"INFINITO_DNS53_SAMPLER_LOG={os.environ['INFINITO_DNS53_SAMPLER_LOG']}",