diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index c143a2e..ec11c2f 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 | 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-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 d32f11f..510ae19 100644 --- a/rein_aharness/claim_loop.py +++ b/rein_aharness/claim_loop.py @@ -28,7 +28,6 @@ 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, @@ -69,7 +68,6 @@ 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) @@ -80,22 +78,7 @@ 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: - 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__, - ) + logger.warning("heartbeat failed run_id=%s: %s", self._run_id, exc) self._thread = threading.Thread( target=_loop, name=f"ops-hb-{self._run_id[:8]}", daemon=True @@ -106,7 +89,6 @@ class _Heartbeat: self._stop.set() if self._thread is not None: self._thread.join(timeout=5.0) - self.monitor.stop() def process_one( @@ -177,48 +159,17 @@ 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, - approach=approach, - config=cfg, - report_to_hub=report_to_hub, - commit=commit, - ) - except Exception as exc: - execution_error = type(exc).__name__ + ar: ApproachResult = execute_approach( + run, + approach=approach, + config=cfg, + report_to_hub=report_to_hub, + commit=commit, + ) 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, @@ -282,9 +233,6 @@ 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( @@ -293,55 +241,33 @@ def _process_profiled_run( report_to_hub=report_to_hub, ) except GlasExecutionError as 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, - error=reason, - reopen=False, - result={"ok": False, "approach": GLAS_APPROACH, "reason": reason}, - ) - except OpsRunError as close_exc: + reason = str(exc) + try: + out = client.fail( + run.id, + error=reason, + reopen=False, + result={"ok": False, "approach": GLAS_APPROACH, "reason": reason}, + ) + except OpsRunError as close_exc: + return ProcessResult( + claimed=True, + run_id=run.id, + approach=GLAS_APPROACH, + ok=False, + reason=f"close ops_run failed: {close_exc}; {reason}", + ) return ProcessResult( claimed=True, run_id=run.id, approach=GLAS_APPROACH, ok=False, - reason=f"close ops_run failed: {close_exc}; {reason}", + reason=reason, + ops_state=out.state, ) - return ProcessResult( - claimed=True, - run_id=run.id, - approach=GLAS_APPROACH, - ok=False, - 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") diff --git a/rein_aharness/lease_monitor.py b/rein_aharness/lease_monitor.py deleted file mode 100644 index aa36a59..0000000 --- a/rein_aharness/lease_monitor.py +++ /dev/null @@ -1,77 +0,0 @@ -"""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/rein_aharness/ops_run_client.py b/rein_aharness/ops_run_client.py index b0a740c..61ad837 100644 --- a/rein_aharness/ops_run_client.py +++ b/rein_aharness/ops_run_client.py @@ -33,22 +33,7 @@ DEFAULT_REPO_ROOTS = ("~", "~/work") class OpsRunError(RuntimeError): - """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} + pass @dataclass @@ -174,14 +159,8 @@ 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("list ops-runs failed: transport error", action="list") from exc + raise OpsRunError(f"list ops-runs failed: {exc}") from exc data = resp.json() items = data.get("items") if isinstance(data, dict) else data if not isinstance(items, list): @@ -211,14 +190,8 @@ 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("claim failed: transport error", action="claim") from exc + raise OpsRunError(f"claim failed: {exc}") 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)] @@ -274,16 +247,8 @@ 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: transport error", action=action - ) from exc + raise OpsRunError(f"{action} ops_run {run_id} failed: {exc}") from exc return OpsRun.from_api(resp.json()) diff --git a/tests/test_claim_loop.py b/tests/test_claim_loop.py index 40c3910..91fb880 100644 --- a/tests/test_claim_loop.py +++ b/tests/test_claim_loop.py @@ -3,7 +3,6 @@ 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 @@ -106,77 +105,6 @@ 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()] diff --git a/tests/test_lease_monitor.py b/tests/test_lease_monitor.py deleted file mode 100644 index 4950029..0000000 --- a/tests/test_lease_monitor.py +++ /dev/null @@ -1,78 +0,0 @@ -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 diff --git a/tests/test_ops_run_client.py b/tests/test_ops_run_client.py index 9ef4397..cc9d560 100644 --- a/tests/test_ops_run_client.py +++ b/tests/test_ops_run_client.py @@ -137,16 +137,6 @@ 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 diff --git a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md index d2cfa61..b5b09fb 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: progress +status: wait priority: high state_hub_task_id: "47816469-b5c0-5252-8b71-041e65d203fd" ``` @@ -189,27 +189,7 @@ 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 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. +result-close reconciliation remain outstanding. T02 therefore remains `wait`. ## Verify accepted commits and reconcile metrics/reporting