tegwick
e5396bda51
Some checks failed
Governed runtime contract / contract (push) Has been cancelled
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
483 lines
17 KiB
Python
483 lines
17 KiB
Python
"""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,
|
|
}
|