From ee4c5f5242ec2fb34e43521df6201d36c376e6f7 Mon Sep 17 00:00:00 2001
From: FetchDoge <297783350+FetchDoge@users.noreply.github.com>
Date: Thu, 22 Oct 2026 21:49:18 -0600
Subject: [PATCH 1/4] fix(gate): re-prove authority at the commit point; halts
 drain in-flight effects

---
 src/interlock/__init__.py |  76 ++++++++++++++++++-----
 src/interlock/estop.py    |  60 ++++++++++++++++--
 src/interlock/leases.py   |  16 ++++-
 tests/test_inflight.py    | 127 ++++++++++++++++++++++++++++++++++++++
 4 files changed, 257 insertions(+), 22 deletions(-)
 create mode 100644 tests/test_inflight.py

diff --git a/src/interlock/__init__.py b/src/interlock/__init__.py
index 9e8e26f..09fec26 100644
--- a/src/interlock/__init__.py
+++ b/src/interlock/__init__.py
@@ -13,6 +13,7 @@
 
 from collections.abc import Callable, Iterator
 from contextlib import contextmanager
+from typing import TypeVar
 
 from interlock._errors import (
     EStopEngaged,
@@ -23,12 +24,14 @@
     LeaseRevoked,
 )
 from interlock.claims import Claim
-from interlock.estop import EStop
+from interlock.estop import EStop, InFlightGate
 from interlock.faults import FaultBus
 from interlock.interfaces import REGISTRY, InterfaceRecord, declare
 from interlock.leases import Lease, LeaseManager
 from interlock.ledger import Ledger
 
+T = TypeVar("T")
+
 __version__ = "0.1.0"
 
 __all__ = [
@@ -60,8 +63,9 @@ def __init__(
         clock: Callable[[], float] | None = None,
         ledger: Ledger | None = None,
     ) -> None:
-        self.leases = LeaseManager(clock)
-        self.estop = EStop()
+        self._gate = InFlightGate()
+        self.leases = LeaseManager(clock, gate=self._gate)
+        self.estop = EStop(self._gate)
         self.faults = FaultBus()
         self.ledger = ledger if ledger is not None else Ledger()
 
@@ -92,24 +96,15 @@ def grant(
         )
         return lease
 
-    @contextmanager
-    def permit(self, lease: Lease, action: str, target: str) -> Iterator[Lease]:
-        """Permit exactly one action on one target, checking in order:
-
-        1. e-stop clear → else :exc:`EStopEngaged`
-        2. lease issued by this facade's manager and unaltered → else
-           :exc:`LeaseNotIssued`; then live (unrevoked, unexpired) → else
-           :exc:`LeaseRevoked` or :exc:`LeaseExpired`
-        3. no gating fault on the lease's resource → else :exc:`FaultActive`
-
-        Permits and denials are appended to the ledger.
-        """
+    def _check(self, lease: Lease, action: str, target: str, phase: str) -> dict:
+        """Run the gate in order; log and raise on the first failure."""
         base = {
             "lease_id": lease.id,
             "holder": lease.holder,
             "resource": lease.resource,
             "action": action,
             "target": target,
+            "phase": phase,
         }
         try:
             self.estop.assert_clear()
@@ -130,8 +125,57 @@ def permit(self, lease: Lease, action: str, target: str) -> Iterator[Lease]:
         if self.faults.gating(lease.resource):
             self.ledger.append("denials", {**base, "reason": "fault"})
             raise FaultActive(f"gating fault active on {lease.resource}")
+        return base
+
+    @contextmanager
+    def permit(self, lease: Lease, action: str, target: str) -> Iterator[Lease]:
+        """Permit exactly one action on one target, checking in order:
+
+        1. e-stop clear → else :exc:`EStopEngaged`
+        2. lease issued by this facade's manager and unaltered → else
+           :exc:`LeaseNotIssued`; then live (unrevoked, unexpired) → else
+           :exc:`LeaseRevoked` or :exc:`LeaseExpired`
+        3. no gating fault on the lease's resource → else :exc:`FaultActive`
+
+        Permits and denials are appended to the ledger.
+
+        ``permit()`` checks at block entry only: it cannot stop an effect the
+        caller performs later inside the block. For irreversible effects use
+        :meth:`execute`, which re-proves the gate at the commit point. If the
+        e-stop engages or the lease is revoked while the block runs, a
+        ``violations`` record is appended on exit so the gap is never silent.
+        """
+        base = self._check(lease, action, target, phase="entry")
         self.ledger.append("permits", base)
-        yield lease
+        try:
+            yield lease
+        finally:
+            reason = None
+            if self.estop.is_engaged:
+                reason = "estop"
+            elif self.leases.is_revoked(lease):
+                reason = "lease_revoked"
+            if reason:
+                self.ledger.append("violations", {**base, "reason": reason})
+
+    def execute(self, lease: Lease, action: str, target: str, effect: Callable[[], T]) -> T:
+        """Run ``effect`` only if the gate is clear at the commit point.
+
+        Fail-closed: the gate (e-stop, lease issued/live/unrevoked, faults) is
+        checked and the effect registered as in flight atomically, under the
+        same lock :meth:`EStop.engage` and :meth:`LeaseManager.revoke` take.
+        Either the halt lands first and the effect never starts, or the effect
+        started first and the halt call waits for it to finish. Any error in
+        the check refuses the effect.
+        """
+        with self._gate.cond:
+            base = self._check(lease, action, target, phase="commit")
+            self._gate.enter(lease.id)
+        try:
+            self.ledger.append("permits", base)
+            return effect()
+        finally:
+            self._gate.leave(lease.id)
 
     def claim(
         self,
diff --git a/src/interlock/estop.py b/src/interlock/estop.py
index 667640e..419f47a 100644
--- a/src/interlock/estop.py
+++ b/src/interlock/estop.py
@@ -4,20 +4,31 @@
 Resuming requires explicit, deliberate release — never an automatic timeout.
 """
 
+import threading
+
 from interlock._errors import EStopEngaged
 
 
 class EStop:
     """A latching global halt. Engagement is sticky; release is deliberate."""
 
-    def __init__(self) -> None:
+    def __init__(self, gate: "InFlightGate | None" = None) -> None:
         self._engaged = False
         self._reason = ""
+        self._gate = gate if gate is not None else InFlightGate()
 
     def engage(self, reason: str) -> None:
-        """Engage the e-stop. All permits are refused until :meth:`release`."""
-        self._engaged = True
-        self._reason = reason
+        """Engage the e-stop. All permits are refused until :meth:`release`.
+
+        The flag is set under the in-flight gate, so no commit-time check can
+        interleave with it, and the call returns only once every effect that
+        was already committed (via ``Interlock.execute``) has finished. When
+        ``engage`` returns, nothing is mid-effect and nothing new can start.
+        """
+        with self._gate.cond:
+            self._engaged = True
+            self._reason = reason
+        self._gate.drain()
 
     def release(self) -> None:
         """Release the e-stop. An operator decision, never a timeout."""
@@ -38,3 +49,44 @@ def assert_clear(self) -> None:
         """Raise :exc:`EStopEngaged` unless the e-stop is clear."""
         if self._engaged:
             raise EStopEngaged(f"e-stop engaged: {self._reason}")
+
+
+class InFlightGate:
+    """Shared condition between the commit-time check and the halt paths.
+
+    ``Interlock.execute`` checks authority and registers its effect as
+    in flight under ``cond``; ``EStop.engage`` and ``LeaseManager.revoke``
+    change state under ``cond`` and then wait for in-flight effects to drain.
+    Effects run concurrently; only check-and-register is serialised. A halt
+    raised from inside an effect does not wait on its own thread's effect.
+    """
+
+    def __init__(self) -> None:
+        self.cond = threading.Condition()
+        self._in_flight: dict[tuple[str, int], int] = {}
+
+    def enter(self, lease_id: str) -> None:
+        """Register one effect for ``lease_id``. Call with ``cond`` held."""
+        key = (lease_id, threading.get_ident())
+        self._in_flight[key] = self._in_flight.get(key, 0) + 1
+
+    def leave(self, lease_id: str) -> None:
+        key = (lease_id, threading.get_ident())
+        with self.cond:
+            n = self._in_flight.get(key, 0) - 1
+            if n <= 0:
+                self._in_flight.pop(key, None)
+            else:
+                self._in_flight[key] = n
+            self.cond.notify_all()
+
+    def _busy(self, lease_id: str | None) -> bool:
+        me = threading.get_ident()
+        return any(
+            tid != me and (lease_id is None or lid == lease_id) for lid, tid in self._in_flight
+        )
+
+    def drain(self, lease_id: str | None = None) -> None:
+        """Block until no other thread has an effect (for ``lease_id``) in flight."""
+        with self.cond:
+            self.cond.wait_for(lambda: not self._busy(lease_id))
diff --git a/src/interlock/leases.py b/src/interlock/leases.py
index 3504a0e..9b8d443 100644
--- a/src/interlock/leases.py
+++ b/src/interlock/leases.py
@@ -49,9 +49,10 @@ class _Issued:
 class LeaseManager:
     """Issues, renews, revokes, and checks leases against an injectable clock."""
 
-    def __init__(self, clock: Callable[[], float] | None = None) -> None:
+    def __init__(self, clock: Callable[[], float] | None = None, gate=None) -> None:
         self._clock = clock if clock is not None else time.monotonic
         self._issued: dict[str, _Issued] = {}
+        self._gate = gate
 
     def now(self) -> float:
         """Current time according to the manager's clock."""
@@ -125,8 +126,19 @@ def revoke(self, lease: Lease) -> None:
         """Revoke a lease immediately. Revocation is permanent.
 
         The manager records the revocation, so resetting ``lease.revoked`` on
-        the object does not restore authority.
+        the object does not restore authority. With an in-flight gate (as
+        wired by ``Interlock``), the revocation is ordered against commit-time
+        checks and the call returns only after this lease's committed effects
+        have finished.
         """
+        if self._gate is None:
+            self._mark_revoked(lease)
+            return
+        with self._gate.cond:
+            self._mark_revoked(lease)
+        self._gate.drain(lease.id)
+
+    def _mark_revoked(self, lease: Lease) -> None:
         lease.revoked = True
         rec = self._record(lease)
         if rec is not None:
diff --git a/tests/test_inflight.py b/tests/test_inflight.py
new file mode 100644
index 0000000..a46f355
--- /dev/null
+++ b/tests/test_inflight.py
@@ -0,0 +1,127 @@
+"""In-flight e-stop and revocation: the gauntlet `revoke-during-spend` case.
+
+Gauntlet (revoke-during-spend): the gate is checked, the e-stop engages
+100 ms later while the action is still preparing (300 ms), and the effect then
+ran anyway, because `permit()` only checks at entry. These tests pin the
+fail-closed contract of `Interlock.execute()`: the gate is re-proven at the
+commit point, atomically with starting the effect, and an e-stop or revoke
+returns only once no effect is mid-flight.
+"""
+
+import threading
+import time
+
+import pytest
+
+from interlock import EStopEngaged, Interlock, LeaseRevoked
+
+
+def _series(ilk, name):
+    return list(ilk.ledger.read_series(name))
+
+
+def _later(delay_s, fn):
+    t = threading.Timer(delay_s, fn)
+    t.start()
+    return t
+
+
+def test_estop_during_slow_preparation_refuses_effect():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    calls = []
+    t = _later(0.1, lambda: ilk.estop.engage("gauntlet"))
+    time.sleep(0.3)  # slow rail: authority was fine at the start
+    with pytest.raises(EStopEngaged):
+        ilk.execute(lease, "pay", "merchant", lambda: calls.append("paid"))
+    t.join()
+    assert calls == []
+    denial = _series(ilk, "denials")[-1]
+    assert denial["reason"] == "estop" and denial["phase"] == "commit"
+
+
+def test_revoke_during_slow_preparation_refuses_effect():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    calls = []
+    t = _later(0.1, lambda: ilk.leases.revoke(lease))
+    time.sleep(0.3)
+    with pytest.raises(LeaseRevoked):
+        ilk.execute(lease, "pay", "merchant", lambda: calls.append("paid"))
+    t.join()
+    assert calls == []
+
+
+def test_execute_runs_effect_and_returns_its_value_when_clear():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    assert ilk.execute(lease, "pay", "merchant", lambda: 42) == 42
+    assert _series(ilk, "permits")[-1]["phase"] == "commit"
+
+
+def test_estop_returns_only_after_running_effect_finishes():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    order = []
+    started = threading.Event()
+
+    def effect():
+        started.set()
+        time.sleep(0.2)
+        order.append("effect_done")
+
+    worker = threading.Thread(target=lambda: ilk.execute(lease, "pay", "m", effect))
+    worker.start()
+    started.wait(1)
+    ilk.estop.engage("halt")
+    order.append("estop_returned")
+    worker.join()
+    assert order == ["effect_done", "estop_returned"]
+
+
+def test_race_no_effect_starts_after_estop_returns():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    lock = threading.Lock()
+    starts = []
+    stop_returned_at = []
+
+    def effect():
+        with lock:
+            starts.append(time.monotonic())
+        time.sleep(0.001)
+
+    def spender():
+        for _ in range(200):
+            try:
+                ilk.execute(lease, "pay", "m", effect)
+            except EStopEngaged:
+                return
+
+    threads = [threading.Thread(target=spender) for _ in range(16)]
+    for t in threads:
+        t.start()
+    time.sleep(0.05)
+    ilk.estop.engage("race")
+    stop_returned_at.append(time.monotonic())
+    for t in threads:
+        t.join()
+    assert starts, "race never exercised"
+    assert all(s < stop_returned_at[0] for s in starts)
+
+
+def test_permit_block_records_violation_when_authority_lost_inside():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    with ilk.permit(lease, "pay", "m"):
+        ilk.estop.engage("mid-block")
+    v = _series(ilk, "violations")[-1]
+    assert v["reason"] == "estop" and v["lease_id"] == lease.id
+
+
+def test_halt_from_inside_an_effect_does_not_deadlock():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    ilk.execute(lease, "pay", "m", lambda: ilk.estop.engage("effect saw fraud"))
+    with pytest.raises(EStopEngaged):
+        ilk.execute(lease, "pay", "m", lambda: None)

From 72e0bc17a05cbc57fce531f8a9dcd1b36a486cb0 Mon Sep 17 00:00:00 2001
From: FetchDoge <297783350+FetchDoge@users.noreply.github.com>
Date: Fri, 23 Oct 2026 18:25:00 +0000
Subject: [PATCH 2/4] fix(gate): order fault reports, concurrent halts, and
 ledger writes against the commit point

---
 src/interlock/__init__.py | 51 ++++++++++++++++++++++++++++-----
 src/interlock/estop.py    | 42 ++++++++++++++++++++++++---
 src/interlock/faults.py   | 47 +++++++++++++++++++++---------
 src/interlock/ledger.py   | 60 +++++++++++++++++++++++----------------
 4 files changed, 150 insertions(+), 50 deletions(-)

diff --git a/src/interlock/__init__.py b/src/interlock/__init__.py
index 09fec26..85680ca 100644
--- a/src/interlock/__init__.py
+++ b/src/interlock/__init__.py
@@ -11,6 +11,7 @@
 facade and live, no gating fault on the lease's resource.
 """
 
+import inspect
 from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from typing import TypeVar
@@ -66,7 +67,7 @@ def __init__(
         self._gate = InFlightGate()
         self.leases = LeaseManager(clock, gate=self._gate)
         self.estop = EStop(self._gate)
-        self.faults = FaultBus()
+        self.faults = FaultBus(gate=self._gate)
         self.ledger = ledger if ledger is not None else Ledger()
 
     def grant(
@@ -144,14 +145,18 @@ def permit(self, lease: Lease, action: str, target: str) -> Iterator[Lease]:
         :meth:`execute`, which re-proves the gate at the commit point. If the
         e-stop engages or the lease is revoked while the block runs, a
         ``violations`` record is appended on exit so the gap is never silent.
+        An engagement counts even if the e-stop was released again before
+        the block exited — the exit check compares engagement counts, not
+        the final state.
         """
         base = self._check(lease, action, target, phase="entry")
         self.ledger.append("permits", base)
+        engagements = self.estop.engagements
         try:
             yield lease
         finally:
             reason = None
-            if self.estop.is_engaged:
+            if self.estop.engagements != engagements:
                 reason = "estop"
             elif self.leases.is_revoked(lease):
                 reason = "lease_revoked"
@@ -167,15 +172,47 @@ def execute(self, lease: Lease, action: str, target: str, effect: Callable[[], T
         Either the halt lands first and the effect never starts, or the effect
         started first and the halt call waits for it to finish. Any error in
         the check refuses the effect.
+
+        ``effect`` must run to completion synchronously. A coroutine,
+        generator, or async generator is deferred work: calling it starts
+        nothing, so the gate cannot hold it in flight. Deferred callables
+        are refused with :exc:`TypeError` before the gate runs, and a
+        deferred result returned by a plain callable is closed unstarted
+        and refused the same way — never left to run after a halt returns.
         """
-        with self._gate.cond:
-            base = self._check(lease, action, target, phase="commit")
-            self._gate.enter(lease.id)
+        if (
+            inspect.iscoroutinefunction(effect)
+            or inspect.isgeneratorfunction(effect)
+            or inspect.isasyncgenfunction(effect)
+        ):
+            raise TypeError(
+                "execute() effect must be a plain synchronous callable; "
+                "async and generator effects are refused"
+            )
+        registered = False
         try:
+            with self._gate.cond:
+                base = self._check(lease, action, target, phase="commit")
+                self._gate.enter(lease.id)
+                registered = True
             self.ledger.append("permits", base)
-            return effect()
+            result = effect()
+            if (
+                inspect.iscoroutine(result)
+                or inspect.isgenerator(result)
+                or inspect.isasyncgen(result)
+            ):
+                close = getattr(result, "close", None)
+                if close is not None:
+                    close()
+                raise TypeError(
+                    "execute() effect returned deferred work (coroutine or generator); "
+                    "it was closed unstarted and refused"
+                )
+            return result
         finally:
-            self._gate.leave(lease.id)
+            if registered:
+                self._gate.leave(lease.id)
 
     def claim(
         self,
diff --git a/src/interlock/estop.py b/src/interlock/estop.py
index 419f47a..44935f7 100644
--- a/src/interlock/estop.py
+++ b/src/interlock/estop.py
@@ -15,6 +15,7 @@ class EStop:
     def __init__(self, gate: "InFlightGate | None" = None) -> None:
         self._engaged = False
         self._reason = ""
+        self._engagements = 0
         self._gate = gate if gate is not None else InFlightGate()
 
     def engage(self, reason: str) -> None:
@@ -24,10 +25,13 @@ def engage(self, reason: str) -> None:
         interleave with it, and the call returns only once every effect that
         was already committed (via ``Interlock.execute``) has finished. When
         ``engage`` returns, nothing is mid-effect and nothing new can start.
+        Every engagement is counted (see :attr:`engagements`), including one
+        that is released again before anyone looks.
         """
         with self._gate.cond:
             self._engaged = True
             self._reason = reason
+            self._engagements += 1
         self._gate.drain()
 
     def release(self) -> None:
@@ -40,6 +44,15 @@ def is_engaged(self) -> bool:
         """Whether the e-stop is currently engaged."""
         return self._engaged
 
+    @property
+    def engagements(self) -> int:
+        """How many times the e-stop has ever been engaged.
+
+        Lets a caller detect an engagement that was released again before
+        it looked — a final-state check on :attr:`is_engaged` cannot.
+        """
+        return self._engagements
+
     @property
     def reason(self) -> str:
         """The reason given at the last engagement, or ``""`` when clear."""
@@ -58,12 +71,16 @@ class InFlightGate:
     in flight under ``cond``; ``EStop.engage`` and ``LeaseManager.revoke``
     change state under ``cond`` and then wait for in-flight effects to drain.
     Effects run concurrently; only check-and-register is serialised. A halt
-    raised from inside an effect does not wait on its own thread's effect.
+    raised from inside an effect does not wait on its own thread's effect,
+    and concurrent halts raised from inside effects do not wait on each
+    other either: a draining thread's effect is suspended inside its halt
+    call, so two such halts waiting on each other could never return.
     """
 
     def __init__(self) -> None:
         self.cond = threading.Condition()
         self._in_flight: dict[tuple[str, int], int] = {}
+        self._halting: dict[int, int] = {}
 
     def enter(self, lease_id: str) -> None:
         """Register one effect for ``lease_id``. Call with ``cond`` held."""
@@ -83,10 +100,27 @@ def leave(self, lease_id: str) -> None:
     def _busy(self, lease_id: str | None) -> bool:
         me = threading.get_ident()
         return any(
-            tid != me and (lease_id is None or lid == lease_id) for lid, tid in self._in_flight
+            tid != me and tid not in self._halting and (lease_id is None or lid == lease_id)
+            for lid, tid in self._in_flight
         )
 
     def drain(self, lease_id: str | None = None) -> None:
-        """Block until no other thread has an effect (for ``lease_id``) in flight."""
+        """Block until no other thread has an effect (for ``lease_id``) in flight.
+
+        Effects on threads that are themselves inside a halt do not block
+        the drain (see the class docstring); effects still running normally
+        are waited on, and a halted thread's effect is waited on again as
+        soon as its halt returns and the effect resumes.
+        """
+        me = threading.get_ident()
         with self.cond:
-            self.cond.wait_for(lambda: not self._busy(lease_id))
+            self._halting[me] = self._halting.get(me, 0) + 1
+            try:
+                self.cond.wait_for(lambda: not self._busy(lease_id))
+            finally:
+                n = self._halting[me] - 1
+                if n <= 0:
+                    del self._halting[me]
+                else:
+                    self._halting[me] = n
+                self.cond.notify_all()
diff --git a/src/interlock/faults.py b/src/interlock/faults.py
index fc66265..311ae8d 100644
--- a/src/interlock/faults.py
+++ b/src/interlock/faults.py
@@ -4,11 +4,23 @@
 severity of minor, major, or critical. An uncleared fault of major or
 critical severity *gates* its resource: new permits on that resource are
 refused until the fault is cleared.
+
+When the bus is wired to an in-flight gate (as :class:`~interlock.Interlock`
+does), reports and clears take the gate's lock, so a gating change is
+ordered against the commit-time check-and-register in
+:meth:`Interlock.execute`: a report that returns before an effect commits
+is always seen by that effect's check, and one that lands after the check
+waits until registration has happened — it can never slip between them.
 """
 
 import time
+from contextlib import nullcontext
 from dataclasses import dataclass, field
 from itertools import count
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from interlock.estop import InFlightGate
 
 SEVERITIES = ("minor", "major", "critical")
 _GATING_SEVERITIES = ("major", "critical")
@@ -30,30 +42,37 @@ class Fault:
 class FaultBus:
     """Collects faults and answers whether a resource is gated."""
 
-    def __init__(self) -> None:
+    def __init__(self, gate: "InFlightGate | None" = None) -> None:
         self._faults: list[Fault] = []
         self._ids = count(1)
+        self._gate = gate
+
+    def _ordered(self):
+        """The gate's lock when wired, so state changes order against commits."""
+        return self._gate.cond if self._gate is not None else nullcontext()
 
     def report(self, code: str, severity: str, resource: str, detail: str = "") -> Fault:
         """Report a fault. Severity must be minor, major, or critical."""
         if severity not in SEVERITIES:
             raise ValueError(f"severity must be one of {SEVERITIES}, got {severity!r}")
-        fault = Fault(
-            id=next(self._ids),
-            code=code,
-            severity=severity,
-            resource=resource,
-            detail=detail,
-        )
-        self._faults.append(fault)
-        return fault
+        with self._ordered():
+            fault = Fault(
+                id=next(self._ids),
+                code=code,
+                severity=severity,
+                resource=resource,
+                detail=detail,
+            )
+            self._faults.append(fault)
+            return fault
 
     def clear(self, fault_id: int) -> Fault:
         """Mark a fault cleared. Clearing is recorded, never erased."""
-        for fault in self._faults:
-            if fault.id == fault_id:
-                fault.cleared = True
-                return fault
+        with self._ordered():
+            for fault in self._faults:
+                if fault.id == fault_id:
+                    fault.cleared = True
+                    return fault
         raise KeyError(f"no fault with id {fault_id}")
 
     def active(self, resource: str | None = None) -> list[Fault]:
diff --git a/src/interlock/ledger.py b/src/interlock/ledger.py
index e3a4354..2576c6d 100644
--- a/src/interlock/ledger.py
+++ b/src/interlock/ledger.py
@@ -18,6 +18,7 @@
 import io
 import json
 import os
+import threading
 from datetime import UTC, datetime
 from typing import Any, TextIO
 
@@ -27,7 +28,13 @@
 
 
 class Ledger:
-    """Append-only JSONL log with format descriptor, series descriptors, and a closing index."""
+    """Append-only JSONL log with format descriptor, series descriptors, and a closing index.
+
+    Appends and close are serialised with an internal lock, so one ledger
+    may be shared across threads: concurrent first appends to a series
+    write exactly one series descriptor, and the closing index counts
+    every record actually written.
+    """
 
     def __init__(self, dest: str | TextIO | None = None, mode: str = "a") -> None:
         """Open a ledger.
@@ -49,6 +56,7 @@ def __init__(self, dest: str | TextIO | None = None, mode: str = "a") -> None:
             self._buf = dest
         self._series: dict[str, dict[str, Any]] = {}
         self._closed = False
+        self._lock = threading.Lock()
         self._write({"_type": "format", "format": FORMAT, "version": VERSION})
 
     def _write(self, obj: dict[str, Any]) -> None:
@@ -76,33 +84,35 @@ def append(self, series: str, record: dict[str, Any]) -> dict[str, Any]:
         The first append to a new series writes a series descriptor declaring
         the schema (the record's sorted keys). Returns the stamped record.
         """
-        if self._closed:
-            raise ValueError("ledger is closed")
-        if series not in self._series:
-            self._write({"_type": "series", "series": series, "schema": sorted(record.keys())})
-            self._series[series] = {"count": 0, "first_ts": None, "last_ts": None}
-        entry = dict(record)
-        entry["_type"] = "record"
-        entry["_series"] = series
-        entry["_ts"] = datetime.now(UTC).isoformat()
-        info = self._series[series]
-        info["count"] += 1
-        info["last_ts"] = entry["_ts"]
-        if info["first_ts"] is None:
-            info["first_ts"] = entry["_ts"]
-        self._write(entry)
-        return entry
+        with self._lock:
+            if self._closed:
+                raise ValueError("ledger is closed")
+            if series not in self._series:
+                self._write({"_type": "series", "series": series, "schema": sorted(record.keys())})
+                self._series[series] = {"count": 0, "first_ts": None, "last_ts": None}
+            entry = dict(record)
+            entry["_type"] = "record"
+            entry["_series"] = series
+            entry["_ts"] = datetime.now(UTC).isoformat()
+            info = self._series[series]
+            info["count"] += 1
+            info["last_ts"] = entry["_ts"]
+            if info["first_ts"] is None:
+                info["first_ts"] = entry["_ts"]
+            self._write(entry)
+            return entry
 
     def close(self) -> None:
         """Append the closing index. Idempotent: closing twice writes one index."""
-        if self._closed:
-            return
-        self._closed = True
-        index = {name: dict(info) for name, info in self._series.items()}
-        self._write({"_type": "index", "series": index})
-        if self._owns_handle:
-            os.fsync(self._buf.fileno())  # one fsync per run: the closed file is on disk
-            self._buf.close()
+        with self._lock:
+            if self._closed:
+                return
+            self._closed = True
+            index = {name: dict(info) for name, info in self._series.items()}
+            self._write({"_type": "index", "series": index})
+            if self._owns_handle:
+                os.fsync(self._buf.fileno())  # one fsync per run: the closed file is on disk
+                self._buf.close()
 
     def read_series(self, series: str):
         """Yield the record dicts of one series, in append order.

From 5853f51e75ff00c203a04409b18d42809db1b5e6 Mon Sep 17 00:00:00 2001
From: FetchDoge <297783350+FetchDoge@users.noreply.github.com>
Date: Fri, 23 Oct 2026 18:25:00 +0000
Subject: [PATCH 3/4] test(inflight): synchronize scheduled halts on events;
 regression tests for the review fixes

---
 tests/test_inflight.py | 173 ++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 170 insertions(+), 3 deletions(-)

diff --git a/tests/test_inflight.py b/tests/test_inflight.py
index a46f355..cf6f1a2 100644
--- a/tests/test_inflight.py
+++ b/tests/test_inflight.py
@@ -8,12 +8,13 @@
 returns only once no effect is mid-flight.
 """
 
+import json
 import threading
 import time
 
 import pytest
 
-from interlock import EStopEngaged, Interlock, LeaseRevoked
+from interlock import EStopEngaged, FaultActive, Interlock, LeaseRevoked, Ledger
 
 
 def _series(ilk, name):
@@ -30,8 +31,17 @@ def test_estop_during_slow_preparation_refuses_effect():
     ilk = Interlock()
     lease = ilk.grant("payments", "agent", ttl_seconds=60)
     calls = []
-    t = _later(0.1, lambda: ilk.estop.engage("gauntlet"))
+    halted = threading.Event()
+
+    def engage():
+        ilk.estop.engage("gauntlet")
+        halted.set()
+
+    t = _later(0.1, engage)
     time.sleep(0.3)  # slow rail: authority was fine at the start
+    # Synchronize on the halt having landed, not on elapsed wall time:
+    # a Timer has no scheduling deadline on a loaded worker.
+    assert halted.wait(5), "the scheduled e-stop never landed"
     with pytest.raises(EStopEngaged):
         ilk.execute(lease, "pay", "merchant", lambda: calls.append("paid"))
     t.join()
@@ -44,8 +54,15 @@ def test_revoke_during_slow_preparation_refuses_effect():
     ilk = Interlock()
     lease = ilk.grant("payments", "agent", ttl_seconds=60)
     calls = []
-    t = _later(0.1, lambda: ilk.leases.revoke(lease))
+    halted = threading.Event()
+
+    def revoke():
+        ilk.leases.revoke(lease)
+        halted.set()
+
+    t = _later(0.1, revoke)
     time.sleep(0.3)
+    assert halted.wait(5), "the scheduled revocation never landed"
     with pytest.raises(LeaseRevoked):
         ilk.execute(lease, "pay", "merchant", lambda: calls.append("paid"))
     t.join()
@@ -125,3 +142,153 @@ def test_halt_from_inside_an_effect_does_not_deadlock():
     ilk.execute(lease, "pay", "m", lambda: ilk.estop.engage("effect saw fraud"))
     with pytest.raises(EStopEngaged):
         ilk.execute(lease, "pay", "m", lambda: None)
+
+
+# --- Codex review regressions (PR #18) -------------------------------------
+
+
+def test_concurrent_halts_from_inside_effects_do_not_deadlock():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    barrier = threading.Barrier(2)
+    returned = []
+
+    def halting(name):
+        def effect():
+            barrier.wait(5)
+            ilk.estop.engage(f"halt from {name}")
+            returned.append(name)
+
+        return effect
+
+    threads = [
+        threading.Thread(
+            target=lambda name=name: ilk.execute(lease, "pay", "m", halting(name)),
+            daemon=True,
+        )
+        for name in ("a", "b")
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(5)
+    assert not any(t.is_alive() for t in threads), "concurrent in-effect halts deadlocked"
+    assert sorted(returned) == ["a", "b"]
+    assert ilk.estop.is_engaged
+
+
+def test_concurrent_in_effect_revocations_do_not_deadlock():
+    ilk = Interlock()
+    lease_a = ilk.grant("payments", "agent-a", ttl_seconds=60)
+    lease_b = ilk.grant("payments", "agent-b", ttl_seconds=60)
+    barrier = threading.Barrier(2)
+    done = []
+
+    def revoking(own, other, name):
+        def effect():
+            barrier.wait(5)
+            ilk.leases.revoke(other)
+            done.append(name)
+
+        return effect
+
+    threads = [
+        threading.Thread(
+            target=lambda: ilk.execute(lease_a, "pay", "m", revoking(lease_a, lease_b, "a")),
+            daemon=True,
+        ),
+        threading.Thread(
+            target=lambda: ilk.execute(lease_b, "pay", "m", revoking(lease_b, lease_a, "b")),
+            daemon=True,
+        ),
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(5)
+    assert not any(t.is_alive() for t in threads), "mutual in-effect revocations deadlocked"
+    assert sorted(done) == ["a", "b"]
+    assert ilk.leases.is_revoked(lease_a) and ilk.leases.is_revoked(lease_b)
+
+
+def test_transient_estop_inside_permit_block_records_violation():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    with ilk.permit(lease, "pay", "m"):
+        ilk.estop.engage("mid-block")
+        ilk.estop.release()
+    v = _series(ilk, "violations")[-1]
+    assert v["reason"] == "estop" and v["lease_id"] == lease.id
+
+
+def test_execute_refuses_deferred_effects():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    calls = []
+
+    async def async_effect():
+        calls.append("async")
+
+    def gen_effect():
+        yield 1
+        calls.append("gen")
+
+    with pytest.raises(TypeError):
+        ilk.execute(lease, "pay", "m", async_effect)
+    with pytest.raises(TypeError):
+        ilk.execute(lease, "pay", "m", gen_effect)
+    with pytest.raises(TypeError):
+        ilk.execute(lease, "pay", "m", lambda: async_effect())
+    with pytest.raises(TypeError):
+        ilk.execute(lease, "pay", "m", lambda: gen_effect())
+    assert calls == []
+    # The refusals leave no stale in-flight entry behind.
+    assert ilk.execute(lease, "pay", "m", lambda: 42) == 42
+
+
+def test_fault_report_is_ordered_against_commit_registration():
+    ilk = Interlock()
+    lease = ilk.grant("payments", "agent", ttl_seconds=60)
+    reported = threading.Event()
+
+    def report():
+        ilk.faults.report("tool_failure", "critical", "payments")
+        reported.set()
+
+    with ilk._gate.cond:
+        t = threading.Thread(target=report)
+        t.start()
+        assert not reported.wait(0.3), "fault report interleaved with the commit gate"
+    t.join(5)
+    assert reported.is_set()
+    with pytest.raises(FaultActive):
+        ilk.execute(lease, "pay", "m", lambda: None)
+
+
+def test_concurrent_execute_ledger_index_matches_records(tmp_path):
+    path = tmp_path / "ledger.jsonl"
+    ledger = Ledger(str(path))
+    ilk = Interlock(ledger=ledger)
+    lease = ilk.grant("payments", "agent", ttl_seconds=600)
+    n_threads, per_thread = 16, 25
+    barrier = threading.Barrier(n_threads)
+
+    def worker():
+        barrier.wait(10)
+        for _ in range(per_thread):
+            ilk.execute(lease, "pay", "m", lambda: None)
+
+    threads = [threading.Thread(target=worker) for _ in range(n_threads)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    ledger.close()
+    assert len(list(ledger.read_series("permits"))) == n_threads * per_thread
+    objs = [json.loads(line) for line in path.read_text().splitlines()]
+    descriptors = [
+        obj for obj in objs if obj.get("_type") == "series" and obj.get("series") == "permits"
+    ]
+    assert len(descriptors) == 1
+    index = [obj for obj in objs if obj.get("_type") == "index"][-1]
+    assert index["series"]["permits"]["count"] == n_threads * per_thread

From f2742a98a27f0311a7412f07c2fd7ec17fabcdc7 Mon Sep 17 00:00:00 2001
From: FetchDoge <297783350+FetchDoge@users.noreply.github.com>
Date: Fri, 23 Oct 2026 18:25:47 +0000
Subject: [PATCH 4/4] docs: route irreversible effects through execute() in
 README and quickstart; changelog

---
 CHANGELOG.md              | 37 +++++++++++++++++++++++++++++++++++++
 README.md                 | 13 +++++++------
 docs/guides/quickstart.md | 28 ++++++++++++++++------------
 3 files changed, 60 insertions(+), 18 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index df3954e..0b2b6f7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,7 +7,44 @@ the current version is `interlock.__version__` (mirrored into `VERSION` by
 `scripts/release.sh`).
 
 ## [Unreleased]
+### Added
+- `Interlock.execute(lease, action, target, effect)`: the gate is re-proven at the commit
+  point, atomically with registering the effect as in flight. `EStop.engage()` and
+  `LeaseManager.revoke()` are ordered against it and return only after committed effects
+  finish, closing the check-then-act race where an e-stop or revocation landing during a
+  slow preparation still let the effect run (gauntlet `revoke-during-spend`). `permit()`
+  keeps its entry-only check and now appends a `violations` record if authority is lost
+  while the block runs. Ledger `permits`/`denials` records gain a `phase` field (`entry`
+  or `commit`). Regression tests: `tests/test_inflight.py`.
+
 ### Fixed
+- Codex review fixes on the commit-point gate (PR #18), each with a regression test in
+  `tests/test_inflight.py` unless noted:
+  - Gating fault reports are ordered against commit registration: `FaultBus` takes the
+    in-flight gate's lock on `report()`/`clear()`, so a major/critical fault can no
+    longer land between `execute()`'s check and its registration and be missed.
+  - Concurrent halts raised from inside effects no longer deadlock: a draining halt
+    does not wait on effects whose threads are themselves inside a halt. Applies to
+    `estop.engage()` and to mutual in-effect `leases.revoke()` calls.
+  - `Ledger` appends and `close()` are serialised with an internal lock, so concurrent
+    `execute()` calls can no longer write duplicate series descriptors or leave the
+    closing index undercounting records.
+  - `permit()` records a `violations` entry when the e-stop is engaged and released
+    again inside the block; the exit check compares `EStop.engagements` counts instead
+    of only the final `is_engaged` state.
+  - `execute()` refuses deferred effects: coroutine/generator callables raise
+    `TypeError` before the gate runs, and a deferred result from a plain callable is
+    closed unstarted. Previously the callable merely created the deferred object, the
+    in-flight registration was dropped immediately, and the body could run after a
+    halt had returned.
+  - `execute()` establishes its in-flight cleanup guard before registration, so an
+    asynchronous exception in the gap after `enter()` cannot leak a stale in-flight
+    entry that blocks later halts forever (structural fix; the interleaving needs an
+    async exception in a two-step window and has no deterministic regression test).
+  - The two scheduled-halt tests in `tests/test_inflight.py` synchronize on an event
+    instead of assuming a `threading.Timer` fired within a fixed sleep.
+  - README and the quickstart guide route irreversible effects through `execute()`
+    instead of `permit()` blocks.
 - `Ledger(path)` no longer erases an existing audit log. A path opens in append mode by
   default, so reopening a ledger keeps every earlier run and starts a new self-describing
   segment. `Ledger(path, mode="w")` truncates deliberately; any other mode raises
diff --git a/README.md b/README.md
index 890eba1..530e381 100644
--- a/README.md
+++ b/README.md
@@ -10,7 +10,7 @@ Mobile robots solved this a decade ago. Boston Dynamics' Spot admits intent thro
 
 ## What it guarantees
 
-Inside the substrate, **no action executes without a live lease, a clear e-stop, and no gating fault.** `Interlock.permit()` checks in this order and raises on the first failure:
+Inside the substrate, **no action executes without a live lease, a clear e-stop, and no gating fault.** The gate — `Interlock.execute()` at the commit point, `Interlock.permit()` at block entry — checks in this order and raises on the first failure:
 
 1. e-stop clear, else `EStopEngaged`
 2. lease issued by this `Interlock` and unaltered, else `LeaseNotIssued`; then unrevoked and unexpired, else `LeaseRevoked` or `LeaseExpired`
@@ -41,9 +41,10 @@ ilk = Interlock()
 # GRANT: authority comes into existence only here, and it expires.
 lease = ilk.grant(resource="docs:write", holder="research-agent", ttl=1800)
 
-# EXERCISE: every action passes the gate.
-with ilk.permit(lease, action="write", target="brief.md"):
-    ...  # raises EStopEngaged, LeaseNotIssued, LeaseRevoked, LeaseExpired, or FaultActive otherwise
+# EXERCISE: every action passes the gate. execute() re-proves the gate at
+# the commit point, atomically with starting the effect — a halt that lands
+# while the action is being prepared refuses it instead of racing it.
+ilk.execute(lease, "write", "brief.md", effect)  # raises EStopEngaged, LeaseNotIssued, LeaseRevoked, LeaseExpired, or FaultActive otherwise
 
 # A major or critical fault refuses further permits on the affected resource.
 ilk.faults.report(code="TOOL_TIMEOUT", severity="major", resource="web:read")
@@ -66,14 +67,14 @@ The fastest path to a first lease is [`docs/guides/quickstart.md`](docs/guides/q
 
 | Condition | Behavior |
 |---|---|
-| E-stop engaged | Every `permit()` raises `EStopEngaged` and logs a `denials` record |
+| E-stop engaged | Every gate check raises `EStopEngaged` and logs a `denials` record |
 | Lease expired (no keepalive in time) | `LeaseExpired`. A late keepalive also raises |
 | Lease not issued by this `Interlock` (hand-built, from another instance, or edited after issue) | `LeaseNotIssued` and a `denials` record with reason `lease_not_issued` |
 | Lease revoked | `LeaseRevoked`, permanently. Resetting `lease.revoked` on the object does not restore it |
 | `major`/`critical` fault on the resource | `FaultActive` until the fault is cleared. `minor` faults are recorded but do not gate |
 
 **Known limits:**
-- The gate runs at `permit()` entry. An e-stop engaged *during* a `with` block does not interrupt code already running inside it; it refuses the next permit.
+- `permit()` checks at block entry only. An e-stop engaged *during* a `with` block does not interrupt code already running inside it; the block's exit logs a `violations` record and the next permit is refused. For an effect that must not start once authority is gone, use `execute()`: it re-proves the gate at the commit point, and an e-stop or revocation returns only after already-committed effects have finished.
 - State is in-process and in-memory. Interlock gates actions routed through it and cannot stop an agent that bypasses the substrate. It is one layer of defense in depth.
 - The ledger is append-only and self-describing JSONL (format descriptor, per-series schemas, closing index). Reopening a path appends a new segment and never truncates unless you pass `mode="w"`. Every record is flushed as it is written, so it survives a crash of the writing process. Only `close()` fsyncs, so a power loss can still drop the last records. It is **not** hash-chained or signed, so treat it as an audit log, not tamper evidence.
 
diff --git a/docs/guides/quickstart.md b/docs/guides/quickstart.md
index 3135a46..bc3f34a 100644
--- a/docs/guides/quickstart.md
+++ b/docs/guides/quickstart.md
@@ -49,15 +49,21 @@ checks the e-stop, lease state, and resource faults at the time of entry.
 ## 3. Exercise it through the gate
 
 ```python
-with ilk.permit(lease, action="write", target="notes.md"):
-    Path("notes.md").write_text("hello")
+ilk.execute(lease, "write", "notes.md", lambda: Path("notes.md").write_text("hello"))
 ```
 
 The gate checks, in order: e-stop clear, lease issued by this `Interlock`
-and live, and no gating fault on `docs:write`. If any check fails, the
-context manager raises instead of yielding: `EStopEngaged`,
-`LeaseNotIssued`, `LeaseExpired`/`LeaseRevoked`, or `FaultActive`. The check
-controls admission to the context. It is not an OS sandbox, and it cannot
+and live, and no gating fault on `docs:write`. If any check fails,
+`execute()` raises instead of running the effect: `EStopEngaged`,
+`LeaseNotIssued`, `LeaseExpired`/`LeaseRevoked`, or `FaultActive`.
+
+`execute()` runs the check at the commit point — atomically with starting
+the effect — so an e-stop or revocation that lands while the effect is
+being prepared still refuses it, and a halt call returns only after
+already-committed effects have finished. This is the path for any effect
+that cannot be taken back. (`permit()` also exists as a context manager,
+but it checks once at block entry only; use it solely for work that is
+safe to abandon mid-block.) The gate is not an OS sandbox, and it cannot
 interrupt an action that has already started.
 
 ## 4. Watch a fault refuse
@@ -68,8 +74,7 @@ fault = ilk.faults.report(
 )
 
 try:
-    with ilk.permit(lease, action="write", target="notes.md"):
-        Path("notes.md").write_text("should not be admitted")
+    ilk.execute(lease, "write", "notes.md", lambda: Path("notes.md").write_text("no"))
 except FaultActive:
     pass
 else:
@@ -77,8 +82,7 @@ else:
 
 assert fault.id is not None
 ilk.faults.clear(fault.id)
-with ilk.permit(lease, action="write", target="notes.md"):
-    Path("notes.md").write_text("resumed")
+ilk.execute(lease, "write", "notes.md", lambda: Path("notes.md").write_text("resumed"))
 ```
 
 The refused entry is appended to the ledger. Clearing the fault permits a
@@ -89,8 +93,8 @@ new admission; it does not retroactively authorize the refused action.
 ```python
 ilk.estop.engage("taking a look")
 try:
-    with ilk.permit(lease, action="write", target="notes.md"):
-        raise AssertionError("e-stop must refuse admission")
+    ilk.execute(lease, "write", "notes.md", lambda: None)
+    raise AssertionError("e-stop must refuse admission")
 except EStopEngaged:
     pass
 ilk.estop.release()  # deliberate, never automatic