fix: bounded grace + audit logging for lease expiry (§14) (#90)

Add a configurable grace window (lease_expiry_grace_sec, default 1s) to
expires_at enforcement in validate_lease_op and the watchdog so small clock
skew does not spuriously expire a lease, and emit a structured
"lease_expired" audit log when the watchdog terminates a job.

Fixes #90

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: nficano

src/arcp/_runtime/_job_runner.py

@@ -201,14 +201,18 @@ async def _revoke_with_retry(runtime: ARCPRuntime, credential_id: str) -> bool:
 
 async def _lease_watchdog(runtime: ARCPRuntime, job: Job, expires_at_iso: str) -> None:
     expiry = _parse_iso_utc(expires_at_iso)
-    delay = (expiry - dt.datetime.now(dt.UTC)).total_seconds()
+    # Apply the same bounded grace (§14) the op-check uses so the proactive
+    # watchdog does not fire ahead of an authority-bearing op-time check.
+    delay = (expiry - dt.datetime.now(dt.UTC)).total_seconds() + runtime.lease_expiry_grace_sec
     if delay > 0:
         try:
             await asyncio.sleep(delay)
         except asyncio.CancelledError:
             return
     if job.state != "running":
         return
+    # §14: log lease expirations for audit.
+    runtime.logger.info("lease_expired", job_id=job.job_id, expires_at=expires_at_iso)
     await job.emit_error(
         JobErrorPayload(
             code="LEASE_EXPIRED",

src/arcp/_runtime/job.py

@@ -306,6 +306,7 @@ def authorize(
             LeaseOpContext(capability=capability, target=target, cost=cost, now=now),
             constraints=self.lease_constraints,
             budget=self.job.budget if "cost.budget" in self.lease else None,
+            grace_sec=self.runtime.lease_expiry_grace_sec,
         )
 
     def authorize_model(self, model_id: str, *, now: dt.datetime | None = None) -> None:
@@ -314,6 +315,7 @@ def authorize_model(self, model_id: str, *, now: dt.datetime | None = None) -> N
             self.lease,
             LeaseOpContext(capability="model.use", target=model_id, now=now),
             constraints=self.lease_constraints,
+            grace_sec=self.runtime.lease_expiry_grace_sec,
         )
 
     async def rotate_credential(self, credential_id: str, new_value: str) -> None:

src/arcp/_runtime/lease.py

@@ -209,17 +209,27 @@ def _glob_lang_subset(child: str, parent: str) -> bool:
     return True
 
 
+DEFAULT_EXPIRY_GRACE_SEC: float = 1.0
+"""Bounded grace window applied to `expires_at` enforcement (§14).
+
+`expires_at` is an absolute ISO timestamp, so its comparison is inherently
+wall-clock; deployments rely on NTP discipline. The grace absorbs small clock
+skew (e.g. an NTP step) so a barely-unexpired lease is not spuriously rejected.
+"""
+
+
 def validate_lease_op(
     lease: Lease,
     ctx: LeaseOpContext,
     *,
     constraints: LeaseConstraints | None = None,
     budget: dict[str, Decimal] | None = None,
+    grace_sec: float = DEFAULT_EXPIRY_GRACE_SEC,
 ) -> None:
     """Authorize an op against the lease: pattern match, expiry, budget. Raise on violation."""
     if constraints is not None and constraints.expires_at is not None:
         n = ctx.now or dt.datetime.now(dt.UTC)
-        expiry = _parse_iso_utc(constraints.expires_at)
+        expiry = _parse_iso_utc(constraints.expires_at) + dt.timedelta(seconds=grace_sec)
         if n >= expiry:
             raise LeaseExpiredError("lease has expired")
 

src/arcp/_runtime/server.py

@@ -109,6 +109,8 @@ class ARCPRuntime:
         idempotency_ttl_sec: TTL for entries in the idempotency store.
         max_concurrent_jobs: Cap on simultaneous running agent tasks.
         chunk_size_cap: Per-`result_chunk` size cap (spec §14 SHOULD).
+        lease_expiry_grace_sec: Bounded grace window (§14) added to
+            `expires_at` enforcement to absorb small clock skew.
         job_authorization_policy: Hook controlling list/subscribe/cancel
             visibility (defaults to same-principal).
         event_log: Storage for replayable envelopes. Defaults to
@@ -134,6 +136,7 @@ def __init__(  # noqa: PLR0913
         idempotency_ttl_sec: float = 24 * 60 * 60,
         max_concurrent_jobs: int = 100,
         chunk_size_cap: int = 1024 * 1024,
+        lease_expiry_grace_sec: float = 1.0,
         job_authorization_policy: JobAuthorizationPolicy | None = None,
         event_log: EventLog | None = None,
         credential_provisioner: CredentialProvisioner | None = None,
@@ -159,6 +162,7 @@ def __init__(  # noqa: PLR0913
         self.resume_window_sec = resume_window_sec
         self.max_concurrent_jobs = max_concurrent_jobs
         self.chunk_size_cap = chunk_size_cap
+        self.lease_expiry_grace_sec = lease_expiry_grace_sec
         self.idempotency = IdempotencyStore(ttl_sec=idempotency_ttl_sec)
         self.event_log: EventLog = event_log if event_log is not None else InMemoryEventLog()
         self.policy = job_authorization_policy or _default_authz_policy

tests/state/test_lease_expiry_grace.py

@@ -0,0 +1,98 @@
+"""#90 — bounded grace on expires_at enforcement and lease-expiry audit logging (§14)."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import datetime as dt
+from typing import Any
+
+import pytest
+
+from arcp import (
+    Capabilities,
+    ClientInfo,
+    LeaseConstraints,
+    LeaseExpiredError,
+    RuntimeInfo,
+    pair_memory_transports,
+)
+from arcp._messages.execution import LeaseConstraints as LC
+from arcp._runtime.lease import LeaseOpContext, validate_lease_op
+from arcp.client import ARCPClient
+from arcp.runtime import ARCPRuntime, StaticBearerVerifier
+
+
+def test_expiry_grace_window_is_applied_and_configurable() -> None:
+    now = dt.datetime.now(dt.UTC)
+    # expires_at 0.5s in the past relative to `now`.
+    expires = (now - dt.timedelta(seconds=0.5)).isoformat().replace("+00:00", "Z")
+    constraints = LC(expires_at=expires)
+    lease = {"fs.read": ["*"]}
+    ctx = LeaseOpContext(capability="fs.read", target="f", now=now)
+
+    # Within a 1s grace window -> still authorized.
+    validate_lease_op(lease, ctx, constraints=constraints, grace_sec=1.0)
+
+    # Grace disabled -> expired.
+    with pytest.raises(LeaseExpiredError):
+        validate_lease_op(lease, ctx, constraints=constraints, grace_sec=0.0)
+
+
+class _CapturingLogger:
+    def __init__(self) -> None:
+        self.records: list[tuple[str, dict[str, Any]]] = []
+
+    def bind(self, **_kw: Any) -> _CapturingLogger:
+        return self
+
+    def info(self, event: str, **kw: Any) -> None:
+        self.records.append((event, kw))
+
+    def warning(self, *_a: Any, **_k: Any) -> None: ...
+    def error(self, *_a: Any, **_k: Any) -> None: ...
+    def debug(self, *_a: Any, **_k: Any) -> None: ...
+    def exception(self, *_a: Any, **_k: Any) -> None: ...
+
+
+async def test_lease_expiration_is_logged_for_audit() -> None:
+    logger = _CapturingLogger()
+    rt = ARCPRuntime(
+        runtime=RuntimeInfo(name="r", version="1"),
+        bearer=StaticBearerVerifier({"tok": "p1"}),
+        heartbeat_interval_sec=None,
+        lease_expiry_grace_sec=0.1,
+        logger=logger,
+    )
+
+    async def slow(input_value, ctx):
+        await asyncio.sleep(5.0)
+        return "never"
+
+    rt.register_agent("slow", slow)
+
+    server_t, client_t = pair_memory_transports()
+    accept_task = asyncio.create_task(rt.accept(server_t))
+    client = ARCPClient(
+        client=ClientInfo(name="c", version="1"),
+        token="tok",
+        capabilities=Capabilities(features=rt.capabilities.features),
+    )
+    await client.connect(client_t)
+    try:
+        expiry = (dt.datetime.now(dt.UTC) + dt.timedelta(milliseconds=100)).isoformat()
+        handle = await client.submit(
+            agent="slow",
+            lease_request={"fs.read": ["/tmp/*"]},
+            lease_constraints=LeaseConstraints(expires_at=expiry.replace("+00:00", "Z")),
+        )
+        with pytest.raises(LeaseExpiredError):
+            await asyncio.wait_for(handle.done, timeout=3.0)
+        assert any(event == "lease_expired" for event, _ in logger.records), logger.records
+    finally:
+        with contextlib.suppress(Exception):
+            await client.close()
+        accept_task.cancel()
+        with contextlib.suppress(asyncio.CancelledError, Exception):
+            await accept_task
+        await rt.close()