rein-aharness/tests/test_spend_admission.py

"""No-provider proofs for admission, persistence, accounting and recovery."""

import json
from concurrent.futures import ProcessPoolExecutor
from dataclasses import replace
from datetime import UTC, datetime, timedelta
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest

from rein_aharness.ops_run_client import ActivityCoreOpsClient, OpsRun, OpsRunConfig
from rein_aharness.repository_grant import RepositoryGrant
from rein_aharness.spend_admission import (
    SpendAdmissionError,
    SpendLedger,
    SpendPolicy,
    digest,
    worker_spend,
)

NOW = datetime(2026, 9, 9, 12, tzinfo=UTC)


def run(key="one"):
    return OpsRun(
        id=f"run-{key}",
        activity_definition_id="def-1",
        idempotency_key=key,
        target_repo="target",
        title="private intent",
        description="private prompt",
        attempt=1,
        claim_owner="worker-1",
        harness_profile_ref="harness.test@1.0.0",
        repository_grant=RepositoryGrant("1", ("docs/",), 1, 1, False),
    )


@pytest.fixture
def ledger(tmp_path):
    private = tmp_path / "private"
    private.mkdir(mode=0o700)
    policy = SpendPolicy(
        version="1",
        envelope_id="test-only",
        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="worker-1",
        activity_definition_id="def-1",
        target_repo=str(tmp_path / "target"),
        project="fixture-factory",
        profile_ref="harness.test@1.0.0",
        profile_sha256="a" * 64,
        descriptor_sha256="b" * 64,
        repository_grant_id=run().repository_grant.grant_id,
        max_budget_usd="4",
        max_liability_usd="5",
        max_turns=8,
        eur_per_usd="1",
        per_run_eur="5",
        daily_eur="10",
        total_eur="15",
    )
    store = SpendLedger(private / "spend.sqlite3", policy)
    store.initialize()
    return store


def success(store, item, cost=1):
    return {
        "ok": True,
        "evidence": {
            "request_id": item.id,
            "profile_ref": store.policy.profile_ref,
            "cost_usd": cost,
            "outcome": "succeeded",
            "session_cleanup": "succeeded",
            "sandbox_destroy": "succeeded",
        },
    }


def configured(store):
    path = store.path.parent / "policy.json"
    path.write_text(json.dumps(store.policy.__dict__))
    path.chmod(0o600)
    return OpsRunConfig(
        worker_id="worker-1",
        execution_project="fixture-factory",
        spend_policy_path=str(path),
        spend_ledger_path=str(store.path),
        repo_roots=(str(Path(store.policy.target_repo).parent),),
    )


def test_daily_total_and_no_self_reported_refund(ledger):
    for key in ("one", "two"):
        item = run(key)
        ledger.reserve(item, now=NOW)
        ledger.observe(item.id, success(ledger, item, 0), now=NOW)
    with pytest.raises(SpendAdmissionError, match="daily"):
        ledger.reserve(run("three"), now=NOW)
    tomorrow = NOW + timedelta(days=1)
    ledger.reserve(run("three"), now=tomorrow)
    ledger.observe(run("three").id, success(ledger, run("three")), now=tomorrow)
    with pytest.raises(SpendAdmissionError, match="total"):
        ledger.reserve(run("four"), now=tomorrow + timedelta(days=1))
    assert sum(r["liability"] for r in ledger.status()["reservations"]) == 15_000_000


def test_crash_survives_reopen_and_blocks_across_days(ledger):
    ledger.reserve(run(), now=NOW)
    reopened = SpendLedger(ledger.path, ledger.policy)
    for day in (NOW, NOW + timedelta(days=5)):
        with pytest.raises(SpendAdmissionError, match="unresolved"):
            reopened.preflight(now=day)
    reopened.reconcile(
        run().id,
        cost_usd="2",
        receipt="audit:stopped-and-final",
        now=NOW + timedelta(days=5),
    )
    reopened.reconcile(
        run().id,
        cost_usd="2",
        receipt="audit:stopped-and-final",
        now=NOW + timedelta(days=5),
    )
    with pytest.raises(SpendAdmissionError, match="replay"):
        reopened.reserve(replace(run(), attempt=2), now=NOW + timedelta(days=5))
    with pytest.raises(SpendAdmissionError, match="replay"):
        reopened.reserve(
            replace(run(), id="different-row"), now=NOW + timedelta(days=5)
        )
    assert reopened.status()["reservations"][0]["liability"] == 5_000_000


def test_full_charge_on_both_sides_of_midnight(ledger):
    before = datetime(2026, 9, 9, 21, 59, tzinfo=UTC)
    after = before + timedelta(minutes=2)
    ledger.reserve(run(), now=before)
    ledger.observe(run().id, success(ledger, run()), now=after)
    row = ledger.status()["reservations"][0]
    assert (row["start_day"], row["end_day"]) == ("2026-09-09", "2026-09-10")
    ledger.reserve(run("two"), now=after)
    ledger.observe(run("two").id, success(ledger, run("two")), now=after)
    with pytest.raises(SpendAdmissionError, match="daily"):
        ledger.preflight(now=after)


@pytest.mark.parametrize("cost", [None, True, -1, "NaN", "Infinity", {}, 1000001])
def test_unknown_accounting_never_releases_hold(ledger, cost):
    ledger.reserve(run(), now=NOW)
    ledger.observe(run().id, success(ledger, run(), cost), now=NOW)
    with pytest.raises(SpendAdmissionError, match="unresolved"):
        ledger.preflight(now=NOW)


@pytest.mark.parametrize(
    "field,value",
    [
        ("request_id", "wrong"),
        ("profile_ref", "wrong"),
        ("outcome", "failed"),
        ("sandbox_destroy", "failed"),
        ("session_cleanup", "failed"),
    ],
)
def test_incomplete_or_mismatched_result_retains_hold(ledger, field, value):
    ledger.reserve(run(), now=NOW)
    raw = success(ledger, run())
    raw["evidence"][field] = value
    ledger.observe(run().id, raw, now=NOW)
    assert ledger.status()["reservations"][0]["state"] == "held"


def test_overrun_is_recorded_and_freezes_even_after_reconcile(ledger):
    ledger.reserve(run(), now=NOW)
    ledger.observe(run().id, success(ledger, run(), 7), now=NOW)
    assert ledger.status()["breached"]
    assert ledger.status()["reservations"][0]["liability"] == 7_000_000
    with pytest.raises(SpendAdmissionError, match="discard"):
        ledger.reconcile(run().id, cost_usd="0", receipt="audit:final", now=NOW)
    ledger.reconcile(run().id, cost_usd="7", receipt="audit:final", now=NOW)
    with pytest.raises(SpendAdmissionError, match="breached"):
        ledger.preflight(now=NOW + timedelta(days=1))


def test_missing_corrupt_changed_policy_and_reinit_refuse(ledger):
    with pytest.raises(SpendAdmissionError):
        ledger.initialize()
    with pytest.raises(SpendAdmissionError, match="policy mismatch"):
        SpendLedger(ledger.path, replace(ledger.policy, total_eur="20")).preflight(
            now=NOW
        )
    ledger.path.unlink()
    with pytest.raises(SpendAdmissionError, match="unavailable"):
        ledger.preflight(now=NOW)
    assert not ledger.path.exists()
    ledger.path.write_bytes(b"corrupt")
    ledger.path.chmod(0o600)
    with pytest.raises(SpendAdmissionError, match="inconsistent"):
        ledger.preflight(now=NOW)


def test_private_paths_and_policy(ledger):
    config = configured(ledger)
    assert worker_spend(config).policy.sha256 == ledger.policy.sha256
    Path(config.spend_policy_path).chmod(0o644)
    with pytest.raises(SpendAdmissionError, match="private"):
        worker_spend(config)
    with pytest.raises(SpendAdmissionError, match="both"):
        worker_spend(replace(config, spend_policy_path=None))
    with pytest.raises(SpendAdmissionError, match="outside"):
        SpendLedger(Path(ledger.policy.target_repo) / "spend.db", ledger.policy)


def test_expiry_clock_rollback_and_upward_rounding(ledger):
    ledger.preflight(now=NOW)
    with pytest.raises(SpendAdmissionError, match="backwards"):
        ledger.preflight(now=NOW - timedelta(seconds=1))
    with pytest.raises(SpendAdmissionError, match="valid"):
        ledger.preflight(now=datetime(2099, 1, 1, tzinfo=UTC))
    policy = replace(ledger.policy, eur_per_usd="0.99999999")
    assert policy.reservation == 5_000_000
    with pytest.raises(SpendAdmissionError, match="envelope"):
        replace(ledger.policy, eur_per_usd="1.00000001")


def _reserve_process(path, policy, key):
    try:
        SpendLedger(Path(path), SpendPolicy(**policy)).reserve(run(key), now=NOW)
        return "reserved"
    except SpendAdmissionError:
        return "refused"


def test_concurrent_processes_cannot_double_admit(ledger):
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = list(
            pool.map(
                _reserve_process,
                [str(ledger.path)] * 4,
                [ledger.policy.__dict__] * 4,
                ["a", "b", "c", "d"],
            )
        )
    assert sorted(results) == ["refused", "refused", "refused", "reserved"]
    assert len(ledger.status()["reservations"]) == 1


def test_pending_spend_blocks_claim_but_close_replay_still_runs(ledger, monkeypatch):
    from rein_aharness.claim_loop import process_one
    from rein_aharness.close_outbox import CloseOutbox, CloseRequest

    ledger.reserve(run(), now=NOW)
    client = MagicMock(spec=ActivityCoreOpsClient)
    client.config = configured(ledger)
    client.complete.return_value = SimpleNamespace(state="succeeded")
    outbox = CloseOutbox()
    outbox.enqueue(
        CloseRequest(
            run_id="run-older",
            transaction_id="tx-1",
            worker_id="worker-1",
            action="complete",
            result={"ok": True},
        )
    )
    result = process_one(client, outbox=outbox)
    assert not result.claimed and "unresolved" in result.reason
    client.claim.assert_not_called()
    client.complete.assert_called_once()


def test_nonprofile_row_cannot_bypass_spend_gate(ledger, monkeypatch):
    from rein_aharness.claim_loop import process_one

    client = MagicMock(spec=ActivityCoreOpsClient)
    client.config = configured(ledger)
    client.claim.return_value = [replace(run(), harness_profile_ref=None)]
    client.fail.return_value = SimpleNamespace(state="failed")
    dispatched = []
    monkeypatch.setattr(
        "rein_aharness.claim_loop.execute_approach",
        lambda *a, **k: dispatched.append(True),
    )
    result = process_one(client)
    assert not result.ok and "profile" in result.reason and not dispatched
    assert ledger.status()["reservations"] == []


@pytest.fixture
def dispatch_case(ledger, monkeypatch):
    import subprocess

    from glas_harness.contract import ExecutionLimits, OperationalReadiness
    from glas_harness.profiles import ProfileCatalog

    target = Path(ledger.policy.target_repo)
    target.mkdir()
    subprocess.run(["git", "init", "-q", str(target)], check=True)
    catalog = ProfileCatalog()
    profile, descriptor = catalog.resolve("harness.agent-dev-local@1.0.0")
    profile = profile.model_copy(
        update={
            "id": "harness.test",
            "operational_readiness": OperationalReadiness(
                status="ready", evidence_ref="test:only"
            ),
            "limits": ExecutionLimits(max_budget_usd=4.0, max_turns=8),
        }
    )
    catalog.profiles()[(profile.id, profile.version)] = profile
    monkeypatch.setattr("glas_harness.profiles.ProfileCatalog", lambda: catalog)
    policy = replace(
        ledger.policy,
        profile_sha256=digest(profile.model_dump(mode="json")),
        descriptor_sha256=digest(descriptor.model_dump(mode="json")),
    )
    store = SpendLedger(ledger.path.parent / "dispatch.sqlite3", policy)
    store.initialize()
    # Artifact transfer is exercised unmodified by the separate actual bwrap test.
    transfer = MagicMock()
    transfer.import_after_teardown.return_value = {"fixture": True}
    monkeypatch.setattr(
        "rein_aharness.repository_artifact.RepositoryArtifactTransfer",
        lambda *a, **k: transfer,
    )
    return store, configured(store), catalog, transfer


def test_reservation_precedes_gateway_and_reclaim_never_repeats(dispatch_case):
    from rein_aharness.glas_execution import GlasExecutionError, execute_profiled_run

    store, config, catalog, transfer = dispatch_case
    calls = []

    def gateway(request, **kwargs):
        assert kwargs["catalog"] is catalog
        assert store.status()["reservations"][0]["state"] == "held"
        calls.append(request)
        return success(store, run())

    execute_profiled_run(run(), config, gateway=gateway, report_to_hub=False)
    assert store.status()["reservations"][0]["state"] == "charged"
    assert calls[0].project == "fixture-factory"
    transfer.import_after_teardown.assert_called_once()
    with pytest.raises(GlasExecutionError, match="replay"):
        execute_profiled_run(
            replace(run(), attempt=2), config, gateway=gateway, report_to_hub=False
        )
    assert len(calls) == 1


@pytest.mark.parametrize(
    "change",
    ["worker", "definition", "profile", "grant", "project", "limits", "descriptor"],
)
def test_scope_mismatch_denied_before_gateway_or_reservation(dispatch_case, change):
    from rein_aharness.glas_execution import GlasExecutionError, execute_profiled_run

    store, config, catalog, _transfer = dispatch_case
    item = run()
    if change == "worker":
        item.claim_owner = "other"
    if change == "definition":
        item.activity_definition_id = "other"
    if change == "profile":
        item.harness_profile_ref = "harness.agent-dev-local@1.0.0"
    if change == "grant":
        item.repository_grant = RepositoryGrant("1", ("elsewhere/",), 1, 1, False)
    if change == "project":
        config.execution_project = "other"
    if change == "limits":
        catalog.profiles()[("harness.test", "1.0.0")].limits.max_budget_usd = 5.0
    if change == "descriptor":
        catalog.reins()["rein-aharness"].version = "99.0.0"
    gateway = MagicMock()
    with pytest.raises(GlasExecutionError):
        execute_profiled_run(item, config, gateway=gateway, report_to_hub=False)
    gateway.assert_not_called()
    assert store.status()["reservations"] == []


@pytest.mark.parametrize(
    "failure", ["exception", "missing-cost", "overrun", "lease-loss"]
)
def test_uncertain_gateway_never_imports_and_blocks_next_claim(dispatch_case, failure):
    from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled
    from rein_aharness.glas_execution import GlasExecutionError, execute_profiled_run

    store, config, _catalog, transfer = dispatch_case
    cancel = ExecutionCancel()

    def gateway(request, **kwargs):
        if failure == "exception":
            raise RuntimeError("private provider response")
        if failure == "lease-loss":
            cancel.cancel("lease-loss")
        return success(
            store,
            run(),
            None if failure == "missing-cost" else 7 if failure == "overrun" else 1,
        )

    with pytest.raises((GlasExecutionError, ExecutionCancelled)):
        execute_profiled_run(
            run(), config, gateway=gateway, cancel=cancel, report_to_hub=False
        )
    transfer.import_after_teardown.assert_not_called()
    assert store.status()["reservations"][0]["state"] == "held"
    assert "private" not in json.dumps(store.status())
    with pytest.raises(SpendAdmissionError):
        store.preflight()


def test_cli_reconciliation_requires_termination_attestation(ledger, capsys):
    from rein_aharness.cli import main

    config = configured(ledger)
    ledger.reserve(run(), now=NOW)
    args = [
        "spend",
        "reconcile",
        "--policy",
        config.spend_policy_path,
        "--ledger",
        config.spend_ledger_path,
        "--run-id",
        run().id,
        "--cost-usd",
        "1",
        "--receipt",
        "audit:final",
    ]
    assert main(args) == 2
    assert ledger.status()["reservations"][0]["state"] == "held"
    assert main(args + ["--provider-stopped"]) == 0
    assert ledger.status()["reservations"][0]["state"] == "charged"
    capsys.readouterr()


def test_required_admission_missing_never_claims():
    from rein_aharness.claim_loop import process_one

    client = MagicMock(spec=ActivityCoreOpsClient)
    client.config = OpsRunConfig(require_spend_admission=True)
    result = process_one(client)
    assert not result.claimed and "not configured" in result.reason
    client.claim.assert_not_called()


def test_policy_file_rejects_duplicates_and_symlinks(ledger):
    config = configured(ledger)
    path = Path(config.spend_policy_path)
    text = path.read_text()
    path.write_text(text[:-1] + ', "max_budget_usd": "4"}')
    with pytest.raises(SpendAdmissionError, match="invalid"):
        SpendPolicy.load(path)
    path.write_text(text)
    link = path.with_name("symlink.json")
    link.symlink_to(path)
    with pytest.raises(SpendAdmissionError, match="private"):
        SpendPolicy.load(link)


def test_reconcile_unknown_identity_or_conflicting_receipt(ledger):
    with pytest.raises(SpendAdmissionError, match="unknown"):
        ledger.reconcile("absent", cost_usd="1", receipt="audit:final", now=NOW)
    ledger.reserve(run(), now=NOW)
    with pytest.raises(SpendAdmissionError, match="receipt"):
        ledger.reconcile(
            run().id, cost_usd="1", receipt="private text with spaces", now=NOW
        )
    ledger.reconcile(run().id, cost_usd="1", receipt="audit:final", now=NOW)
    with pytest.raises(SpendAdmissionError, match="conflict"):
        ledger.reconcile(run().id, cost_usd="0", receipt="audit:other", now=NOW)


def test_decimal_conversion_never_rounds_liability_down(ledger):
    from rein_aharness.spend_admission import cap_micros, converted_micros

    assert converted_micros("1.00000000000000000000000000000001", "1") == 1_000_001
    assert cap_micros("0.99999999999999999999999999999999") == 999_999
    assert (
        ledger.observe(run().id, success(ledger, run(), "1e-99999"), now=NOW) is False
    )