diff --git a/tests/_deadline_capture.py b/tests/_deadline_capture.py
new file mode 100644
index 000000000..577a75e55
--- /dev/null
+++ b/tests/_deadline_capture.py
@@ -0,0 +1,143 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+# Copyright (C) 2026 MessageFoundry Foundation, LLC and contributors
+"""Fire the request deadline from inside the operation it is meant to cut off (vault BACKLOG #3871).
+
+A test that sets ``request_timeout_seconds`` to a fraction of a second races wall time against the
+request: the deadline counts from the start of the request, so on a loaded runner the auth
+dependencies, the body read or the file save alone could outlast it. The 503 then came before the
+route had started the operation, nothing was left running to outlive the request, and the test
+failed on a state the code under test never reached.
+
+:func:`capture_deadlines` replaces ``asyncio`` inside the deadline middleware's module only, so the
+:class:`DeadlineCapture` it returns keeps each deadline the middleware opens, and gives the app a
+deadline long enough never to fire on its own. The operation the test holds awaits
+:meth:`DeadlineCapture.cut_off`. That expires the middleware's own ``asyncio.Timeout`` on the loop's
+next pass, wherever the request is, so the 503 lands mid-operation and nowhere else, then holds the
+operation until :meth:`DeadlineCapture.held` lets it go.
+
+Engine PR 2438, open and not on ``main`` when this was written, carries a simpler copy of the
+stand-in in the web console's ``test_ui_dr_outlive.py``: it fires the latest deadline, with none of
+the checks below.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import types
+from collections.abc import AsyncIterator, Awaitable
+from typing import Any
+
+import httpx
+import pytest
+
+from messagefoundry.api import request_timeout
+from messagefoundry.api.outlive import OutlivingOperations
+
+__all__ = ["GENEROUS_DEADLINE_SECONDS", "DeadlineCapture", "capture_deadlines", "expect_cut_off"]
+
+#: Long enough that a deadline never fires on its own during a test: :meth:`DeadlineCapture.fire`
+#: fires it. Short enough that a test whose operation never fires it still fails on its own
+#: assertion inside the per-test timeout, after this, the drain below and its cancel grace.
+GENEROUS_DEADLINE_SECONDS = 15.0
+
+#: How long a test waits for the held operation to finish once it is let go.
+_DRAIN_SECONDS = 10.0
+
+
+class DeadlineCapture:
+    """Keeps each deadline the middleware opens, so a test can make the real one fire when it
+    chooses. :func:`capture_deadlines` builds and installs it."""
+
+    def __init__(self, outliving: OutlivingOperations) -> None:
+        self.deadlines: list[asyncio.Timeout] = []
+        self.fired = False
+        #: Why a fire went wrong. Recorded rather than raised: :meth:`fire` runs on the operation's
+        #: task, where the code under test may catch an error and turn it into its own outcome.
+        #: :func:`expect_cut_off` reports it on the test's task.
+        self.problem: str | None = None
+        self._outliving = outliving
+        self._released = asyncio.Event()
+
+    def timeout(self, delay: float | None) -> asyncio.Timeout:
+        deadline = asyncio.timeout(delay)
+        self.deadlines.append(deadline)
+        return deadline
+
+    def fire(self) -> None:
+        """Expire the one request deadline opened since this capture was installed, now.
+
+        Exactly one, still armed: a test installs the capture just before the request it cuts off,
+        so a second deadline means a request the test did not expect, and a disarmed one means the
+        response has already started, where firing would cut the body rather than the operation.
+        Anything else fires nothing: it is recorded, and the hold is let go so the request ends
+        and the test reports it, rather than every await waiting on the next."""
+        if len(self.deadlines) != 1:
+            self._refuse(f"expected one request deadline, saw {len(self.deadlines)}")
+            return
+        deadline = self.deadlines[0]
+        if deadline.when() is None:
+            self._refuse("the request had already started its response")
+            return
+        try:
+            deadline.reschedule(asyncio.get_running_loop().time())
+        except RuntimeError as exc:  # it had already expired, or its request had ended
+            self._refuse(f"the deadline could not be fired: {exc}")
+            return
+        self.fired = True
+
+    def _refuse(self, problem: str) -> None:
+        self.problem = problem
+        self._released.set()
+
+    async def cut_off(self) -> None:
+        """Awaited from inside the held operation: fire the deadline, then hold the operation until
+        :meth:`held` lets it go, so it cannot end before the 503 is in."""
+        self.fire()
+        await self._released.wait()
+
+    @contextlib.asynccontextmanager
+    async def held(self) -> AsyncIterator[None]:
+        """While the block runs, the operation stays held in :meth:`cut_off`. On the way out it is
+        let go and waited for, so it never runs on into the fixture's engine stop, and on a clean
+        exit it must have finished by itself rather than been cancelled by the drain. Whether it
+        succeeded is for the test's own assertions on its outcome rows."""
+        try:
+            yield
+        finally:
+            self._released.set()
+            cancelled = await self._outliving.drain(timeout=_DRAIN_SECONDS)
+        assert cancelled == [], f"never finished once let go: {cancelled}"
+
+
+def capture_deadlines(monkeypatch: pytest.MonkeyPatch, app: Any) -> DeadlineCapture:
+    """Install a :class:`DeadlineCapture` for ``app`` and give it a deadline that never fires on
+    its own; both are undone with ``monkeypatch``.
+
+    Call it after the sign-in and any proof, just before the request the test cuts off. The
+    middleware's module sees only ``timeout``, which is all it uses, so a new use fails loudly."""
+    capture = DeadlineCapture(app.state.outliving_operations)
+    monkeypatch.setattr(request_timeout, "asyncio", types.SimpleNamespace(timeout=capture.timeout))
+    monkeypatch.setattr(
+        app.state, request_timeout.TIMEOUT_STATE_ATTR, GENEROUS_DEADLINE_SECONDS, raising=False
+    )
+    return capture
+
+
+async def expect_cut_off(capture: DeadlineCapture, request: Awaitable[httpx.Response]) -> None:
+    """Await ``request`` and check the answer is the deadline middleware's own 503, fired by
+    ``capture`` from inside the operation: not a route's 503, and not the deadline running out.
+
+    The generous deadline bounds the wait: a deadline the operation never fires answers 503 there,
+    and the ``fired`` check fails. A fire that went wrong is reported first, even when the request
+    raised because of it."""
+    try:
+        r = await request
+    except Exception as exc:
+        if capture.problem is not None:
+            raise AssertionError(capture.problem) from exc
+        raise
+    assert capture.problem is None, capture.problem
+    assert r.status_code == 503, r.text
+    assert r.json() == {"detail": "the server timed out building a response"}
+    assert capture.fired, "the deadline was not fired from inside the operation"
diff --git a/tests/conftest.py b/tests/conftest.py
index d8934b109..47bebb84c 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -41,6 +41,8 @@
 # A helper module holding the content-free asserts several connector tests share. pytest rewrites
 # asserts only in test modules unless told, and an unrewritten assert vanishes under `python -O`.
 pytest.register_assert_rewrite("tests._content_free")
+# The same for the request-deadline helper the cut-off tests share (vault BACKLOG #3871).
+pytest.register_assert_rewrite("tests._deadline_capture")
 
 # ---------------------------------------------------------------------------------------------------
 # Per-PROCESS test slot.
diff --git a/tests/test_dr_outlive_deadline.py b/tests/test_dr_outlive_deadline.py
index e91ef43bc..9efe3e11f 100644
--- a/tests/test_dr_outlive_deadline.py
+++ b/tests/test_dr_outlive_deadline.py
@@ -11,9 +11,9 @@
 it (``api/outlive.py``), and the coordinator's own cancellation arms restore state and record an
 interrupted outcome for what can still cancel one, such as a shutdown.
 
-The HTTP tests patch the deadline down to a fraction of a second rather than wait 120 s; the
-mechanism under test is the cancellation, not the number. No test here measures how long a real
-activation takes on a restored store.
+The HTTP tests fire the middleware's own deadline from inside the operation rather than wait 120 s
+(``tests/_deadline_capture.py``, vault BACKLOG #3871); the mechanism under test is the cancellation,
+not the number. No test here measures how long a real activation takes on a restored store.
 """
 
 from __future__ import annotations
@@ -54,6 +54,7 @@
 from messagefoundry.store import MessageStore, Store
 from messagefoundry.store.crypto import generate_key, make_cipher
 from tests._admin_account import create_local_user_chosen
+from tests._deadline_capture import capture_deadlines, expect_cut_off
 
 PW = "Sup3rSecret!!DR-outlive"
 #: Synthetic, never real PHI (CLAUDE.md section 9).
@@ -168,8 +169,8 @@ async def seeded(tmp_path: Path) -> AsyncIterator[tuple[Engine, str]]:
 async def _admin_client(
     engine: Engine, deadline: float, *, provision: bool = True
 ) -> tuple[httpx.AsyncClient, Any]:
-    """A signed-in administrator client. The deadline is set AFTER the login, whose password hash
-    alone takes longer than the fractions of a second these tests use."""
+    """A signed-in administrator client, with ``deadline`` set AFTER the login. A test here that
+    cuts a request off then installs ``tests/_deadline_capture.py``, which replaces it."""
     service = AuthService(
         engine.store, AuthSettings(require_mfa=False, admin_write_min_interval_seconds=0)
     )
@@ -255,25 +256,29 @@ async def test_a_release_cut_off_by_the_deadline_still_hands_back_and_records_th
     depth = await _activate_directly(engine)
     rr = engine.registry_runner
     assert rr is not None
+    coord = _coordinator(engine)
 
-    # The drain gives up after 1.0 s; the request deadline fires at 0.3 s, well inside it -- the
-    # shape the 120 s pair had, where any non-draining row meant the deadline fired first.
-    monkeypatch.setattr(engine_module, "DR_RELEASE_DRAIN_TIMEOUT_SECONDS", 1.0)
-    client, app = await _admin_client(engine, deadline=0.3)
-    async with client:
-        r = await client.post("/dr/release")
-        assert r.status_code == 503
-        assert "timed out" in r.json()["detail"]
-
-        # The release goes on without its caller and completes the hand-back.
-        coord = _coordinator(engine)
-
-        async def released() -> bool:
-            return not coord.active and not app.state.outliving_operations.inflight
-
-        await _wait(released)
-
-    assert engine.dr_active is False
+    # The request deadline fires inside the drain, before the drain gives up -- the shape the
+    # 120 s pair had, where any non-draining row meant the deadline fired first. The drain is held
+    # until the 503 is in, then runs for real and gives up on the row that cannot drain.
+    monkeypatch.setattr(engine_module, "DR_RELEASE_DRAIN_TIMEOUT_SECONDS", 0.3)
+    real_drain = engine._drain_pipeline
+    client, app = await _admin_client(engine, deadline=30.0)
+    capture = capture_deadlines(monkeypatch, app)
+
+    async def drain_cut_off(**kwargs: Any) -> tuple[int, int]:
+        await capture.cut_off()
+        return await real_drain(**kwargs)
+
+    monkeypatch.setattr(engine, "_drain_pipeline", drain_cut_off)
+    outliving = app.state.outliving_operations
+    async with client, capture.held():
+        await expect_cut_off(capture, client.post("/dr/release"))
+        # Cut off in its drain: intake is already unbound, and the box is still active.
+        assert outliving.inflight == ["DR release"]
+        assert coord.active and not rr.inbound_running("file_in")
+    # Let go, the release went on without its caller and completed the hand-back.
+    assert not coord.active and engine.dr_active is False
     assert not rr.inbound_running("file_in")
     rows = await _rows(engine.store, "dr.release")
     assert len(rows) == 1
@@ -347,27 +352,25 @@ async def draining() -> bool:
 
 
 async def test_an_activation_cut_off_by_the_deadline_completes_and_a_retry_reads_its_outcome(
-    seeded: tuple[Engine, str],
+    seeded: tuple[Engine, str], monkeypatch: pytest.MonkeyPatch
 ) -> None:
     engine, _archive = seeded
     coord = _coordinator(engine)
     real_profile = coord._activate_profile
+    client, app = await _admin_client(engine, deadline=30.0)
+    capture = capture_deadlines(monkeypatch, app)
 
-    async def slow_profile() -> None:
-        await asyncio.sleep(0.6)  # outlasts the 0.2 s deadline below
+    async def profile_cut_off() -> None:
+        await capture.cut_off()  # the deadline fires in step 4
         await real_profile()
 
-    coord._activate_profile = slow_profile
-    client, app = await _admin_client(engine, deadline=0.2)
+    coord._activate_profile = profile_cut_off
+    outliving = app.state.outliving_operations
     async with client:
-        r = await client.post("/dr/activate", json={})
-        assert r.status_code == 503
-        assert "timed out" in r.json()["detail"]
-
-        async def settled() -> bool:
-            return not app.state.outliving_operations.inflight
-
-        await _wait(settled)
+        async with capture.held():
+            await expect_cut_off(capture, client.post("/dr/activate", json={}))
+            assert outliving.inflight == ["DR activation"]
+            assert await _rows(engine.store, "dr.activate") == []
         assert coord.active and engine.dr_active
         rows = await _rows(engine.store, "dr.activate")
         assert len(rows) == 1 and "recorded_late" not in rows[0]
diff --git a/tests/test_reload_cancellation.py b/tests/test_reload_cancellation.py
index 2a898dca5..52ef812e4 100644
--- a/tests/test_reload_cancellation.py
+++ b/tests/test_reload_cancellation.py
@@ -15,7 +15,7 @@
 
 import asyncio
 import json
-from collections.abc import AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterator
 from pathlib import Path
 from typing import Any
 
@@ -41,18 +41,11 @@
 from messagefoundry.store import MessageStore
 from tests._admin_account import create_local_user_chosen
 from tests._bound_proof import prove, reproving_transport
+from tests._deadline_capture import capture_deadlines, expect_cut_off
 
 PW = "Sup3rSecret!!reload-outlive"
 
 
-async def _wait(predicate: Callable[[], Awaitable[bool]], timeout: float = 10.0) -> None:
-    deadline = asyncio.get_running_loop().time() + timeout
-    while not await predicate():
-        if asyncio.get_running_loop().time() > deadline:
-            raise AssertionError("condition not met within timeout")
-        await asyncio.sleep(0.02)
-
-
 def _registry(tmp_path: Path, *, inbound: str = "file_in", directory: str = "in") -> Registry:
     inbox, outdir = tmp_path / directory, tmp_path / "out"
     inbox.mkdir(exist_ok=True)
@@ -172,7 +165,7 @@ async def engine(tmp_path: Path) -> AsyncIterator[Engine]:
     await eng.stop()
 
 
-async def _deployer(engine: Engine, deadline: float) -> tuple[httpx.AsyncClient, Any]:
+async def _deployer(engine: Engine) -> tuple[httpx.AsyncClient, Any]:
     # The shipped default: the action-bound proof is on, and reproving_transport re-proves for it
     # as a JSON client does (vault BACKLOG #3241). tests/test_bound_step_up_injection.py
     # pins the gate itself.
@@ -206,12 +199,10 @@ async def _deployer(engine: Engine, deadline: float) -> tuple[httpx.AsyncClient,
             "/auth/login", json={"username": "deployer", "password": PW, "provider": "local"}
         )
     assert r.status_code == 200, r.text
-    # The reload takes a proof bound to it (vault BACKLOG #2625). Mint it here, before the
-    # deadline: a re-proof hashes a password, which alone outlasts the deadlines these tests set.
+    # The reload takes a proof bound to it (vault BACKLOG #2625). Mint it here, up front: a
+    # re-proof is a second request, and the test cuts off the reload request alone.
     async with httpx.AsyncClient(transport=transport, base_url="http://t") as proving:
         proved = await prove(proving, r.json()["token"], PW, "config_reload")
-    # After the login: its password hash alone outlasts the deadline these tests set.
-    app.state.request_timeout_seconds = deadline
     client = httpx.AsyncClient(
         transport=transport,
         base_url="http://t",
@@ -228,22 +219,21 @@ async def test_a_reload_cut_off_by_the_deadline_finishes_and_writes_its_row(
     engine: Engine, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     registry = _registry(tmp_path)
+    client, app = await _deployer(engine)
+    # The middleware's own deadline, fired from inside the reload (tests/_deadline_capture.py).
+    capture = capture_deadlines(monkeypatch, app)
 
-    async def slow_reload(*_args: Any, **_kwargs: Any) -> ReloadOutcome:
-        await asyncio.sleep(0.6)  # outlasts the 0.2 s deadline below
+    async def reload_cut_off(*_args: Any, **_kwargs: Any) -> ReloadOutcome:
+        await capture.cut_off()
         return ReloadOutcome(registry=registry, applied=True, directory=tmp_path)
 
-    monkeypatch.setattr(engine, "reload_detail", slow_reload)
-    client, app = await _deployer(engine, deadline=0.2)
-    async with client:
-        r = await client.post("/config/reload", json={})
-    assert r.status_code == 503
-    assert "timed out" in r.json()["detail"]
-
-    async def settled() -> bool:
-        return not app.state.outliving_operations.inflight
-
-    await _wait(settled)
+    monkeypatch.setattr(engine, "reload_detail", reload_cut_off)
+    outliving = app.state.outliving_operations
+    async with client, capture.held():
+        await expect_cut_off(capture, client.post("/config/reload", json={}))
+        # Cut off while the reload was still running, and not cancelled with the request.
+        assert outliving.inflight == ["config reload"]
+        assert "config_reload" not in await _actions(engine.store)
     actions = await _actions(engine.store)
     assert "config_reload" in actions  # the reload's own row, written after its caller left
     assert "config_reload_interrupted" not in actions
diff --git a/tests/test_upload_api.py b/tests/test_upload_api.py
index c29478265..0dc0fa264 100644
--- a/tests/test_upload_api.py
+++ b/tests/test_upload_api.py
@@ -44,6 +44,7 @@
 from messagefoundry.uploads import UploadStore
 from tests._admin_account import create_local_user_chosen
 from tests._bound_proof import reproving_transport
+from tests._deadline_capture import DeadlineCapture, capture_deadlines, expect_cut_off
 from tests.test_uploads_strict_ciphertext import _BREAKS
 
 PW = "Correct-Horse-Battery-Staple-9"
@@ -992,7 +993,7 @@ async def test_the_prune_audit_row_names_the_system_not_the_pruned_files_owner(
 
 
 async def test_a_deadline_mid_sweep_still_writes_a_prune_row_per_deleted_file(
-    engine: Engine, tmp_path: Path
+    engine: Engine, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     """BACKLOG #2261, end to end. The request deadline cancels POST /uploads while its save-time
     sweep is parked mid-pass. The response is the deadline's 503, and every file the sweep removed
@@ -1031,6 +1032,7 @@ async def test_a_deadline_mid_sweep_still_writes_a_prune_row_per_deleted_file(
             )
 
         # Park the sweep's thread before its second aged pair until the pass is told to stop.
+        loop = asyncio.get_running_loop()
         aborts: list[threading.Event] = []
         parked = threading.Event()
         real_prune, real_paths = us.prune_expired, us._paths
@@ -1046,15 +1048,20 @@ def _paths(file_id: str) -> tuple[Path, Path]:
                 seen += 1
                 if seen == 2:
                     parked.set()
+                    loop.call_soon_threadsafe(capture.fire)
                     aborts[-1].wait(timeout=5)
             return real_paths(file_id)
 
         us.prune_expired = _prune
         us._paths = _paths
-        app.state.request_timeout_seconds = 2.0
-        r = await c.post("/uploads", files={"file": ("new.hl7", BATCH, "text/plain")}, headers=h)
+        # The middleware's own deadline, fired by the parked sweep (tests/_deadline_capture.py).
+        capture = capture_deadlines(monkeypatch, app)
+        upload = asyncio.ensure_future(
+            c.post("/uploads", files={"file": ("new.hl7", BATCH, "text/plain")}, headers=h)
+        )
+        await asyncio.wait([upload])
         assert parked.is_set(), "the sweep never reached its second aged pair"
-        assert r.status_code == 503, r.text
+        await expect_cut_off(capture, upload)
 
     deleted = aged - {m.file_id for m in await us.list_files()}
     rows = await engine.store.list_audit(action="upload.prune")
@@ -1434,28 +1441,22 @@ async def test_an_override_holder_deletes_an_upload_rotate_key_cannot_seal(
 # --- vault BACKLOG #3178: the upload.create and upload.delete rows outlive the request -------------
 
 
-def _park_audit(engine: Engine, action: str, monkeypatch: pytest.MonkeyPatch) -> asyncio.Event:
-    """Make ``record_audit`` for ``action`` wait until the returned event is set, so the request
-    deadline fires while that row is being written. Every other row writes through."""
-    release = asyncio.Event()
+def _park_audit(
+    engine: Engine, app: Any, action: str, monkeypatch: pytest.MonkeyPatch
+) -> DeadlineCapture:
+    """Fire the request deadline while ``record_audit`` writes ``action``, and hold that write until
+    the returned capture's ``held`` block ends (``tests/_deadline_capture.py``). Every other row
+    writes through."""
     real = engine.store.record_audit
+    capture = capture_deadlines(monkeypatch, app)
 
     async def _parked(name: str, **kwargs: Any) -> Any:
         if name == action:
-            await release.wait()
+            await capture.cut_off()
         return await real(name, **kwargs)
 
     monkeypatch.setattr(engine.store, "record_audit", _parked)
-    return release
-
-
-async def _settle(app: Any) -> None:
-    """Wait for the app's outliving operations to finish."""
-    for _ in range(200):
-        if not app.state.outliving_operations.inflight:
-            return
-        await asyncio.sleep(0.02)
-    raise AssertionError("an outliving operation never finished")
+    return capture
 
 
 async def test_a_deadline_during_the_create_row_still_writes_it(
@@ -1472,12 +1473,16 @@ async def test_a_deadline_during_the_create_row_still_writes_it(
         transport=reproving_transport(PW, app=app), base_url="http://t"
     ) as c:
         h = await _login(c, "op")
-        release = _park_audit(engine, "upload.create", monkeypatch)
-        app.state.request_timeout_seconds = 0.5
-        r = await c.post("/uploads", files={"file": ("a.hl7", BATCH, "text/plain")}, headers=h)
-        assert r.status_code == 503, r.text
-        release.set()
-        await _settle(app)
+        capture = _park_audit(engine, app, "upload.create", monkeypatch)
+        outliving = app.state.outliving_operations
+        async with capture.held():
+            await expect_cut_off(
+                capture,
+                c.post("/uploads", files={"file": ("a.hl7", BATCH, "text/plain")}, headers=h),
+            )
+            # Cut off inside the row's write, with the file already stored.
+            assert outliving.inflight == ["upload.create audit"]
+            assert len(await app.state.upload_store.list_files()) == 1
     [stored] = await app.state.upload_store.list_files()
     [row] = await engine.store.list_audit(action="upload.create")
     assert row["actor"] == "op" and json.loads(str(row["detail"]))["file_id"] == stored.file_id
@@ -1527,12 +1532,13 @@ async def test_a_deadline_during_the_delete_row_still_writes_it(
         h = await _login(c, "op")
         r = await c.post("/uploads", files={"file": ("a.hl7", BATCH, "text/plain")}, headers=h)
         fid = r.json()["file_id"]
-        release = _park_audit(engine, "upload.delete", monkeypatch)
-        app.state.request_timeout_seconds = 0.5
-        d = await c.delete(f"/uploads/{fid}", headers=h)
-        assert d.status_code == 503, d.text
-        release.set()
-        await _settle(app)
+        capture = _park_audit(engine, app, "upload.delete", monkeypatch)
+        outliving = app.state.outliving_operations
+        async with capture.held():
+            await expect_cut_off(capture, c.delete(f"/uploads/{fid}", headers=h))
+            # Cut off inside the row's write, with the file already removed.
+            assert outliving.inflight == ["upload delete"]
+            assert await app.state.upload_store.list_files() == []
     assert await app.state.upload_store.list_files() == []
     [row] = await engine.store.list_audit(action="upload.delete")
     assert row["actor"] == "op" and json.loads(str(row["detail"]))["file_id"] == fid