rein-aharness/tests/test_spend_settlement.py

"""Opt-in observed-usage settlement and run limits (REINAH-WP-0008, decision a93619a9)."""

from __future__ import annotations

import json
from dataclasses import replace
from datetime import UTC, datetime, timedelta

import pytest

from rein_aharness.request_admission import RequestLedger
from rein_aharness.spend_admission import (
    SpendAdmissionError,
    SpendLedger,
    SpendPolicy,
    digest,
)
from test_spend_admission import ledger as ledger, run, success

POLICY_SHA = "c" * 64


def _observed(ledger, **changes):
    policy = replace(ledger.policy, settlement="observed-usage", **changes)
    store = SpendLedger(ledger.path.with_name("observed.sqlite3"), policy)
    store.initialize()
    meter = RequestLedger(store)
    meter.initialize()
    return store, meter


def _route(meter, key):
    return meter.bind_route(
        run(key).id, POLICY_SHA, lease_id=f"lease-{key}",
        expires_at=(datetime.now(UTC) + timedelta(seconds=60)).isoformat())


def _liability(store, key):
    row = next(r for r in store.status()["reservations"] if r["run_id"] == run(key).id)
    return row["state"], row["liability"]


def test_default_policy_digest_is_unchanged(ledger, tmp_path):
    policy = ledger.policy
    legacy = {k: v for k, v in policy.__dict__.items() if k not in ("settlement", "max_runs")}
    assert policy.sha256 == digest(legacy)
    assert replace(policy, settlement="observed-usage").sha256 != policy.sha256
    assert replace(policy, max_runs=3).sha256 != policy.sha256
    path = tmp_path / "legacy.json"
    path.write_text(json.dumps(legacy)); path.chmod(0o600)
    assert SpendPolicy.load(path).sha256 == policy.sha256


@pytest.mark.parametrize("changes", [{"settlement": "refund"}, {"max_runs": 0},
                                     {"max_runs": "3"}, {"max_runs": True}])
def test_invalid_terms_refuse(ledger, changes):
    with pytest.raises(SpendAdmissionError):
        replace(ledger.policy, **changes)


def test_full_hold_default_still_charges_whole_hold(ledger):
    meter = RequestLedger(ledger)
    meter.initialize()
    ledger.reserve(run())
    token = _route(meter, "one")
    meter.complete_request(meter.reserve_request(token, POLICY_SHA, 2_000_000), 10)
    assert meter.status()[0]["liability_microusd"] == 2_000_000


def test_completed_requests_settle_to_metered_cost(ledger):
    store, meter = _observed(ledger)
    store.reserve(run())
    token = _route(meter, "one")
    # Four 2 USD holds in a 5 USD run: impossible under full hold.
    for _ in range(4):
        meter.complete_request(meter.reserve_request(token, POLICY_SHA, 2_000_000), 150_000)
    assert [r["liability_microusd"] for r in meter.status()] == [150_000] * 4
    assert store.observe(run().id, success(store, run(), 0.1))
    assert _liability(store, "one") == ("charged", 600_000)


def test_reported_cost_wins_when_higher_than_metered(ledger):
    store, meter = _observed(ledger)
    store.reserve(run())
    token = _route(meter, "one")
    meter.complete_request(meter.reserve_request(token, POLICY_SHA, 2_000_000), 100_000)
    assert store.observe(run().id, success(store, run(), 0.9))
    assert _liability(store, "one") == ("charged", 900_000)


def test_zero_usage_still_records_a_positive_charge(ledger):
    store, meter = _observed(ledger)
    store.reserve(run())
    token = _route(meter, "one")
    meter.complete_request(meter.reserve_request(token, POLICY_SHA, 2_000_000), 0)
    assert meter.status()[0]["liability_microusd"] == 1


def test_unfinished_request_keeps_full_hold_through_reconciliation(ledger):
    store, meter = _observed(ledger)
    store.reserve(run())
    token = _route(meter, "one")
    meter.complete_request(meter.reserve_request(token, POLICY_SHA, 2_000_000), 100_000)
    meter.reserve_request(token, POLICY_SHA, 2_000_000)  # stream never completed
    assert not store.observe(run().id, success(store, run(), 0.05))
    assert _liability(store, "one")[0] == "held"
    store.reconcile(run().id, cost_usd="0.05", receipt="audit:provider-stopped-final")
    assert _liability(store, "one") == ("charged", 2_100_000)


def test_failed_run_without_requests_reconciles_to_attested_cost(ledger):
    # The run 6efa9436 shape: failure before any provider request.
    store, meter = _observed(ledger)
    store.reserve(run())
    _route(meter, "one")
    assert not store.observe(run().id, {"ok": False, "evidence": {}})
    store.reconcile(run().id, cost_usd="0.05", receipt="owner-estimate:launcher-exit127")
    assert _liability(store, "one") == ("charged", 50_000)


def test_overrun_still_freezes_the_envelope(ledger):
    store, meter = _observed(ledger)
    store.reserve(run())
    token = _route(meter, "one")
    receipt = meter.reserve_request(token, POLICY_SHA, 100_000)
    with pytest.raises(SpendAdmissionError, match="breached"):
        meter.complete_request(receipt, 200_000)
    assert store.status()["breached"] is True


def test_three_attempts_within_eur_10_then_run_limit(ledger):
    store, meter = _observed(ledger, total_eur="10", daily_eur="10", max_runs=3)
    for key in ("one", "two", "three"):
        store.reserve(run(key))
        token = _route(meter, key)
        meter.complete_request(meter.reserve_request(token, POLICY_SHA, 4_640_000), 400_000)
        assert store.observe(run(key).id, success(store, run(key), 0.2))
    assert sum(r["liability"] for r in store.status()["reservations"]) == 1_200_000
    with pytest.raises(SpendAdmissionError, match="run limit"):
        store.reserve(run("four"))