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 diff --git a/src/interlock/__init__.py b/src/interlock/__init__.py index 9e8e26f..85680ca 100644 --- a/src/interlock/__init__.py +++ b/src/interlock/__init__.py @@ -11,8 +11,10 @@ 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 from interlock._errors import ( EStopEngaged, @@ -23,12 +25,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,9 +64,10 @@ def __init__( clock: Callable[[], float] | None = None, ledger: Ledger | None = None, ) -> None: - self.leases = LeaseManager(clock) - self.estop = EStop() - self.faults = FaultBus() + self._gate = InFlightGate() + self.leases = LeaseManager(clock, gate=self._gate) + self.estop = EStop(self._gate) + self.faults = FaultBus(gate=self._gate) self.ledger = ledger if ledger is not None else Ledger() def grant( @@ -92,24 +97,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 +126,93 @@ 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. + 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) - yield lease + engagements = self.estop.engagements + try: + yield lease + finally: + reason = None + if self.estop.engagements != engagements: + 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. + + ``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. + """ + 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) + 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: + if registered: + self._gate.leave(lease.id) def claim( self, diff --git a/src/interlock/estop.py b/src/interlock/estop.py index 667640e..44935f7 100644 --- a/src/interlock/estop.py +++ b/src/interlock/estop.py @@ -4,20 +4,35 @@ 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._engagements = 0 + 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. + 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: """Release the e-stop. An operator decision, never a timeout.""" @@ -29,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.""" @@ -38,3 +62,65 @@ 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, + 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.""" + 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 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. + + 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._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/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/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. diff --git a/tests/test_inflight.py b/tests/test_inflight.py new file mode 100644 index 0000000..cf6f1a2 --- /dev/null +++ b/tests/test_inflight.py @@ -0,0 +1,294 @@ +"""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 json +import threading +import time + +import pytest + +from interlock import EStopEngaged, FaultActive, Interlock, LeaseRevoked, Ledger + + +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 = [] + 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() + 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 = [] + 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() + 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) + + +# --- 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