Observed-usage settlement: tests, renewal support and docs (REINAH-WP-0008-T02)

Completes c7dceee. owner renew accepts settlement and max_runs and writes
policies via as_dict, so unset terms stay out of the file and digest.
Tests cover metered request settlement, reported-cost floor, unfinished
holds, the 6efa9436 reconciliation shape, overrun freeze, and three
attempts within EUR 10 with the fourth refused.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 269692@bnt-lap001
Assistant-Session: 8077e68a-0985-4b41-898c-2dc602b0d6be

docs/owner-bootstrap.md

@@ -47,6 +47,23 @@ No digest is typed by hand. Paths do not change, so the Secrets Engine
 `exec_owner` catalog needs only the new file hashes, which are printed as
 `catalog_file_pins`. Update those pins, then run `--rehearse`.
 
+**Settlement and attempt limit (opt-in per envelope).** Two optional
+envelope fields change how spend is charged:
+
+- `settlement`: `full-hold` (default) charges every request and run at
+  least its whole hold. `observed-usage` charges a completed provider
+  request its gateway-metered cost (provider-reported tokens at the admitted
+  rates). A run is then charged the larger of its metered requests and the
+  cost the CLI reports, at completion and at owner reconciliation alike.
+  Requests whose stream never completed still count at their full hold, and
+  an overrun still freezes the envelope.
+- `max_runs`: the number of runs the envelope admits.
+
+Left unset, neither field is written or included in the policy digest, so
+existing pins are unchanged. Observed-usage trusts provider-reported usage
+until the owner's billing check, which is why it needs explicit owner
+acceptance (decision `a93619a9`).
+
 Use the standalone candidate interpreter as `python -I -B -m rein_aharness.cli`.
 `-I` excludes editable PYTHONPATH/user-site fallbacks; `-B` prevents bytecode writes
 from changing the complete artifact digest. The source and namespace rein commands

rein_aharness/owner_renew.py

@@ -33,6 +33,7 @@ ENVELOPE_FIELDS = frozenset({
     "envelope_id", "authority_ref", "valid_from", "expires_at",
     "eur_per_usd", "per_run_eur", "daily_eur", "total_eur",
     "max_budget_usd", "max_liability_usd",
+    "settlement", "max_runs",
 })
 _OWNER_KEYS = {"version", "authority_ref", "spend_policy_sha256", "messages_policy", "runtime"}
 
@@ -151,7 +152,7 @@ def apply(renewal: dict[str, Any]) -> dict[str, Any]:
         _write_private(backup, source.read_text())
     os.rename(ledger_path, retired)
     os.rename(staged, ledger_path)
-    _write_private(policy_path, json.dumps(renewal["new_policy"].__dict__, indent=2) + "\n")
+    _write_private(policy_path, json.dumps(renewal["new_policy"].as_dict(), indent=2) + "\n")
     _write_private(owner_config, json.dumps(renewal["new_owner"], indent=2) + "\n")
     return {
         "applied": True,

scripts/prove-metered-runtime.py

@@ -160,7 +160,7 @@ def case(root, runtime, checksum, server, cap, args, bootstrap=False):
     parent = SpendLedger(private / "spend.sqlite3", policy); parent.initialize()
     meter = RequestLedger(parent); meter.initialize()
     policy_path = private / "spend-policy.json"
-    policy_path.write_text(json.dumps(policy.__dict__)); policy_path.chmod(0o600)
+    policy_path.write_text(json.dumps(policy.as_dict())); policy_path.chmod(0o600)
     admission_requests = []
     class RecordingPolicy(MessagesPolicy):
         def validate(self, data, betas):

tests/test_owner_renew.py

@@ -116,3 +116,23 @@ def test_charged_reservation_renews_and_survives_in_retired_ledger(prepared, tmp
         ("charged", 10000000)]
     retired.close()
     assert SpendLedger(paths[2], SpendPolicy.load(paths[1])).status()["reservations"] == []
+
+
+def test_renewal_to_observed_usage_with_run_limit(prepared, tmp_path):
+    path, config, _, _ = prepared
+    envelope = {**ENVELOPE, "settlement": "observed-usage", "max_runs": 3,
+                "total_eur": "10", "daily_eur": "10"}
+    paths = _paths(prepared, tmp_path, envelope)
+    apply(plan(*paths))
+    written = json.loads(paths[1].read_text())
+    assert written["settlement"] == "observed-usage" and written["max_runs"] == 3
+    policy = SpendPolicy.load(paths[1])
+    assert json.loads(paths[0].read_text())["spend_policy_sha256"] == policy.sha256
+    prepare(path, config)
+
+
+def test_default_renewal_writes_no_optional_terms(prepared, tmp_path):
+    paths = _paths(prepared, tmp_path)
+    apply(plan(*paths))
+    written = json.loads(paths[1].read_text())
+    assert "settlement" not in written and "max_runs" not in written

tests/test_spend_settlement.py

@@ -0,0 +1,140 @@
+"""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"))

workplans/REINAH-WP-0008-simpler-metered-retry.md

@@ -92,7 +92,7 @@ glas-harness 0, claude 0.
 
 ```task
 id: REINAH-WP-0008-T02
-status: todo
+status: done
 priority: high
 state_hub_task_id: "6a413f5a-4ac2-5d5e-9b19-71b33445c0a9"
 ```
@@ -113,6 +113,30 @@ reservation, or with a live route, refuses the release. A second, identical
 reconcile changes nothing. The envelope is never marked breached by a
 release.
 
+### Widened to observed-usage settlement — 2026-10-14
+
+On 2026-10-13 the owner asked for three attempts within EUR 10 and chose
+option B: settle to actual usage instead of releasing only unused holds.
+Under full-hold accounting each request holds USD 4.64 (1M context plus
+64k output), so EUR 10 cannot fit three attempts.
+
+Implemented as an opt-in envelope term (`settlement: observed-usage`) plus
+`max_runs`; see `docs/owner-bootstrap.md`. Under it:
+
+- a completed request is charged its gateway-metered cost;
+- runs and reconciliations are charged
+  `max(metered requests, reported or attested cost)`;
+- unfinished requests keep their full hold, and overruns still freeze the
+  envelope.
+
+The run 6efa9436 shape (failure before any request) now reconciles to its
+attested cost. Defaults and existing digests are unchanged. Tests:
+`tests/test_spend_settlement.py`, plus renewal cases in
+`tests/test_owner_renew.py`.
+
+Merge-ready, but the rule still needs the owner's acceptance in decision
+`a93619a9`.
+
 ## D. One owner command for renewal
 
 ```task