diff --git a/docs/evidence/2026-10-08-fixture-recovery-rate-queue.json b/docs/evidence/2026-10-08-fixture-recovery-rate-queue.json new file mode 100644 index 0000000..f0ed785 --- /dev/null +++ b/docs/evidence/2026-10-08-fixture-recovery-rate-queue.json @@ -0,0 +1,36 @@ +{ + "schema": "rein-t05-fixture-admission-evidence/v1", + "observed_at": "2026-10-08T10:28:42.242896+00:00", + "workplan": "REINAH-WP-0003", + "task": "REINAH-WP-0003-T05", + "decision": "e642c8e2-ce31-4999-8058-c051b6bdbbbb", + "activity_core_message": "27cd4c47-117c-46a8-a3cf-204f65caa540", + "deployed": false, + "paid_request": false, + "secret_values_read": false, + "live_tariff": false, + "production_queue": false, + "checks": [ + "readiness refuses a missing accepted-rate or queue-injection owner before claim", + "accepted heartbeat lease replaces the claim lease before dispatch", + "fixture upper rates reserve full context liability; a different rate refuses before forward", + "route token is injected; the provider key is not in the workload route", + "lost close keeps evidence, replays once, and leaves no claimed row or held lock", + "killed lock holder releases the repository lock", + "sandbox teardown leaves no orphan sandbox or workspace" + ], + "source_files_sha256": { + "rein_aharness/readiness.py": "9d6945a0ef597c84846e378df8109c88c1d33a4f16c3650da2b110b8501729b1", + "tests/test_admission_boundaries.py": "ddc3fa943c1ed3b09846e4e5ed73a19c8abdeef6a8484f97d0ea7c77b90f74c9", + "tests/test_request_admission.py": "8ab46017c5eac69719b2bceeb63add56700231c79e0302d812cfbccfbf051fec", + "scripts/verify-recovery-contracts.sh": "71b1cf0bd45c0053ddd4d38f8f0eaee950a4ba436ee9de13568a58584e6f4bf2", + "scripts/verify-runtime-contracts.sh": "414475b919e6b101004af6b3d72bfb0062c5d7d11de92e4cc76f72f1107e4455", + ".forgejo/workflows/runtime-contract.yaml": "5e93f6b44b76a47a4c895bc5edc09dd22ea9eb1ea0c3b3ad20f35206a2fc3ac0", + "deploy/runtime-contract-lock.json": "52c464e3d2a9defbb8a6a8f9413d881b5fd2a895ae24ff983370f05333e63c29" + }, + "still_missing": [ + "activity-core has not replied to 27cd4c47; a 409 before dispatch stays claimed because this worker must not terminal-close a lost lease", + "fixture micro-USD rates are not an accepted live tariff or FX", + "the admitted host was not redeployed and no paid request was started" + ] +} diff --git a/docs/ops-run-claim-loop.md b/docs/ops-run-claim-loop.md index 903ca90..86719b9 100644 --- a/docs/ops-run-claim-loop.md +++ b/docs/ops-run-claim-loop.md @@ -166,9 +166,28 @@ 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. +every permanent Activity Core close code (including `expired_lease`), Glas +cancellation, and the fixture admission boundaries in +`tests/test_admission_boundaries.py`. Those boundaries are: readiness fails +before claim when a required accepted-rate or queue-injection owner is +missing; an accepted heartbeat lease replaces the claim lease; a mismatched +fixture rate refuses before any forward; a lost close keeps its evidence and +replays without a second dispatch; a killed lock holder can be reacquired; +and a local bwrap sandbox teardown leaves no workspace. The gate uses the +same `deploy/runtime-contract-lock.json` contract versions as +`make contract-test`. It does not replace the isolated expired-row API smoke, +an accepted live tariff, or a natural-run observation on the admitted host. + +Regenerate the non-secret fixture receipt, without a deploy or a paid call: + +```bash +.venv/bin/python scripts/prove-admission-boundaries.py +``` + +Receipt: `docs/evidence/2026-10-08-fixture-recovery-rate-queue.json`. +A past or missing accepted lease does not dispatch. A 409 from Activity Core +still leaves that row claimed; this worker does not terminal-close a lease it +no longer holds. Message `27cd4c47-117c-46a8-a3cf-204f65caa540` has no reply. ### Installed-runtime recovery drill diff --git a/docs/request-admission.md b/docs/request-admission.md index a350d63..58b0b38 100644 --- a/docs/request-admission.md +++ b/docs/request-admission.md @@ -42,6 +42,17 @@ calls; with adequate fixture capacity a two-request tool session succeeds. That verifies request admission and accounting, not a live EUR ceiling or owner/workload secret and network separation. +When `AGENT_HARNESS_REQUIRE_REQUEST_ADMISSION` is set, preflight also refuses +before any claim unless the in-process owner carries positive integer +input and output micro-USD upper rates and an `activate` path that can bind +an accepted queue lease to a route token. Those rates are the fixture or +owner pin already on `MessagesPolicy`. They are not a live tariff or FX. +The 2026-10-08 fixture receipt is +`docs/evidence/2026-10-08-fixture-recovery-rate-queue.json`. It does not +deploy the admitted runtime or start a paid request. Activity Core still +owes a reply to message `27cd4c47-117c-46a8-a3cf-204f65caa540`: a heartbeat +409 stays claimed because this worker must not terminal-close a lost lease. + Owner lifecycle and sole-route integration are now documented in [owner-messages-route.md](owner-messages-route.md). This is local source/fixture evidence; live bootstrap and placement acceptance remain open. diff --git a/rein_aharness/readiness.py b/rein_aharness/readiness.py index 36bf3c0..d70e9e3 100644 --- a/rein_aharness/readiness.py +++ b/rein_aharness/readiness.py @@ -77,6 +77,7 @@ def run_readiness_checks( checks.extend(_repository_checks(client, env)) checks.extend(_profile_checks(env)) + checks.extend(_request_admission_checks(client)) if check_activity_core: try: @@ -205,6 +206,50 @@ def _profile_checks(env: Mapping[str, str]) -> list[ReadinessCheck]: return checks +def _request_admission_checks(client: ActivityCoreOpsClient) -> list[ReadinessCheck]: + """Fail closed before claim when a metered route has no accepted boundary. + + Fixture upper rates are not a live tariff. This check never reads a + credential and never calls the queue claim endpoint. + """ + config = client.config + if not config.require_request_admission: + return [] + owner = config.messages_owner + policy = getattr(owner, "policy", None) + input_rate = getattr(policy, "input_microusd_per_token", None) + output_rate = getattr(policy, "output_microusd_per_token", None) + tariff = getattr(policy, "tariff_ref", None) + digest = getattr(policy, "sha256", None) + rates_ok = ( + type(input_rate) is int + and type(output_rate) is int + and input_rate > 0 + and output_rate > 0 + and isinstance(tariff, str) + and bool(tariff) + and isinstance(digest, str) + and len(digest) == 64 + ) + injection_ok = owner is not None and callable(getattr(owner, "activate", None)) + return [ + ReadinessCheck( + "accepted_rates", + rates_ok, + "accepted upper rates pinned" + if rates_ok + else "accepted rate boundary unavailable", + ), + ReadinessCheck( + "queue_injection", + injection_ok, + "lease-to-token owner present" + if injection_ok + else "queue lease injection owner unavailable", + ), + ] + + def _probe_bwrap(env: Mapping[str, str]) -> ReadinessCheck: configured = env.get("SANDBOXER_BWRAP_BIN", "").strip() executable = configured or shutil.which("bwrap") diff --git a/scripts/prove-admission-boundaries.py b/scripts/prove-admission-boundaries.py new file mode 100644 index 0000000..d43ab46 --- /dev/null +++ b/scripts/prove-admission-boundaries.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +"""Record fixture recovery, rate, and queue-injection evidence. + +This does not deploy, claim a production row, read a secret, or call a provider. +""" + +from __future__ import annotations + +import hashlib +import json +import sys +import tempfile +from datetime import UTC, datetime +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path[:0] = [str(ROOT), str(ROOT / "tests")] + +from test_admission_boundaries import ( # noqa: E402 + test_queue_injection_uses_accepted_lease_and_fixture_rates, + test_readiness_fails_before_claim_when_rate_or_owner_is_missing, + test_recovery_drills_release_lock_claim_sandbox_and_evidence, +) +from test_request_admission import open_fake_provider # noqa: E402 + + +def _run_test(function, root: Path, *, provider: bool = False, monkey: bool = False) -> None: + root.mkdir(parents=True) + if provider: + with open_fake_provider() as started: + function(root, started, _Monkey()) + return + if monkey: + function(root, _Monkey()) + return + function(root) + + +class _Monkey: + def setenv(self, name: str, value: str) -> None: + import os + + os.environ[name] = value + + +def main() -> int: + output = ROOT / "docs/evidence/2026-10-08-fixture-recovery-rate-queue.json" + with tempfile.TemporaryDirectory(prefix="admission-boundaries-") as raw: + root = Path(raw) + _run_test(test_readiness_fails_before_claim_when_rate_or_owner_is_missing, root / "ready") + _run_test( + test_queue_injection_uses_accepted_lease_and_fixture_rates, + root / "queue", + provider=True, + ) + _run_test( + test_recovery_drills_release_lock_claim_sandbox_and_evidence, + root / "recover", + monkey=True, + ) + sources = [ + "rein_aharness/readiness.py", + "tests/test_admission_boundaries.py", + "tests/test_request_admission.py", + "scripts/verify-recovery-contracts.sh", + "scripts/verify-runtime-contracts.sh", + ".forgejo/workflows/runtime-contract.yaml", + "deploy/runtime-contract-lock.json", + ] + receipt = { + "schema": "rein-t05-fixture-admission-evidence/v1", + "observed_at": datetime.now(UTC).isoformat(), + "workplan": "REINAH-WP-0003", + "task": "REINAH-WP-0003-T05", + "decision": "e642c8e2-ce31-4999-8058-c051b6bdbbbb", + "activity_core_message": "27cd4c47-117c-46a8-a3cf-204f65caa540", + "deployed": False, + "paid_request": False, + "secret_values_read": False, + "live_tariff": False, + "production_queue": False, + "checks": [ + "readiness refuses a missing accepted-rate or queue-injection owner before claim", + "accepted heartbeat lease replaces the claim lease before dispatch", + "fixture upper rates reserve full context liability; a different rate refuses before forward", + "route token is injected; the provider key is not in the workload route", + "lost close keeps evidence, replays once, and leaves no claimed row or held lock", + "killed lock holder releases the repository lock", + "sandbox teardown leaves no orphan sandbox or workspace", + ], + "source_files_sha256": { + name: hashlib.sha256((ROOT / name).read_bytes()).hexdigest() for name in sources + }, + "still_missing": [ + "activity-core has not replied to 27cd4c47; a 409 before dispatch stays claimed because this worker must not terminal-close a lost lease", + "fixture micro-USD rates are not an accepted live tariff or FX", + "the admitted host was not redeployed and no paid request was started", + ], + } + output.write_text(json.dumps(receipt, indent=2) + "\n") + print(json.dumps({"ok": True, "receipt": str(output.relative_to(ROOT))})) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/verify-recovery-contracts.sh b/scripts/verify-recovery-contracts.sh index 3a0378e..a2db2ac 100755 --- a/scripts/verify-recovery-contracts.sh +++ b/scripts/verify-recovery-contracts.sh @@ -14,6 +14,7 @@ PYTHONPATH="${REPO_ROOT}:${REPO_ROOT}/../llm-connect" \ tests/test_request_admission.py \ tests/test_messages_owner.py \ tests/test_owner_bootstrap.py \ + tests/test_admission_boundaries.py \ 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 \ diff --git a/tests/test_admission_boundaries.py b/tests/test_admission_boundaries.py new file mode 100644 index 0000000..1dcf7a2 --- /dev/null +++ b/tests/test_admission_boundaries.py @@ -0,0 +1,483 @@ +"""Fixture recovery, rate, and queue-injection evidence. + +No deploy, no paid provider request, and no secret values. Fixture micro-USD +rates are not an accepted live tariff. A 409 before dispatch leaves the queue +row claimed because this worker must not terminal-close a lost lease. +""" + +from __future__ import annotations + +import json +import os +import selectors +import shutil +import socket +import subprocess +import sys +from dataclasses import replace +from datetime import UTC, datetime, timedelta +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from llm_connect.messages_gate import MessagesPolicy + +from rein_aharness.close_outbox import CloseOutbox +from rein_aharness.execution_cancel import ExecutionCancel +from rein_aharness.messages_owner import MessagesOwner, accepted_lease +from rein_aharness.ops_run_client import OpsRun, OpsRunConfig +from rein_aharness.readiness import run_readiness_checks +from rein_aharness.repository_grant import RepositoryGrant +from rein_aharness.repository_transaction import RepositoryBusyError, RepositoryTransaction +from rein_aharness.request_admission import RequestLedger +from rein_aharness.spend_admission import SpendAdmissionError, SpendLedger, SpendPolicy +from test_request_admission import fake_provider as fake_provider + + +ROOT = Path(__file__).resolve().parents[1] +PROVIDER_KEY = "fixture-owner-key" + + +def _git(repo: Path, *args: str) -> str: + return subprocess.check_output(["git", "-C", str(repo), *args], text=True).strip() + + +def _repository(path: Path) -> None: + path.mkdir() + _git(path, "init", "-q") + (path / "README.md").write_text("Disposable admission fixture.\n") + _git(path, "add", "README.md") + _git( + path, + "-c", + "user.name=Fixture", + "-c", + "user.email=fixture@example.invalid", + "commit", + "-qm", + "baseline", + ) + + +def _policy_file(tmp: Path, repo: Path) -> tuple[SpendPolicy, Path, Path]: + private = tmp / "private" + private.mkdir(mode=0o700) + policy = SpendPolicy( + version="1", + envelope_id="fixture-envelope", + authority_ref="fixture:no-live-authority", + valid_from="2026-09-01T00:00:00Z", + expires_at="2099-01-01T00:00:00Z", + timezone="Europe/Berlin", + worker_id="fixture-worker", + activity_definition_id="def-fixture", + target_repo=str(repo.resolve()), + project="fixture-factory", + profile_ref="harness.fixture@1.0.0", + profile_sha256="a" * 64, + descriptor_sha256="b" * 64, + repository_grant_id=RepositoryGrant("1", ("docs/",), 1, 1, False).grant_id, + max_budget_usd="4", + max_liability_usd="5", + max_turns=4, + eur_per_usd="1", + per_run_eur="5", + daily_eur="10", + total_eur="15", + ) + path = private / "policy.json" + path.write_text(json.dumps(policy.__dict__)) + path.chmod(0o600) + ledger_path = private / "spend.sqlite3" + ledger = SpendLedger(ledger_path, policy) + ledger.initialize() + return policy, path, ledger_path + + +def _run(repo: Path) -> OpsRun: + return OpsRun( + id="run-fixture", + activity_definition_id="def-fixture", + idempotency_key="fixture-once", + target_repo=str(repo), + title="fixture", + description="", + state="open", + attempt=1, + harness_profile_ref="harness.fixture@1.0.0", + ) + + +class _Queue: + """Local claim double. It never opens an Activity Core connection.""" + + def __init__(self, run: OpsRun, config: OpsRunConfig) -> None: + self.run = run + self.config = config + self.claims = 0 + self.dispatches = 0 + self.claim_lease: str | None = None + self.injected_lease: str | None = None + self.reject_lease = False + self.complete_failures = 0 + + def claim(self, limit: int = 1) -> list[OpsRun]: + assert limit == 1 + self.claims += 1 + if self.claims != 1 or self.run.state != "open": + return [] + self.run.state = "claimed" + self.run.claim_owner = self.config.worker_id + self.claim_lease = (datetime.now(UTC) + timedelta(minutes=5)).isoformat() + self.run.lease_until = self.claim_lease + return [self.run] + + def heartbeat(self, run_id: str, lease_seconds: int | None = None) -> OpsRun: + assert run_id == self.run.id and self.run.state == "claimed" + if self.reject_lease: + self.run.lease_until = "2020-01-01T00:00:00+00:00" + return replace(self.run, lease_until=self.run.lease_until) + self.injected_lease = (datetime.now(UTC) + timedelta(seconds=45)).isoformat() + self.run.lease_until = self.injected_lease + return replace(self.run, lease_until=self.injected_lease) + + def list_open(self, limit: int = 50) -> list[OpsRun]: + return [] + + def complete(self, run_id: str, *, result: dict) -> OpsRun: + assert run_id == self.run.id and self.run.state == "claimed" + if self.complete_failures: + self.complete_failures -= 1 + raise OpsRunError("fixture response lost after apply", action="complete") + self.run.state = "succeeded" + return replace(self.run, state="succeeded", close_disposition="applied") + + def fail(self, run_id: str, *, error: str, reopen: bool, result: dict) -> OpsRun: + assert run_id == self.run.id and self.run.state == "claimed" + self.run.state = "open" if reopen else "failed" + return replace(self.run, state=self.run.state) + + +class _Owner: + def __init__(self, policy: MessagesPolicy) -> None: + self.policy = policy + + def activate(self, *args, **kwargs): + raise AssertionError("patched dispatch must not activate the owner") + + +def _config(spend: SpendPolicy, policy_path: Path, ledger_path: Path, owner) -> OpsRunConfig: + return OpsRunConfig( + worker_id=spend.worker_id, + execution_project=spend.project, + repo_roots=(), + require_request_admission=True, + require_spend_admission=True, + spend_policy_path=str(policy_path), + spend_ledger_path=str(ledger_path), + messages_owner=owner, + ) + + +def test_readiness_fails_before_claim_when_rate_or_owner_is_missing(tmp_path: Path) -> None: + repo = tmp_path / "repo" + _repository(repo) + spend, policy_path, ledger_path = _policy_file(tmp_path, repo) + calls = {"claim": 0} + + class _Client: + def __init__(self, owner) -> None: + self.config = _config(spend, policy_path, ledger_path, owner) + + def list_open(self, *, limit: int = 50): + assert limit == 1 + return [] + + def claim(self, limit: int = 1): + calls["claim"] += 1 + raise AssertionError("readiness claimed work") + + missing = run_readiness_checks( + _Client(None), # type: ignore[arg-type] + CloseOutbox(state_dir=tmp_path / "state"), + environ={}, + check_activity_core=False, + ) + assert not missing.ok + failed = {check.name for check in missing.checks if not check.ok} + assert failed == {"accepted_rates", "queue_injection"} + + rates = MessagesPolicy("fixture:not-live-tariff", "fixture-model", 1000, 1000, 1000, 1000) + present = run_readiness_checks( + _Client(_Owner(rates)), # type: ignore[arg-type] + CloseOutbox(state_dir=tmp_path / "state-ok"), + environ={}, + check_activity_core=False, + ) + assert present.ok + assert calls["claim"] == 0 + + +def test_queue_injection_uses_accepted_lease_and_fixture_rates( + tmp_path: Path, fake_provider, monkeypatch +) -> None: + monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "xdg")) + repo = tmp_path / "repo" + _repository(repo) + spend, policy_path, ledger_path = _policy_file(tmp_path, repo) + rates = MessagesPolicy("fixture:not-live-tariff", "fixture-model", 1000, 1000, 1000, 1000) + other = MessagesPolicy("fixture:not-live-tariff", "fixture-model", 1000, 1000, 1001, 1000) + assert other.sha256 != rates.sha256 + queue = _Queue(_run(repo), _config(spend, policy_path, ledger_path, _Owner(rates))) + from rein_aharness.claim_loop import process_one + + seen_lease: dict[str, str] = {} + + def execute(run, config, **kwargs): + queue.dispatches += 1 + seen_lease["value"] = run.lease_until + assert config.messages_owner.policy.sha256 == rates.sha256 + return {"ok": False, "evidence": {"outcome": "failed", "error": "fixture stop"}} + + with patch("rein_aharness.claim_loop.execute_profiled_run", side_effect=execute): + result = process_one(queue, outbox=CloseOutbox(state_dir=tmp_path / "close")) + + assert result.ok is False + assert queue.dispatches == 1 + assert seen_lease["value"] == queue.injected_lease + assert seen_lease["value"] != queue.claim_lease + assert queue.run.state == "failed" + assert "fixture-owner-key" not in json.dumps(result.detail) + + refused = _Queue(_run(tmp_path / "unused"), queue.config) + refused.run = _run(repo) + refused.reject_lease = True + with patch( + "rein_aharness.claim_loop.execute_profiled_run", + side_effect=AssertionError("dispatch after rejected lease"), + ): + rejected = process_one(refused, outbox=CloseOutbox(state_dir=tmp_path / "rejected")) + assert "accepted queue lease" in rejected.reason + assert refused.dispatches == 0 + assert refused.run.state == "claimed" + + provider, seen, _mode = fake_provider + ledger = SpendLedger(ledger_path, spend) + RequestLedger(ledger).initialize() + item = replace( + _run(repo), + id="run-route", + idempotency_key="route-once", + state="claimed", + claim_owner=spend.worker_id, + lease_until=(datetime.now(UTC) + timedelta(seconds=60)).isoformat(), + ) + ledger.reserve(item) + owner = MessagesOwner( + rates, + PROVIDER_KEY, + upstream_url=f"http://127.0.0.1:{provider.server_port}", + allow_test_http=True, + ) + profile = SimpleNamespace( + credential_route_refs=[], + model=SimpleNamespace(model=rates.model), + sandbox_profile="profile.bwrap-local", + ) + with owner.activate(item, queue.config, ledger, profile, ExecutionCancel()) as manager: + route = manager._messages_route + assert route.token != PROVIDER_KEY + assert PROVIDER_KEY not in repr(route) + expiry = accepted_lease(item, item, spend.worker_id) + assert expiry == item.lease_until + meter = RequestLedger(ledger) + with pytest_raises(SpendAdmissionError): + meter.reserve_request(route.token, other.sha256, 1) + assert seen == [] + status, body = _unix_messages(route.socket_path, route.token, rates) + assert status == 200 and b"message_stop" in body + assert len(seen) == 1 and seen[0]["key"] == PROVIDER_KEY + assert seen[0]["key"] != route.token + assert meter.status()[0]["liability_microusd"] == ( + rates.context_tokens * rates.input_microusd_per_token + + 1000 * rates.output_microusd_per_token + ) + route_token = route.token + socket_path = route.socket_path + assert not socket_path.exists() + with pytest_raises(SpendAdmissionError): + RequestLedger(ledger).reserve_request(route_token, rates.sha256, 1) + assert len(seen) == 1 + assert PROVIDER_KEY not in json.dumps(RequestLedger(ledger).status()) + + +def test_recovery_drills_release_lock_claim_sandbox_and_evidence( + tmp_path: Path, monkeypatch +) -> None: + monkeypatch.setenv("SANDBOXER_NO_STATE_HUB", "1") + monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "xdg")) + repo = tmp_path / "repo" + _repository(repo) + spend, policy_path, ledger_path = _policy_file(tmp_path, repo) + rates = MessagesPolicy("fixture:not-live-tariff", "fixture-model", 1000, 1000, 1000, 1000) + queue = _Queue(_run(repo), _config(spend, policy_path, ledger_path, _Owner(rates))) + queue.complete_failures = 1 + outbox = CloseOutbox(state_dir=tmp_path / "close") + from rein_aharness.claim_loop import process_one + + def execute(run, config, **kwargs): + queue.dispatches += 1 + return { + "ok": True, + "evidence": {"outcome": "succeeded", "error": None}, + } + + with patch("rein_aharness.claim_loop.execute_profiled_run", side_effect=execute): + first = process_one(queue, outbox=outbox, report_to_hub=False) + assert first.reason == "close evidence remains pending" + assert outbox.status()["pending"] == 1 + assert queue.dispatches == 1 + with RepositoryTransaction(repo) as locked: + assert locked.locked + queue.run.state = "claimed" + with patch( + "rein_aharness.claim_loop.execute_profiled_run", + side_effect=AssertionError("replay dispatched work"), + ): + second = process_one(queue, outbox=outbox, report_to_hub=False) + assert second.empty + assert queue.dispatches == 1 + assert queue.run.state == "succeeded" + assert outbox.status() == {"pending": 0, "delivered": 1, "quarantined": 0} + + lock = _killed_lock(tmp_path / "lock-repo") + sandbox = _sandbox_teardown(tmp_path / "sandbox") + assert lock["lock_reacquired"] and not lock["held_lock"] + assert sandbox["sandbox_destroyed"] and sandbox["workspace_removed"] + assert sandbox["orphan_sandbox"] is False + + workflow = (ROOT / ".forgejo/workflows/runtime-contract.yaml").read_text() + lockfile = json.loads((ROOT / "deploy/runtime-contract-lock.json").read_text()) + assert "runtime-contract-lock.json" in workflow + assert "make contract-test recovery-test" in workflow + assert lockfile["contracts"]["activity_core_schema"] == "0010" + assert lockfile["contracts"]["glas_contract_version"] == "1.0" + + +def pytest_raises(error): + import pytest + + return pytest.raises(error) + + +def _unix_messages(path: Path, token: str, policy: MessagesPolicy) -> tuple[int, bytes]: + body = json.dumps( + { + "model": policy.model, + "max_tokens": policy.max_output_tokens, + "stream": True, + "messages": [{"role": "user", "content": "synthetic fixture"}], + } + ).encode() + request = ( + "POST /v1/messages HTTP/1.1\r\n" + "Host: localhost\r\n" + "Content-Type: application/json\r\n" + f"Content-Length: {len(body)}\r\n" + f"x-api-key: {token}\r\n" + "Connection: close\r\n\r\n" + ).encode() + body + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: + client.settimeout(5) + client.connect(str(path)) + client.sendall(request) + chunks = [] + while True: + chunk = client.recv(4096) + if not chunk: + break + chunks.append(chunk) + raw = b"".join(chunks) + status = int(raw.split(b" ", 2)[1]) + return status, raw + + +def _killed_lock(repo: Path) -> dict: + _repository(repo) + child = subprocess.Popen( + [ + sys.executable, + "-c", + "import sys,time\n" + "from pathlib import Path\n" + "from rein_aharness.repository_transaction import RepositoryTransaction\n" + "with RepositoryTransaction(Path(sys.argv[1])):\n" + " print('locked', flush=True)\n" + " time.sleep(30)\n", + str(repo), + ], + stdout=subprocess.PIPE, + text=True, + ) + try: + assert child.stdout is not None + with selectors.DefaultSelector() as selector: + selector.register(child.stdout, selectors.EVENT_READ) + assert selector.select(10), "lock child did not start" + assert child.stdout.readline().strip() == "locked" + try: + with RepositoryTransaction(repo): + raise AssertionError("second process acquired a held lock") + except RepositoryBusyError: + pass + child.kill() + child.wait(timeout=5) + with RepositoryTransaction(repo) as recovered: + assert recovered.locked + return {"held_lock": False, "lock_reacquired": True} + finally: + if child.poll() is None: + child.kill() + child.wait(timeout=5) + + +def _sandbox_teardown(tmp: Path) -> dict: + tmp.mkdir() + os.environ["SANDBOXER_NO_STATE_HUB"] = "1" + os.environ["XDG_DATA_HOME"] = str(tmp / "xdg") + executable = shutil.which("bwrap") + assert executable, "bubblewrap boundary unavailable" + from sandboxer.core.manager import SandboxManager + from sandboxer.models import Consumer, SandboxCreateRequest, SandboxExecRequest + + source = tmp / "source" + source.mkdir() + (source / "README.md").write_text("sandbox fixture\n") + consumer = Consumer(actor="agt", project="fixture-factory", run_id="sandbox-drill") + manager = SandboxManager() + status = manager.create( + SandboxCreateRequest( + profile="profile.bwrap-local", + inputs={"repo": str(source)}, + consumer=consumer, + ) + ) + workspace = Path(status.inputs["workspace_dir"]) + try: + assert status.state.value == "ready", status.error + assert workspace.exists() + ran = manager.execute( + status.sandbox_id, + SandboxExecRequest(command=["/bin/true"], consumer=consumer, timeout_seconds=10), + ) + assert ran.exit_code == 0, ran.stderr + finally: + destroyed = manager.destroy(status.sandbox_id) + assert destroyed.state.value == "destroyed" + assert not workspace.exists() + assert [item.state.value for item in manager.list()] == ["destroyed"] + return { + "sandbox_destroyed": True, + "workspace_removed": True, + "orphan_sandbox": False, + } diff --git a/tests/test_request_admission.py b/tests/test_request_admission.py index 783c38c..c0da3ce 100644 --- a/tests/test_request_admission.py +++ b/tests/test_request_admission.py @@ -5,6 +5,7 @@ import http.server import json import threading from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager from datetime import UTC, datetime, timedelta import pytest @@ -155,8 +156,9 @@ def stream_bytes(*, truncated=False, output=10): ).encode() -@pytest.fixture -def fake_provider(): +@contextmanager +def open_fake_provider(): + """Loopback fixture upstream. Not a provider and not a paid call.""" seen = [] mode = {"status": 200, "truncated": False, "output": 10} @@ -186,9 +188,17 @@ def fake_provider(): server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) threading.Thread(target=server.serve_forever, daemon=True).start() - yield server, seen, mode - server.shutdown() - server.server_close() + try: + yield server, seen, mode + finally: + server.shutdown() + server.server_close() + + +@pytest.fixture +def fake_provider(): + with open_fake_provider() as started: + yield started @pytest.fixture diff --git a/workplans/REINAH-WP-0003-governed-runtime-integrity.md b/workplans/REINAH-WP-0003-governed-runtime-integrity.md index 1be0bcd..51dd518 100644 --- a/workplans/REINAH-WP-0003-governed-runtime-integrity.md +++ b/workplans/REINAH-WP-0003-governed-runtime-integrity.md @@ -10,7 +10,7 @@ owner: codex topic_slug: rein-aharness priority: high created: "2026-08-23" -updated: "2026-10-05" +updated: "2026-10-08" related: - REIN-A-0004 - GLAS-IN-0002 @@ -521,7 +521,7 @@ Freedom Intelligence `d0b45acb-0002-405b-9620-cc44568170fb`, Binky id: REINAH-WP-0003-T05 status: wait blocked_on: message-from:activity-core -blocking_reason: "Founder admitted the protected runtime on 2026-10-05 (e642c8e2). Recovery-drill, rate, and queue-injection evidence are still absent. Admission does not deploy or start a paid request." +blocking_reason: "Fixture recovery, rate, and queue-injection evidence is local only (2026-10-08). Activity-core has not replied to 27cd4c47: a 409 stays claimed because this worker must not terminal-close a lost lease. Fixture micro-USD rates are not a live tariff. Admission does not deploy or start a paid request." priority: high state_hub_task_id: "c4a3f08f-2874-5672-9f29-aecddd697d90" ``` @@ -732,6 +732,22 @@ queue-injection evidence are still absent. The admission does not deploy or start a paid request. Activity-core message `27cd4c47-117c-46a8-a3cf-204f65caa540` asks for that evidence. +2026-10-08: local fixtures now cover the three missing evidence classes +without a deploy, a secret read, or a paid request. Preflight fails before +claim when request admission is required and the accepted-rate pin or the +lease-to-token owner is missing. A fixture queue injects only an accepted +heartbeat lease; a past lease does not dispatch. The owner injects the route +token, not the provider key, and a different micro-USD rate refuses before +any forward. Recovery covers a lost close (evidence kept, one replay, no +second dispatch, row not left claimed), a killed repository-lock holder, and +a local bwrap sandbox teardown with no leftover workspace. The receipt is +`docs/evidence/2026-10-08-fixture-recovery-rate-queue.json`. The same +runtime-lock contract versions are what `make contract-test` and +`make recovery-test` run. T05 stays `wait`. Activity-core has not replied to +`27cd4c47`; a 409 before dispatch stays claimed because this worker must not +terminal-close a lost lease. Fixture rates are not a live tariff or FX. +T06 stays `wait`. + ## Re-prove one governed profiled run and close residuals ```task