From 44f5bb9d88a3f633ef85ddc8f43646f649a43c1e Mon Sep 17 00:00:00 2001 From: tegwick Date: Fri, 4 Sep 2026 20:38:01 +0200 Subject: [PATCH] feat(runtime): heartbeat immediately after claim Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a06ba0-10aa-7ea0-b20a-4f3fac39efe9 --- .forgejo/workflows/runtime-contract.yaml | 44 +++++++ Makefile | 5 +- docs/ops-run-claim-loop.md | 16 +++ rein_aharness/claim_loop.py | 24 ++++ scripts/verify-recovery-contracts.sh | 20 ++++ tests/test_claim_loop.py | 113 +++++++++++++++++- ...INAH-WP-0003-governed-runtime-integrity.md | 31 ++++- 7 files changed, 242 insertions(+), 11 deletions(-) create mode 100644 .forgejo/workflows/runtime-contract.yaml create mode 100755 scripts/verify-recovery-contracts.sh diff --git a/.forgejo/workflows/runtime-contract.yaml b/.forgejo/workflows/runtime-contract.yaml new file mode 100644 index 0000000..6fbc0de --- /dev/null +++ b/.forgejo/workflows/runtime-contract.yaml @@ -0,0 +1,44 @@ +name: Governed runtime contract + +on: + push: + branches: + - main + workflow_dispatch: + +jobs: + contract: + runs-on: self-hosted + steps: + # The Railiance host runner does not support actions/checkout. Clone the + # exact pushed revision and the source revisions declared by the lock. + - name: Verify pinned cross-package runtime + env: + FORGE_BASE: ${{ github.server_url }} + REPO_NAME: ${{ github.repository }} + REPO_SHA: ${{ github.sha }} + run: | + set -eu + work_dir="$(mktemp -d /tmp/rein-contract.XXXXXX)" + trap 'rm -rf "${work_dir}"' EXIT + + git clone "${FORGE_BASE}/${REPO_NAME}.git" "${work_dir}/rein-aharness" + git -C "${work_dir}/rein-aharness" checkout --detach "${REPO_SHA}" + + lock="${work_dir}/rein-aharness/deploy/runtime-contract-lock.json" + llm_sha="$(python3 -c 'import json,sys; rows=json.load(open(sys.argv[1]))["dependencies"]; print(next(row["commit"] for row in rows if row["distribution"] == "llm-connect"))' "${lock}")" + glas_sha="$(python3 -c 'import json,sys; rows=json.load(open(sys.argv[1]))["dependencies"]; print(next(row["commit"] for row in rows if row["distribution"] == "glas-harness"))' "${lock}")" + sandbox_sha="$(python3 -c 'import json,sys; rows=json.load(open(sys.argv[1]))["dependencies"]; print(next(row["commit"] for row in rows if row["distribution"] == "sandboxer"))' "${lock}")" + + git clone "${FORGE_BASE}/coulomb/llm-connect.git" "${work_dir}/llm-connect" + git -C "${work_dir}/llm-connect" checkout --detach "${llm_sha}" + git clone "${FORGE_BASE}/coulomb/glas-harness.git" "${work_dir}/glas-harness" + git -C "${work_dir}/glas-harness" checkout --detach "${glas_sha}" + git clone "${FORGE_BASE}/coulomb/sand-boxer.git" "${work_dir}/sand-boxer" + git -C "${work_dir}/sand-boxer" checkout --detach "${sandbox_sha}" + + cd "${work_dir}/rein-aharness" + python3 -m venv .venv + .venv/bin/pip install --disable-pip-version-check 'uv==0.8.0' + .venv/bin/uv sync --frozen --no-editable --extra glas --extra llm --extra dev + make contract-test diff --git a/Makefile b/Makefile index 5af37cc..e59dd15 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: test contract-test image image-export deploy-rsync help +.PHONY: test contract-test recovery-test image image-export deploy-rsync help UV ?= $(shell command -v uv 2>/dev/null || echo uv) IMAGE ?= rein-aharness:railiance01 @@ -14,6 +14,9 @@ test: ## Run unit tests contract-test: ## Run non-skippable cross-package runtime contract tests ./scripts/verify-runtime-contracts.sh +recovery-test: ## Run deterministic lease/crash/close recovery checks + ./scripts/verify-recovery-contracts.sh + image: ## Build container image (vendors ../llm-connect when present) rm -rf llm_connect_vendor mkdir -p llm_connect_vendor diff --git a/docs/ops-run-claim-loop.md b/docs/ops-run-claim-loop.md index d11884d..4976e29 100644 --- a/docs/ops-run-claim-loop.md +++ b/docs/ops-run-claim-loop.md @@ -14,6 +14,7 @@ activity-core emit_tasks ▼ rein-aharness claim-loop POST /ops-runs/claim + POST /ops-runs/{id}/heartbeat (immediate ownership proof; fail closed) harness_profile_ref present? ─ yes → Glas ExecutionRequest → gateway │ no └─→ select_approach(labels, definition, hint) → legacy executor @@ -41,9 +42,11 @@ Profile-absent rows keep the legacy coexistence path, but cannot carry a grant. | `AGENT_HARNESS_OPS_LABELS_MODE` | `any` | `any` or `all` | | `AGENT_HARNESS_OPS_LEASE_SECONDS` | `900` | claim lease | | `AGENT_HARNESS_CLAIM_INTERVAL` | `30` | empty-queue poll interval | +| `AGENT_HARNESS_REQUIRED_PROFILE_REFS` | unset | comma-separated exact production `profile@version` refs checked at startup | | `AGENT_HARNESS_LEGACY_APPROACHES_UNTIL` | unset | ISO expiry date required for profile-absent compatibility routing | | `AGENT_HARNESS_REPO_MAP` | `{}` | JSON slug→checkout path | | `AGENT_HARNESS_REPO_ROOTS` | `~:~/work` | slug search roots | +| `REIN_AHARNESS_STATE_DIR` | user state directory | private repository-lock, metrics, and close-outbox state | Example map on railiance01: @@ -154,6 +157,19 @@ Glas, sand-boxer, and llm-connect before running the contract/claim suites, so the optional `pytest.importorskip` development behavior cannot turn a missing production dependency into a green release result. +Forgejo runs the same gate from +`.forgejo/workflows/runtime-contract.yaml` on every main push. It follows the +estate runner contract (direct clone, no `actions/checkout`), detaches every +sibling at the revision declared in the runtime lock, applies the frozen uv +resolution, and needs no deployment or cluster credential. + +`make recovery-test` is the deterministic pre-deploy recovery gate. It covers +initial and periodic lease rejection, adapter cancellation, repository-lock +release, response-lost close reconciliation, pending-close claim blocking, +every permanent Activity Core close code (including `expired_lease`), and Glas +cancellation. It does not replace the isolated expired-row API smoke or the +natural-run and sandbox-cleanup observations required from the deployed host. + ### Host access to cluster services (no port-forward) On railiance01 (single-node k3s), set **k8s://** pseudo-URLs in diff --git a/rein_aharness/claim_loop.py b/rein_aharness/claim_loop.py index 04a46f4..9616226 100644 --- a/rein_aharness/claim_loop.py +++ b/rein_aharness/claim_loop.py @@ -305,6 +305,30 @@ def process_one( return ProcessResult(claimed=False, empty=True, reason="queue empty") run = claimed[0] + try: + # Make active lease ownership observable even for normal short jobs, + # and fail closed before dispatch if Activity Core no longer accepts it. + client.heartbeat(run.id, lease_seconds=cfg.lease_seconds) + logger.info("accepted initial heartbeat run_id=%s", run.id) + except OpsRunError as exc: + rejected = exc.status_code == 409 + detail_key = "lease_loss" if rejected else "heartbeat_failure" + return ProcessResult( + claimed=True, + run_id=run.id, + ok=False, + retry_full_interval=True, + reason=( + f"initial heartbeat {'rejected' if rejected else 'failed'}: {exc}" + ), + detail={ + detail_key: { + "stage": "pre-dispatch-heartbeat", + "status_code": exc.status_code, + "code": exc.code, + } + }, + ) approach = GLAS_APPROACH if run.harness_profile_ref else select_approach(run) logger.info( "claimed run_id=%s approach=%s title=%r labels=%s", diff --git a/scripts/verify-recovery-contracts.sh b/scripts/verify-recovery-contracts.sh new file mode 100755 index 0000000..75479cf --- /dev/null +++ b/scripts/verify-recovery-contracts.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +# Deterministic recovery gate; live host drills remain separately required. +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "$0")/.." && pwd)" +cd "${REPO_ROOT}" + +PYTHON_BIN="${REIN_CONTRACT_PYTHON:-${REPO_ROOT}/.venv/bin/python}" +"${PYTHON_BIN}" -c 'import glas_harness.contract, llm_connect, sandboxer.models' +PYTHONPATH="${REPO_ROOT}:${REPO_ROOT}/../llm-connect" \ + "${PYTHON_BIN}" -m pytest \ + tests/test_claim_loop.py::test_process_one_initial_heartbeat_rejection_refuses_dispatch \ + tests/test_claim_loop.py::test_process_one_lease_loss_cancels_registered_adapter_process \ + tests/test_claim_loop.py::test_profiled_close_failure_happens_after_repository_lock_release \ + tests/test_claim_loop.py::test_response_lost_close_replays_without_reexecuting_workload \ + tests/test_claim_loop.py::test_pending_close_failure_blocks_new_claim \ + tests/test_claim_loop.py::test_permanent_close_refusal_is_quarantined_before_claim \ + tests/test_repository_transaction.py::test_exception_releases_lock_for_next_transaction \ + tests/test_glas_execution.py::test_profiled_run_refuses_success_if_cancelled_during_gateway \ + -q diff --git a/tests/test_claim_loop.py b/tests/test_claim_loop.py index c592fdf..c3a6a67 100644 --- a/tests/test_claim_loop.py +++ b/tests/test_claim_loop.py @@ -8,6 +8,8 @@ import time from pathlib import Path from unittest.mock import MagicMock, patch +import pytest + from rein_aharness.approaches import ApproachResult, APPROACH_FI_RESEARCH_BRIEF from rein_aharness.claim_loop import ( _cancel_active_run, @@ -118,6 +120,24 @@ def _commit_file(repo: Path, relative: str, value: str = "result\n") -> str: ).stdout.strip() +def _accept_initial_heartbeat_then_reject(run: OpsRun): + calls = 0 + + def heartbeat(*_args, **_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + return run + raise OpsRunError( + "lease rejected", + action="heartbeat", + status_code=409, + code="expired_lease", + ) + + return heartbeat + + def test_process_one_empty() -> None: client = MagicMock(spec=ActivityCoreOpsClient) client.config = OpsRunConfig(worker_id="w", lease_seconds=90) @@ -166,6 +186,67 @@ def test_process_one_success_completes() -> None: client.fail.assert_not_called() +def test_process_one_sends_initial_heartbeat_before_dispatch() -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.config = OpsRunConfig(worker_id="w", lease_seconds=90) + run = _claimed_run() + events: list[str] = [] + client.claim.side_effect = lambda **_kwargs: events.append("claim") or [run] + client.heartbeat.side_effect = ( + lambda *_args, **_kwargs: events.append("heartbeat") or run + ) + client.complete.side_effect = lambda *_args, **_kwargs: ( + events.append("complete") or run + ) + result = ApproachResult( + ok=True, + approach=APPROACH_FI_RESEARCH_BRIEF, + reason="ok", + ) + + def execute(*_args, **_kwargs): + events.append("execute") + return result + + with patch("rein_aharness.claim_loop.execute_approach", side_effect=execute): + processed = process_one(client) + + assert processed.ok is True + assert events == ["claim", "heartbeat", "execute", "complete"] + client.heartbeat.assert_called_once_with(run.id, lease_seconds=90) + + +def test_process_one_initial_heartbeat_rejection_refuses_dispatch() -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.config = OpsRunConfig(worker_id="w", lease_seconds=90) + run = _claimed_run() + client.claim.return_value = [run] + client.heartbeat.side_effect = OpsRunError( + "lease rejected", + action="heartbeat", + status_code=409, + code="expired_lease", + ) + + with patch("rein_aharness.claim_loop.execute_approach") as execute: + result = process_one(client) + + assert result.claimed is True + assert result.ok is False + assert result.retry_full_interval is True + assert result.reason == "initial heartbeat rejected: lease rejected" + assert result.detail == { + "lease_loss": { + "stage": "pre-dispatch-heartbeat", + "status_code": 409, + "code": "expired_lease", + } + } + execute.assert_not_called() + client.complete.assert_not_called() + client.fail.assert_not_called() + + def test_process_one_failure_reopens() -> None: client = MagicMock(spec=ActivityCoreOpsClient) client.config = OpsRunConfig(worker_id="w", lease_seconds=90) @@ -222,7 +303,9 @@ 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) + client.heartbeat.side_effect = _accept_initial_heartbeat_then_reject( + client.claim.return_value[0] + ) ar = ApproachResult( ok=True, approach=APPROACH_FI_RESEARCH_BRIEF, @@ -251,7 +334,9 @@ def test_process_one_lease_loss_cancels_registered_adapter_process() -> 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) + client.heartbeat.side_effect = _accept_initial_heartbeat_then_reject( + client.claim.return_value[0] + ) proc = MagicMock() proc.poll.return_value = None @@ -339,7 +424,9 @@ def test_process_one_records_adapter_exception_without_unbound_result() -> None: def test_profiled_exception_after_lease_loss_skips_close(tmp_path: Path) -> None: repo, _run, client = _profiled_case(tmp_path) - client.heartbeat.side_effect = OpsRunError("lease rejected", status_code=409) + client.heartbeat.side_effect = _accept_initial_heartbeat_then_reject( + client.claim.return_value[0] + ) def slow_profile(*_args, **_kwargs): time.sleep(0.05) @@ -693,14 +780,28 @@ def test_pending_close_failure_blocks_new_claim(tmp_path: Path) -> None: client.claim.assert_not_called() -def test_permanent_close_conflict_is_quarantined_before_claim(tmp_path: Path) -> None: +@pytest.mark.parametrize( + "code", + ( + "not_found", + "wrong_owner", + "expired_lease", + "state_conflict", + "terminal_conflict", + "evidence_conflict", + ), +) +def test_permanent_close_refusal_is_quarantined_before_claim( + tmp_path: Path, + code: str, +) -> None: client = MagicMock(spec=ActivityCoreOpsClient) client.config = OpsRunConfig(worker_id="w", lease_seconds=90) client.complete.side_effect = OpsRunError( "conflict", action="complete", - status_code=409, - code="terminal_conflict", + status_code=404 if code == "not_found" else 409, + code=code, ) client.claim.return_value = [] outbox = CloseOutbox(state_dir=tmp_path / "state") diff --git a/workplans/REINAH-WP-0003-governed-runtime-integrity.md b/workplans/REINAH-WP-0003-governed-runtime-integrity.md index c1cd8c2..3fd5884 100644 --- a/workplans/REINAH-WP-0003-governed-runtime-integrity.md +++ b/workplans/REINAH-WP-0003-governed-runtime-integrity.md @@ -585,10 +585,33 @@ uses the frozen uv lock with `--no-editable`. `make contract-test` first imports all three runtime packages and then runs the cross-package contract/claim suites, making a missing optional dependency a hard release failure. -T05 remains `progress`: the non-skippable gate still needs CI-runner wiring, -the compatible Activity Core artifact and migration need deployment evidence, -and the documented crash/lease/API-close/sandbox cleanup recovery drills remain -to be executed against the pinned host artifact. +Wired the non-skippable gate into Forgejo Actions on the canonical `self-hosted` +runner label. The workflow uses direct Git clones because the estate host runner +does not support `actions/checkout`, detaches each runtime sibling at the locked +revision, and needs no deployment or cluster credential. + +T05 remains `progress`: the compatible Activity Core artifact and migration +need deployment evidence, and the documented crash/lease/API-close/sandbox +cleanup recovery drills remain to be executed against the pinned host artifact. + +### Observable lease acceptance — 2026-09-04 + +Activity Core production evidence showed normal jobs finish before the first +periodic heartbeat interval. The worker now sends one authenticated heartbeat +immediately after claim and before route selection or execution, then retains +the existing periodic heartbeat for longer work. Any initial heartbeat failure +refuses dispatch; a 409 is recorded as pre-dispatch lease loss and neither work +nor terminal close is attempted. Unit coverage proves claim → heartbeat → +execute → close ordering and the 409 fail-closed case. + +The deterministic recovery gate passes 13 lease/cancellation/lock/outbox cases, +the non-skippable cross-package gate passes 51 cases, and the complete suite +passes 255 tests without skips. + +Source is ready for the requested bounded production observation. Revision, +deployment, and the next natural claim → active heartbeat → terminal run id +will be returned to Activity Core after this change is published and deployed; +no real workload will be delayed or expired to manufacture evidence. ## Re-prove one governed profiled run and close residuals