From 9097d69e1d5baca235358e1807d52d0b4f85b5d2 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 23 Aug 2026 14:16:11 +0200 Subject: [PATCH 1/4] Expose lease loss to execution boundaries Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d --- WORK-RECORDS.md | 2 +- rein_aharness/claim_loop.py | 11 ++- rein_aharness/lease_monitor.py | 77 +++++++++++++++++++ tests/test_lease_monitor.py | 65 ++++++++++++++++ ...NESS-WP-0003-governed-runtime-integrity.md | 16 +++- 5 files changed, 167 insertions(+), 4 deletions(-) create mode 100644 rein_aharness/lease_monitor.py create mode 100644 tests/test_lease_monitor.py 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 From 75f49147776b24d6472c359a544deb13fb41fd87 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 23 Aug 2026 14:37:18 +0200 Subject: [PATCH 2/4] Refuse close after lease loss Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d --- rein_aharness/claim_loop.py | 34 +++++++++++++++++++ tests/test_claim_loop.py | 30 ++++++++++++++++ ...NESS-WP-0003-governed-runtime-integrity.md | 5 ++- 3 files changed, 68 insertions(+), 1 deletion(-) diff --git a/rein_aharness/claim_loop.py b/rein_aharness/claim_loop.py index eefeeed..0d04921 100644 --- a/rein_aharness/claim_loop.py +++ b/rein_aharness/claim_loop.py @@ -179,6 +179,23 @@ def process_one( 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, + } + }, + ) + payload = { "approach": ar.approach, "ok": ar.ok, @@ -277,6 +294,23 @@ def _process_profiled_run( 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, + } + }, + ) + evidence = gateway_result["evidence"] ok = gateway_result["ok"] reason = str(evidence.get("error") or evidence.get("outcome") or "Glas execution failed") diff --git a/tests/test_claim_loop.py b/tests/test_claim_loop.py index 91fb880..0e1b689 100644 --- a/tests/test_claim_loop.py +++ b/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,35 @@ 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") + 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_poll_peek() -> None: client = MagicMock(spec=ActivityCoreOpsClient) client.list_open.return_value = [_claimed_run()] diff --git a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md index 389049f..fc9ed11 100644 --- a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md +++ b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md @@ -201,7 +201,10 @@ 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. +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. ## Verify accepted commits and reconcile metrics/reporting From ad4074c20a432b4651e74bbcdedb185d0e121253 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 23 Aug 2026 14:40:29 +0200 Subject: [PATCH 3/4] Handle lease loss during adapter errors Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d --- rein_aharness/claim_loop.py | 84 ++++++++++++------- tests/test_claim_loop.py | 42 ++++++++++ ...NESS-WP-0003-governed-runtime-integrity.md | 4 +- 3 files changed, 98 insertions(+), 32 deletions(-) diff --git a/rein_aharness/claim_loop.py b/rein_aharness/claim_loop.py index 0d04921..ace2a33 100644 --- a/rein_aharness/claim_loop.py +++ b/rein_aharness/claim_loop.py @@ -168,14 +168,18 @@ def process_one( hb = _Heartbeat(client, run.id, cfg.lease_seconds) hb.start() + execution_error: str | None = None try: - ar: ApproachResult = execute_approach( - run, - approach=approach, - config=cfg, - report_to_hub=report_to_hub, - commit=commit, - ) + 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__ finally: hb.stop() @@ -196,6 +200,16 @@ def process_one( }, ) + 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, @@ -259,6 +273,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( @@ -267,30 +284,8 @@ def _process_profiled_run( report_to_hub=report_to_hub, ) except GlasExecutionError as 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=reason, - ops_state=out.state, - ) + execution_error = type(exc).__name__ + execution_reason = str(exc) finally: hb.stop() @@ -311,6 +306,33 @@ def _process_profiled_run( }, ) + 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: + 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=reason, + ops_state=out.state, + ) + + 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/tests/test_claim_loop.py b/tests/test_claim_loop.py index 0e1b689..78bf62b 100644 --- a/tests/test_claim_loop.py +++ b/tests/test_claim_loop.py @@ -135,6 +135,48 @@ def test_process_one_refuses_close_after_lease_loss() -> None: 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") + + 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/workplans/HARNESS-WP-0003-governed-runtime-integrity.md b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md index fc9ed11..923e1ac 100644 --- a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md +++ b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md @@ -204,7 +204,9 @@ 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. +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. ## Verify accepted commits and reconcile metrics/reporting From 84b4089f8acfc92241e77e34e6ef4cc2a1f91df7 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 23 Aug 2026 14:42:30 +0200 Subject: [PATCH 4/4] Classify heartbeat lease failures Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d --- rein_aharness/claim_loop.py | 23 +++++++--- rein_aharness/ops_run_client.py | 43 +++++++++++++++++-- tests/test_claim_loop.py | 4 +- tests/test_lease_monitor.py | 15 ++++++- tests/test_ops_run_client.py | 10 +++++ ...NESS-WP-0003-governed-runtime-integrity.md | 3 ++ 6 files changed, 84 insertions(+), 14 deletions(-) diff --git a/rein_aharness/claim_loop.py b/rein_aharness/claim_loop.py index ace2a33..d32f11f 100644 --- a/rein_aharness/claim_loop.py +++ b/rein_aharness/claim_loop.py @@ -80,13 +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: - 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, - ) + 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 diff --git a/rein_aharness/ops_run_client.py b/rein_aharness/ops_run_client.py index 61ad837..b0a740c 100644 --- a/rein_aharness/ops_run_client.py +++ b/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()) diff --git a/tests/test_claim_loop.py b/tests/test_claim_loop.py index 78bf62b..40c3910 100644 --- a/tests/test_claim_loop.py +++ b/tests/test_claim_loop.py @@ -110,7 +110,7 @@ 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") + client.heartbeat.side_effect = OpsRunError("lease rejected", status_code=409) ar = ApproachResult( ok=True, approach=APPROACH_FI_RESEARCH_BRIEF, @@ -159,7 +159,7 @@ def test_profiled_exception_after_lease_loss_skips_close() -> None: 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") + client.heartbeat.side_effect = OpsRunError("lease rejected", status_code=409) def slow_profile(*_args, **_kwargs): time.sleep(0.05) diff --git a/tests/test_lease_monitor.py b/tests/test_lease_monitor.py index c2f8484..4950029 100644 --- a/tests/test_lease_monitor.py +++ b/tests/test_lease_monitor.py @@ -50,7 +50,7 @@ def test_concurrent_loss_still_notifies_once() -> None: def test_heartbeat_publishes_loss_without_raw_exception_text() -> None: client = MagicMock() - client.heartbeat.side_effect = OpsRunError("secret provider response") + 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): @@ -63,3 +63,16 @@ def test_heartbeat_publishes_loss_without_raw_exception_text() -> None: 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 cc9d560..9ef4397 100644 --- a/tests/test_ops_run_client.py +++ b/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 diff --git a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md index 923e1ac..d2cfa61 100644 --- a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md +++ b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md @@ -207,6 +207,9 @@ evidence and skips Activity Core completion/failure calls, leaving reconciliatio 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