Expose lease loss to execution boundaries

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d

rein_aharness/claim_loop.py

@@ -28,6 +28,7 @@ from rein_aharness.glas_execution import (
     GlasExecutionError,
     execute_profiled_run,
 )
+from rein_aharness.lease_monitor import LeaseMonitor
 from rein_aharness.ops_run_client import (
     ActivityCoreOpsClient,
     OpsRun,
@@ -68,6 +69,7 @@ class _Heartbeat:
         self._lease = lease_seconds
         self._stop = threading.Event()
         self._thread: threading.Thread | None = None
+        self.monitor = LeaseMonitor()
 
     def start(self) -> None:
         interval = _heartbeat_interval(self._lease)
@@ -78,7 +80,13 @@ class _Heartbeat:
                     self._client.heartbeat(self._run_id, lease_seconds=self._lease)
                     logger.info("heartbeat ok run_id=%s", self._run_id)
                 except OpsRunError as exc:
-                    logger.warning("heartbeat failed run_id=%s: %s", self._run_id, exc)
+                    evidence = self.monitor.mark_lost(type(exc).__name__)
+                    logger.warning(
+                        "heartbeat lost run_id=%s error_type=%s observed_at=%s",
+                        self._run_id,
+                        evidence.error_type,
+                        evidence.observed_at,
+                    )
 
         self._thread = threading.Thread(
             target=_loop, name=f"ops-hb-{self._run_id[:8]}", daemon=True
@@ -89,6 +97,7 @@ class _Heartbeat:
         self._stop.set()
         if self._thread is not None:
             self._thread.join(timeout=5.0)
+        self.monitor.stop()
 
 
 def process_one(

rein_aharness/lease_monitor.py

@@ -0,0 +1,77 @@
+"""Observable lease state for long-running executions.
+
+The monitor is deliberately independent of an adapter.  A heartbeat worker can
+mark a lease lost, while the executor observes that fact and decides how to
+cancel or refuse acceptance at its own boundary.
+"""
+
+from __future__ import annotations
+
+import threading
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Callable
+
+
+@dataclass(frozen=True)
+class LeaseLoss:
+    """Bounded evidence that the worker no longer owns a lease."""
+
+    error_type: str
+    observed_at: str
+
+
+class LeaseMonitor:
+    """Thread-safe active/lost/stopped state with a one-shot loss callback."""
+
+    def __init__(self, on_lost: Callable[[LeaseLoss], None] | None = None) -> None:
+        self._lock = threading.Lock()
+        self._lost = threading.Event()
+        self._stopped = False
+        self._evidence: LeaseLoss | None = None
+        self._on_lost = on_lost
+
+    @property
+    def lost(self) -> bool:
+        return self._lost.is_set()
+
+    @property
+    def stopped(self) -> bool:
+        with self._lock:
+            return self._stopped
+
+    @property
+    def evidence(self) -> LeaseLoss | None:
+        with self._lock:
+            return self._evidence
+
+    def wait_lost(self, timeout: float | None = None) -> bool:
+        """Wait until lease loss is observed; return whether it occurred."""
+        return self._lost.wait(timeout)
+
+    def mark_lost(self, error_type: str) -> LeaseLoss:
+        """Record the first loss and invoke the callback once.
+
+        Error text is intentionally not retained: provider responses can carry
+        credentials, prompts, or other unbounded data.
+        """
+        callback: Callable[[LeaseLoss], None] | None = None
+        with self._lock:
+            if self._evidence is None:
+                evidence = LeaseLoss(
+                    error_type=error_type[:80] or "OpsRunError",
+                    observed_at=datetime.now(timezone.utc).isoformat(),
+                )
+                self._evidence = evidence
+                self._lost.set()
+                callback = self._on_lost
+            else:
+                evidence = self._evidence
+        if callback is not None:
+            callback(evidence)
+        return evidence
+
+    def stop(self) -> None:
+        with self._lock:
+            self._stopped = True
+