WIP: opt-in observed-usage settlement and run limit for spend envelopes

SpendPolicy gains optional settlement (full-hold default | observed-usage)
and max_runs, omitted from the digest when unset so existing pins hold.
Under observed-usage, completed requests charge gateway-metered cost and
runs/reconciliations charge max(metered requests, reported cost).
Not yet done: dedicated tests, owner renew envelope fields, docs, memo.

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

.custodian-brief.md

@@ -2,7 +2,7 @@
 # Custodian Brief — rein-aharness
 
 **Domain:** agents  
-**Last synced:** 2026-10-13 21:25 UTC  
+**Last synced:** 2026-10-13 21:26 UTC  
 **State Hub:** http://127.0.0.1:8000 *(adjust if running on a remote machine)*
 
 ## Active Workstreams
@@ -14,8 +14,8 @@ Progress: 2/7 done  |  workplan_id: `187c2599-84a6-5764-a14f-7e9121975bf2`
 - ! C. Envelopes that cover several attempts (decision)  `898e02af`
 - ! E. Ship the owner runtime as a container image (decision)  `b486e86b`
 - ! F. Approvals per envelope (decision)  `4cf721e6`
-- ! Retry on the corrected runtime  `80318842`
 - · B. Release holds that provably cost nothing  `6a413f5a`
+- · Retry on the corrected runtime  `80318842`
 
 ### Governed runtime integrity and intent convergence
 Progress: 3/6 done  |  workplan_id: `eba2eff1-10a7-50a3-a70b-14e7d398f27f`

rein_aharness/request_admission.py

@@ -204,7 +204,10 @@ class RequestLedger:
                 "UPDATE request_reservations SET state=?, liability_microusd=?, observed_microusd=? WHERE receipt=?",
                 (
                     "held" if breached else "charged",
-                    max(row["liability_microusd"], observed_microusd),
+                    max(row["liability_microusd"], observed_microusd)
+                    if breached or not self.parent.policy.settles_observed
+                    # Gateway-metered usage of a completed stream (decision a93619a9).
+                    else max(observed_microusd, 1),
                     observed_microusd,
                     receipt,
                 ),

rein_aharness/spend_admission.py

@@ -26,6 +26,14 @@ class SpendAdmissionError(RuntimeError):
     """A bounded refusal; never contains prompt, credential or provider output."""
 
 
+# Full hold: every reservation is charged at least its whole hold (default).
+# Observed usage: a completed provider request is charged its gateway-metered
+# cost; a run is charged the larger of its metered requests and its reported
+# cost. Unfinished requests still count at their full hold.
+SETTLEMENT_FULL_HOLD = "full-hold"
+SETTLEMENT_OBSERVED = "observed-usage"
+
+
 def digest(value: Any) -> str:
     return hashlib.sha256(
         json.dumps(
@@ -105,12 +113,24 @@ class SpendPolicy:
     per_run_eur: str
     daily_eur: str
     total_eur: str
+    # Optional envelope terms (REINAH-WP-0008, decision a93619a9). Unset, they
+    # are omitted from the digest, so existing policies keep their pins.
+    settlement: str = SETTLEMENT_FULL_HOLD
+    max_runs: int | None = None
 
     def __post_init__(self) -> None:
         if self.version != "1":
             raise SpendAdmissionError("unsupported spend policy version")
+        if self.settlement not in (SETTLEMENT_FULL_HOLD, SETTLEMENT_OBSERVED):
+            raise SpendAdmissionError("invalid settlement rule")
+        if self.max_runs is not None and (
+            type(self.max_runs) is not int or not 1 <= self.max_runs <= 100
+        ):
+            raise SpendAdmissionError("invalid run limit")
         for name in self.__dataclass_fields__:
             value = getattr(self, name)
+            if name in ("settlement", "max_runs"):
+                continue
             if name == "max_turns":
                 if type(value) is not int or not 1 <= value <= 10000:
                     raise SpendAdmissionError("invalid turn limit")
@@ -156,7 +176,20 @@ class SpendPolicy:
 
     @property
     def sha256(self) -> str:
-        return digest(self.__dict__)
+        return digest(self.as_dict())
+
+    def as_dict(self) -> dict[str, Any]:
+        """Policy fields, omitting optional terms that are left at their default."""
+        values = dict(self.__dict__)
+        if values["settlement"] == SETTLEMENT_FULL_HOLD:
+            del values["settlement"]
+        if values["max_runs"] is None:
+            del values["max_runs"]
+        return values
+
+    @property
+    def settles_observed(self) -> bool:
+        return self.settlement == SETTLEMENT_OBSERVED
 
     @classmethod
     def load(cls, path: Path) -> SpendPolicy:
@@ -303,6 +336,10 @@ class SpendLedger:
             raise SpendAdmissionError(
                 "unresolved spend reservation; reconcile before dispatch"
             )
+        if self.policy.max_runs is not None and db.execute(
+            "SELECT count(*) FROM reservations"
+        ).fetchone()[0] >= self.policy.max_runs:
+            raise SpendAdmissionError("envelope run limit reached")
         total = db.execute(
             "SELECT COALESCE(SUM(liability),0) FROM reservations"
         ).fetchone()[0]
@@ -396,9 +433,11 @@ class SpendLedger:
             ).fetchone()
             if row is None or row["state"] != "held":
                 raise SpendAdmissionError("reservation observation conflict")
-            liability = max(
-                row["liability"], converted_micros(str(cost), self.policy.eur_per_usd)
-            )
+            reported = converted_micros(str(cost), self.policy.eur_per_usd)
+            if self.policy.settles_observed and requests_resolved:
+                liability = max(self._metered(db, run_id), reported)
+            else:
+                liability = max(row["liability"], reported)
             breached = cost > amount(self.policy.max_liability_usd)
             complete = (
                 raw.get("ok") is True
@@ -424,6 +463,24 @@ class SpendLedger:
                 db.execute("UPDATE envelope SET breached=1")
         return complete
 
+    @staticmethod
+    def _metering(db: sqlite3.Connection) -> bool:
+        return bool(db.execute(
+            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='request_routes'"
+        ).fetchone())
+
+    def _metered(self, db: sqlite3.Connection, run_id: str) -> int:
+        """The run's gateway-metered request liability, in EUR micros."""
+        if not self._metering(db):
+            return 0
+        total = db.execute(
+            "SELECT COALESCE(SUM(liability_microusd),0) FROM request_reservations WHERE run_id=?",
+            (run_id,),
+        ).fetchone()[0]
+        return converted_micros(
+            f"{total // 1_000_000}.{total % 1_000_000:06d}", self.policy.eur_per_usd
+        )
+
     @staticmethod
     def _close_requests(
         db: sqlite3.Connection, run_id: str, *, reconcile: bool = False
@@ -473,9 +530,12 @@ class SpendLedger:
                     "reconciliation cannot discard observed liability"
                 )
             self._close_requests(db, run_id, reconcile=True)
-            liability = max(
-                row["liability"], converted_micros(str(cost), self.policy.eur_per_usd)
-            )
+            attested = converted_micros(str(cost), self.policy.eur_per_usd)
+            if self.policy.settles_observed and self._metering(db):
+                # Unfinished requests were just charged at their full hold.
+                liability = max(self._metered(db, run_id), attested)
+            else:
+                liability = max(row["liability"], attested)
             db.execute(
                 "UPDATE reservations SET state='charged', liability=?, observed_usd=?, end_day=?, receipt=? WHERE run_id=?",
                 (liability, str(cost), day, receipt, run_id),