diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index ec11c2f..c143a2e 100644 --- a/WORK-RECORDS.md +++ b/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 | diff --git a/rein_aharness/claim_loop.py b/rein_aharness/claim_loop.py index 510ae19..eefeeed 100644 --- a/rein_aharness/claim_loop.py +++ b/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( diff --git a/rein_aharness/lease_monitor.py b/rein_aharness/lease_monitor.py new file mode 100644 index 0000000..aa36a59 --- /dev/null +++ b/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 + diff --git a/tests/test_lease_monitor.py b/tests/test_lease_monitor.py new file mode 100644 index 0000000..c2f8484 --- /dev/null +++ b/tests/test_lease_monitor.py @@ -0,0 +1,65 @@ +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") + 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 diff --git a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md index b5b09fb..389049f 100644 --- a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md +++ b/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,19 @@ 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. ## Verify accepted commits and reconcile metrics/reporting