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