Classify heartbeat lease failures

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

WORK-RECORDS.md

@@ -27,7 +27,7 @@
 | task | HARNESS-WP-0002-T03 | done | — | workplans/HARNESS-WP-0002-rename-and-glas-harness-alignment.md |
 | task | HARNESS-WP-0002-T04 | done | — | workplans/HARNESS-WP-0002-rename-and-glas-harness-alignment.md |
 | task | HARNESS-WP-0003-T01 | progress | — | workplans/HARNESS-WP-0003-governed-runtime-integrity.md |
-| task | HARNESS-WP-0003-T02 | wait | — | workplans/HARNESS-WP-0003-governed-runtime-integrity.md |
+| task | HARNESS-WP-0003-T02 | progress | — | workplans/HARNESS-WP-0003-governed-runtime-integrity.md |
 | task | HARNESS-WP-0003-T03 | wait | — | workplans/HARNESS-WP-0003-governed-runtime-integrity.md |
 | task | HARNESS-WP-0003-T04 | wait | — | workplans/HARNESS-WP-0003-governed-runtime-integrity.md |
 | task | HARNESS-WP-0003-T05 | wait | — | workplans/HARNESS-WP-0003-governed-runtime-integrity.md |

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,22 @@ 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)
+                    if exc.lease_rejected:
+                        evidence = self.monitor.mark_lost(
+                            f"{type(exc).__name__}:{exc.status_code}"
+                        )
+                        logger.warning(
+                            "heartbeat lost run_id=%s error_type=%s observed_at=%s",
+                            self._run_id,
+                            evidence.error_type,
+                            evidence.observed_at,
+                        )
+                    else:
+                        logger.warning(
+                            "heartbeat transient failure run_id=%s error_type=%s",
+                            self._run_id,
+                            type(exc).__name__,
+                        )
 
         self._thread = threading.Thread(
             target=_loop, name=f"ops-hb-{self._run_id[:8]}", daemon=True
@@ -89,6 +106,7 @@ class _Heartbeat:
         self._stop.set()
         if self._thread is not None:
             self._thread.join(timeout=5.0)
+        self.monitor.stop()
 
 
 def process_one(
@@ -159,6 +177,8 @@ def process_one(
 
     hb = _Heartbeat(client, run.id, cfg.lease_seconds)
     hb.start()
+    execution_error: str | None = None
+    try:
         try:
             ar: ApproachResult = execute_approach(
                 run,
@@ -167,9 +187,38 @@ def process_one(
                 report_to_hub=report_to_hub,
                 commit=commit,
             )
+        except Exception as exc:
+            execution_error = type(exc).__name__
     finally:
         hb.stop()
 
+    if hb.monitor.lost:
+        loss = hb.monitor.evidence
+        assert loss is not None
+        return ProcessResult(
+            claimed=True,
+            run_id=run.id,
+            approach=approach,
+            ok=False,
+            reason=f"lease lost ({loss.error_type})",
+            detail={
+                "lease_loss": {
+                    "error_type": loss.error_type,
+                    "observed_at": loss.observed_at,
+                }
+            },
+        )
+
+    if execution_error is not None:
+        return ProcessResult(
+            claimed=True,
+            run_id=run.id,
+            approach=approach,
+            ok=False,
+            reason=f"approach failed ({execution_error})",
+            detail={"execution_error_type": execution_error},
+        )
+
     payload = {
         "approach": ar.approach,
         "ok": ar.ok,
@@ -233,6 +282,9 @@ def _process_profiled_run(
     """Execute a profiled row without consulting or falling back to legacy routing."""
     hb = _Heartbeat(client, run.id, client.config.lease_seconds)
     hb.start()
+    execution_error: str | None = None
+    execution_reason = ""
+    gateway_result: dict[str, Any] | None = None
     try:
         try:
             gateway_result = execute_profiled_run(
@@ -241,7 +293,30 @@ def _process_profiled_run(
                 report_to_hub=report_to_hub,
             )
         except GlasExecutionError as exc:
-            reason = str(exc)
+            execution_error = type(exc).__name__
+            execution_reason = str(exc)
+    finally:
+        hb.stop()
+
+    if hb.monitor.lost:
+        loss = hb.monitor.evidence
+        assert loss is not None
+        return ProcessResult(
+            claimed=True,
+            run_id=run.id,
+            approach=GLAS_APPROACH,
+            ok=False,
+            reason=f"lease lost ({loss.error_type})",
+            detail={
+                "lease_loss": {
+                    "error_type": loss.error_type,
+                    "observed_at": loss.observed_at,
+                }
+            },
+        )
+
+    if execution_error is not None:
+        reason = execution_reason or f"profiled execution failed ({execution_error})"
         try:
             out = client.fail(
                 run.id,
@@ -265,9 +340,8 @@ def _process_profiled_run(
             reason=reason,
             ops_state=out.state,
         )
-    finally:
-        hb.stop()
 
+    assert gateway_result is not None
     evidence = gateway_result["evidence"]
     ok = gateway_result["ok"]
     reason = str(evidence.get("error") or evidence.get("outcome") or "Glas execution failed")

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
+

rein_aharness/ops_run_client.py

@@ -33,7 +33,22 @@ DEFAULT_REPO_ROOTS = ("~", "~/work")
 
 
 class OpsRunError(RuntimeError):
-    pass
+    """Bounded Activity Core failure with optional HTTP classification."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        action: str | None = None,
+        status_code: int | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.action = action
+        self.status_code = status_code
+
+    @property
+    def lease_rejected(self) -> bool:
+        return self.status_code in {401, 403, 404, 409, 410, 412, 423}
 
 
 @dataclass
@@ -159,8 +174,14 @@ class ActivityCoreOpsClient:
                 timeout=self.config.timeout,
             )
             resp.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            raise OpsRunError(
+                f"list ops-runs failed: HTTP {exc.response.status_code}",
+                action="list",
+                status_code=exc.response.status_code,
+            ) from exc
         except httpx.HTTPError as exc:
-            raise OpsRunError(f"list ops-runs failed: {exc}") from exc
+            raise OpsRunError("list ops-runs failed: transport error", action="list") from exc
         data = resp.json()
         items = data.get("items") if isinstance(data, dict) else data
         if not isinstance(items, list):
@@ -190,8 +211,14 @@ class ActivityCoreOpsClient:
                 timeout=self.config.timeout,
             )
             resp.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            raise OpsRunError(
+                f"claim failed: HTTP {exc.response.status_code}",
+                action="claim",
+                status_code=exc.response.status_code,
+            ) from exc
         except httpx.HTTPError as exc:
-            raise OpsRunError(f"claim failed: {exc}") from exc
+            raise OpsRunError("claim failed: transport error", action="claim") from exc
         data = resp.json()
         items = data.get("items") if isinstance(data, dict) else []
         return [OpsRun.from_api(item) for item in items if isinstance(item, dict)]
@@ -247,8 +274,16 @@ class ActivityCoreOpsClient:
                 timeout=self.config.timeout,
             )
             resp.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            raise OpsRunError(
+                f"{action} ops_run {run_id} failed: HTTP {exc.response.status_code}",
+                action=action,
+                status_code=exc.response.status_code,
+            ) from exc
         except httpx.HTTPError as exc:
-            raise OpsRunError(f"{action} ops_run {run_id} failed: {exc}") from exc
+            raise OpsRunError(
+                f"{action} ops_run {run_id} failed: transport error", action=action
+            ) from exc
         return OpsRun.from_api(resp.json())
 
 

tests/test_claim_loop.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from unittest.mock import MagicMock, patch
+import time
 
 from rein_aharness.approaches import ApproachResult, APPROACH_FI_RESEARCH_BRIEF
 from rein_aharness.claim_loop import process_one, poll_peek
@@ -105,6 +106,77 @@ def test_process_one_failure_reopens() -> None:
     assert client.fail.call_args.kwargs["reopen"] is True
 
 
+def test_process_one_refuses_close_after_lease_loss() -> None:
+    client = MagicMock(spec=ActivityCoreOpsClient)
+    client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
+    client.claim.return_value = [_claimed_run()]
+    client.heartbeat.side_effect = OpsRunError("lease rejected", status_code=409)
+    ar = ApproachResult(
+        ok=True,
+        approach=APPROACH_FI_RESEARCH_BRIEF,
+        result={"path": "briefs/x.md"},
+        reason="ok",
+    )
+
+    def slow_execute(*_args, **_kwargs):
+        time.sleep(0.05)
+        return ar
+
+    with (
+        patch("rein_aharness.claim_loop._heartbeat_interval", return_value=0.01),
+        patch("rein_aharness.claim_loop.execute_approach", side_effect=slow_execute),
+    ):
+        result = process_one(client)
+
+    assert result.ok is False
+    assert result.reason.startswith("lease lost")
+    assert "lease_loss" in result.detail
+    client.complete.assert_not_called()
+    client.fail.assert_not_called()
+
+
+def test_process_one_records_adapter_exception_without_unbound_result() -> None:
+    client = MagicMock(spec=ActivityCoreOpsClient)
+    client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
+    client.claim.return_value = [_claimed_run()]
+
+    with patch(
+        "rein_aharness.claim_loop.execute_approach",
+        side_effect=RuntimeError("provider output must not be retained"),
+    ):
+        result = process_one(client)
+
+    assert result.ok is False
+    assert result.reason == "approach failed (RuntimeError)"
+    assert result.detail == {"execution_error_type": "RuntimeError"}
+    client.complete.assert_not_called()
+    client.fail.assert_not_called()
+
+
+def test_profiled_exception_after_lease_loss_skips_close() -> None:
+    client = MagicMock(spec=ActivityCoreOpsClient)
+    client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
+    run = _claimed_run()
+    run.harness_profile_ref = "harness.agent-dev-local@1.0.0"
+    client.claim.return_value = [run]
+    client.heartbeat.side_effect = OpsRunError("lease rejected", status_code=409)
+
+    def slow_profile(*_args, **_kwargs):
+        time.sleep(0.05)
+        raise GlasExecutionError("profile failed")
+
+    with (
+        patch("rein_aharness.claim_loop._heartbeat_interval", return_value=0.01),
+        patch("rein_aharness.claim_loop.execute_profiled_run", side_effect=slow_profile),
+    ):
+        result = process_one(client)
+
+    assert result.ok is False
+    assert result.reason.startswith("lease lost")
+    client.complete.assert_not_called()
+    client.fail.assert_not_called()
+
+
 def test_poll_peek() -> None:
     client = MagicMock(spec=ActivityCoreOpsClient)
     client.list_open.return_value = [_claimed_run()]

tests/test_lease_monitor.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import time
+from unittest.mock import MagicMock, patch
+
+from rein_aharness.claim_loop import _Heartbeat
+from rein_aharness.lease_monitor import LeaseMonitor
+from rein_aharness.ops_run_client import OpsRunError
+
+
+def test_loss_is_one_shot_and_bounded() -> None:
+    seen = []
+    monitor = LeaseMonitor(seen.append)
+
+    first = monitor.mark_lost("ProviderError" * 20)
+    second = monitor.mark_lost("OtherError")
+
+    assert monitor.lost is True
+    assert first == second
+    assert len(first.error_type) <= 80
+    assert first.error_type.startswith("ProviderError")
+    assert seen == [first]
+
+
+def test_wait_lost_and_stop_are_distinct() -> None:
+    monitor = LeaseMonitor()
+
+    assert monitor.wait_lost(timeout=0) is False
+    monitor.stop()
+    assert monitor.stopped is True
+    assert monitor.lost is False
+    monitor.mark_lost("OpsRunError")
+    assert monitor.wait_lost(timeout=0) is True
+
+
+def test_concurrent_loss_still_notifies_once() -> None:
+    seen = []
+    monitor = LeaseMonitor(seen.append)
+
+    import threading
+
+    threads = [threading.Thread(target=monitor.mark_lost, args=("OpsRunError",)) for _ in range(8)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert len(seen) == 1
+
+
+def test_heartbeat_publishes_loss_without_raw_exception_text() -> None:
+    client = MagicMock()
+    client.heartbeat.side_effect = OpsRunError("secret provider response", status_code=409)
+    heartbeat = _Heartbeat(client, "run-1", lease_seconds=90)
+
+    with patch("rein_aharness.claim_loop._heartbeat_interval", return_value=0.01):
+        heartbeat.start()
+        deadline = time.monotonic() + 1.0
+        while not heartbeat.monitor.lost and time.monotonic() < deadline:
+            time.sleep(0.005)
+        heartbeat.stop()
+
+    assert heartbeat.monitor.lost is True
+    assert heartbeat.monitor.evidence is not None
+    assert "secret" not in heartbeat.monitor.evidence.error_type
+
+
+def test_heartbeat_transport_error_does_not_immediately_mark_loss() -> None:
+    client = MagicMock()
+    client.heartbeat.side_effect = OpsRunError("connection reset")
+    heartbeat = _Heartbeat(client, "run-1", lease_seconds=90)
+
+    with patch("rein_aharness.claim_loop._heartbeat_interval", return_value=0.01):
+        heartbeat.start()
+        time.sleep(0.04)
+        heartbeat.stop()
+
+    assert heartbeat.monitor.lost is False

tests/test_ops_run_client.py

@@ -137,6 +137,16 @@ def test_claim_http_error() -> None:
             client.claim()
 
 
+def test_ops_run_error_classifies_lease_rejection_without_message_details() -> None:
+    rejected = OpsRunError("provider response included credentials", status_code=409)
+    transient = OpsRunError("connection reset")
+
+    assert rejected.lease_rejected is True
+    assert rejected.status_code == 409
+    assert transient.lease_rejected is False
+    assert transient.status_code is None
+
+
 def test_ops_run_to_taskspec(tmp_path: Path) -> None:
     import subprocess
 

workplans/HARNESS-WP-0003-governed-runtime-integrity.md

@@ -145,7 +145,7 @@ and T02–T06 can cite stable decisions rather than infer ownership from code.
 
 ```task
 id: HARNESS-WP-0003-T02
-status: wait
+status: progress
 priority: high
 state_hub_task_id: "47816469-b5c0-5252-8b71-041e65d203fd"
 ```
@@ -189,7 +189,27 @@ The primitive is intentionally not wired into `runner.py`, legacy approaches,
 or the claim loop while ADR-002 still awaits Activity Core, sand-boxer, and
 llm-connect acknowledgements. Controlled moved-`HEAD` acceptance is prepared
 under T03 below; lease-loss cancellation, timeout/signal integration, and
-result-close reconciliation remain outstanding. T02 therefore remains `wait`.
+result-close reconciliation remain outstanding; T02 is now `progress`.
+
+### Lease observability slice — 2026-08-23
+
+Added `rein_aharness/lease_monitor.py`, a thread-safe, adapter-neutral lease
+state primitive. It records one bounded loss envelope, provides a waitable loss
+event, and invokes an optional cancellation callback exactly once. The claim
+loop heartbeat now feeds this monitor and emits only bounded error type/time
+evidence instead of raw provider exception text. This establishes the signal
+needed by execution and acceptance boundaries without pretending that adapters
+can already cancel in-flight work. Remaining work is to classify Activity Core
+responses, connect cancellation at each adapter boundary, and refuse result
+acceptance/close after loss. The claim loop now implements that last refusal for
+normal and successful profiled executions: a lost lease returns bounded loss
+evidence and skips Activity Core completion/failure calls, leaving reconciliation
+to the owner of the expired lease. Adapter exception paths now also stop before
+close when lease loss raced the failure; otherwise they return only an exception
+class marker rather than leaving an unbound result or retaining provider text.
+`OpsRunError` now carries bounded action/status metadata, and heartbeats treat
+explicit ownership-rejection statuses as lease loss while allowing transport
+errors to retry without immediately abandoning a run.
 
 ## Verify accepted commits and reconcile metrics/reporting