rein-aharness/scripts/prove-admission-boundaries.py
tegwick e5396bda51
Some checks failed
Governed runtime contract / contract (push) Has been cancelled
Record fixture recovery, rate, and queue-injection evidence.
Admission of the protected runtime still does not deploy or start a paid request. T05 stays wait on the unanswered activity-core lease expiry and a live tariff.

Assistant: grok
Assistant-Session: 01a11b05-37d4-7390-b2df-72297bb7ce1f
2026-10-08 12:28:56 +02:00

106 lines
3.9 KiB
Python

#!/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())