feat(wp-0002): complete T07 — control loop on the live backup resource

The backup is procured and proven, so the loop runs on real evidence.

- data/actuals/2026-08.json: first real observation. database 0.6365 GB,
  stored 0.0066 GB over 8 objects, backup success 1/1, restore RTO 1.08 min.
  Five proxies null, each with a named owner in measurement_gaps.
- data/thresholds/platform-audit-storage.json + tools/thresholds.py: budget
  variance, abnormal growth, stale backup, unused commitment. Fail-closed —
  an unmeasured value is reported as unmeasured, never as within.
- financial_exchange.py gains a usage mode emitting technical_usage records to
  fin-hub, with measurement gaps carried through and no infrastructure amount:
  fin-hub owns the booked fact and a null is never sent as 0.00.
- observation schema 0.2 allows null cost and usage proxies; variance.py fails
  closed rather than reporting a 100% favourable variance on a missing amount.
- platform-audit-storage: ordered -> active, commissioned 2026-08-14, on
  operational fact rather than on the purchase.

The optimization case is now approved by the founder. That needed a schema
change: Host Europe never supplied written terms, so options gained
excluded/exclusion_reason. Previously an unevaluable alternative blocked its
case forever, leaving the record claiming no decision while the bucket was in
production. An excluded option keeps its unknowns and must say what would
bring it back.

August produces no variance and should not: the decision forecast starts at
2026-09, so August is a commissioning baseline. Threshold run is 2 within,
1 not applicable, 6 unmeasured, 0 breaches.

Also fixes a pre-existing test failure: reef-storage consumers_actual is now
rapp-postgres, which the assertion still expected to be empty.

136 tests pass.

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

tools/financial_exchange.py

@@ -76,6 +76,67 @@ def forecast_records(payload: dict, *, resource_id: str, source_ref: str) -> lis
     return records
 
 
+def usage_records(payload: dict, *, resource_id: str, source_ref: str) -> list[dict]:
+    """Normalize a monthly technical observation for the fin-hub exchange.
+
+    Only the resource-control-to-fin-hub direction: technical usage and valued
+    internal labour. Infrastructure cost is deliberately absent — fin-hub owns
+    the booked fact, and an uninvoiced period carries no amount to send. A null
+    stays null; it is never sent as 0.00.
+    """
+    if payload.get("schema_version") not in {"0.1", "0.2"}:
+        raise ValueError("expected a resource-control observation v0.1 or v0.2")
+    if payload.get("record_type") != "usage_observation":
+        raise ValueError("expected record_type usage_observation")
+    if not resource_id.startswith("resource:"):
+        raise ValueError("resource_id must use the resource: namespace")
+
+    records = []
+    for row in payload["rows"]:
+        year, month = (int(part) for part in row["period"].split("-"))
+        period_start = date(year, month, 1)
+        period_end = date(year, month, monthrange(year, month)[1])
+        labor = row["internal_labor_eur"]
+        records.append({
+            "schema_version": "0.1",
+            "record_type": "technical_usage",
+            "record_id": (
+                f"usage:{payload['provider_id']}:{payload['cost_attribution_key']}:{row['period']}"
+            ),
+            "resource_id": resource_id,
+            "service_id": payload["provider_id"],
+            "workload_id": payload["workload"],
+            "tenant_id": None,
+            "environment": "production",
+            "cost_attribution_key": payload["cost_attribution_key"],
+            "period_start": period_start.isoformat(),
+            "period_end": period_end.isoformat(),
+            "currency": "EUR",
+            "created_at": payload["created_at"],
+            "source_evidence": [source_ref, *payload.get("evidence", [])],
+            "usage": {
+                key: row[key]
+                for key in (
+                    "database_gb", "stored_gb", "wal_gb", "restore_egress_gb",
+                    "write_requests", "read_requests",
+                )
+            },
+            "service_evidence": {
+                key: row.get(key)
+                for key in ("backup_success_pct", "restore_rto_minutes")
+            },
+            "costs": {
+                # resource-control originates internal labour only. Booked
+                # infrastructure cost is fin-hub's to record, not ours to echo.
+                "internal_labor": None if labor is None else money_text(labor),
+                "internal_labor_hours": row["internal_labor_hours"],
+            },
+            "measurement_gaps": row.get("measurement_gaps", []),
+            "notes": payload.get("notes", []),
+        })
+    return records
+
+
 def validate_booked_cost(record: dict) -> dict:
     required = {
         "schema_version", "record_type", "financial_fact_id", "adjustment_kind",
@@ -126,13 +187,19 @@ def reconcile(forecasts: list[dict], booked_costs: list[dict]) -> list[dict]:
 
 
 def main() -> int:
-    if len(sys.argv) < 3 or sys.argv[1] not in {"forecast", "reconcile"}:
-        print(f"usage: {sys.argv[0]} forecast FORECAST.json [RESOURCE_ID] | reconcile FORECAST_EVIDENCE.json BOOKED_COST.json", file=sys.stderr)
+    if len(sys.argv) < 3 or sys.argv[1] not in {"forecast", "usage", "reconcile"}:
+        print(
+            f"usage: {sys.argv[0]} forecast FORECAST.json [RESOURCE_ID]"
+            f" | usage OBSERVATION.json [RESOURCE_ID]"
+            f" | reconcile FORECAST_EVIDENCE.json BOOKED_COST.json",
+            file=sys.stderr,
+        )
         return 2
-    if sys.argv[1] == "forecast":
+    if sys.argv[1] in {"forecast", "usage"}:
         payload = json.loads(Path(sys.argv[2]).read_text())
         resource_id = sys.argv[3] if len(sys.argv) > 3 else "resource:platform_audit_storage"
-        print(json.dumps(forecast_records(payload, resource_id=resource_id, source_ref=sys.argv[2]), indent=2))
+        build = forecast_records if sys.argv[1] == "forecast" else usage_records
+        print(json.dumps(build(payload, resource_id=resource_id, source_ref=sys.argv[2]), indent=2))
         return 0
     forecasts = json.loads(Path(sys.argv[2]).read_text())
     booked = json.loads(Path(sys.argv[3]).read_text())

tools/optimization.py

@@ -64,6 +64,18 @@ def utilization_ratios(option: dict) -> dict:
 
 def compare_option(baseline: dict, alternative: dict) -> dict:
     """Compare one alternative against the baseline on the decision fields."""
+    if alternative.get("excluded"):
+        # The deciding authority set this option aside rather than wait for
+        # evidence that was not going to arrive. It no longer blocks the case,
+        # and the reason travels with the decision.
+        return {
+            "option_id": alternative["option_id"],
+            "label": alternative["label"],
+            "verdict": "excluded",
+            "exclusion_reason": alternative.get("exclusion_reason"),
+            "blocking_evidence": [],
+        }
+
     blocking = sorted(set(
         [f"baseline.{field}" for field in missing_fields(baseline)]
         + [f"{alternative['option_id']}.{field}" for field in missing_fields(alternative)]
@@ -150,6 +162,12 @@ def validate_case(case: dict) -> None:
     if len(set(option_ids)) != len(option_ids):
         raise ValueError("option identifiers must be unique within a case")
 
+    for alternative in case["alternatives"]:
+        if alternative.get("excluded") and not alternative.get("exclusion_reason"):
+            raise ValueError(
+                f"excluded option {alternative['option_id']} must record an exclusion_reason"
+            )
+
     report = evaluate(case)
     blocked = any(r["verdict"] == "blocked_on_evidence" for r in report["comparisons"])
 

tools/thresholds.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""Evaluate a monthly observation against a resource's declared thresholds.
+
+Fail-closed in both directions. An unknown measurement never passes a
+threshold — a missing number is reported as `unmeasured`, because a threshold
+that silently passes on absent evidence is worse than no threshold at all.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import variance
+
+# Verdicts, in the order a reviewer should read them.
+BREACH = "breach"
+UNMEASURED = "unmeasured"
+NOT_APPLICABLE = "not_applicable"
+WITHIN = "within"
+
+
+def _row_for(payload: dict, period: str) -> dict | None:
+    for row in payload["rows"]:
+        if row["period"] == period:
+            return row
+    return None
+
+
+def evaluate_threshold(threshold: dict, observed: dict, compared: dict | None) -> dict:
+    metric = threshold["metric"]
+    result = {
+        "id": threshold["id"],
+        "kind": threshold["kind"],
+        "metric": metric,
+        "limit": threshold["limit"],
+        "action": threshold["action"],
+    }
+
+    if threshold.get("status") == "not_applicable":
+        return {**result, "verdict": NOT_APPLICABLE, "detail": threshold["action"]}
+
+    value = observed.get(metric)
+    if value is None:
+        gaps = [gap for gap in observed.get("measurement_gaps", []) if gap.startswith(f"{metric}:")]
+        return {
+            **result,
+            "verdict": UNMEASURED,
+            "detail": gaps[0] if gaps else f"{metric} is not present in the observation",
+        }
+
+    comparison = threshold["comparison"]
+    if comparison == "minimum":
+        breached = value < threshold["limit"]
+        return {**result, "verdict": BREACH if breached else WITHIN, "observed": value}
+    if comparison == "maximum":
+        breached = value > threshold["limit"]
+        return {**result, "verdict": BREACH if breached else WITHIN, "observed": value}
+    if comparison == "unplanned":
+        breached = value > threshold["limit"]
+        return {**result, "verdict": BREACH if breached else WITHIN, "observed": value}
+
+    # The remaining comparisons need a forecast to compare against.
+    if compared is None or metric not in compared:
+        return {
+            **result,
+            "verdict": UNMEASURED,
+            "observed": value,
+            "detail": "no forecast row exists for this period, so no variance can be computed",
+        }
+    entry = compared[metric]
+    if entry.get("status") == "unknown":
+        return {**result, "verdict": UNMEASURED, "observed": value, "detail": "forecast or actual amount is unknown"}
+    if comparison == "absolute_percentage_error":
+        measured = entry["absolute_percentage_error"]
+        if measured is None:
+            return {
+                **result,
+                "verdict": UNMEASURED,
+                "observed": value,
+                "detail": "forecast is zero, so percentage error is undefined; review the absolute error instead",
+            }
+    else:
+        measured = abs(entry["error"])
+    return {
+        **result,
+        "verdict": BREACH if measured > threshold["limit"] else WITHIN,
+        "observed": value,
+        "measured": measured,
+    }
+
+
+def evaluate(config: dict, observation: dict, forecast: dict | None, period: str) -> dict:
+    observed = _row_for(observation, period)
+    if observed is None:
+        raise ValueError(f"observation has no row for {period}")
+
+    compared = None
+    if forecast is not None:
+        report = variance.compare(forecast, observation)
+        for row in report["rows"]:
+            if row["period"] == period and row["status"] == "compared":
+                compared = row["metrics"]
+
+    results = [evaluate_threshold(t, observed, compared) for t in config["thresholds"]]
+    counts: dict[str, int] = {}
+    for result in results:
+        counts[result["verdict"]] = counts.get(result["verdict"], 0) + 1
+    return {
+        "resource_id": config["resource_id"],
+        "period": period,
+        "forecast_available": compared is not None,
+        "results": results,
+        "summary": dict(sorted(counts.items())),
+        "breaches": [r["id"] for r in results if r["verdict"] == BREACH],
+        "unmeasured": [r["id"] for r in results if r["verdict"] == UNMEASURED],
+        "known_gaps": config.get("known_gaps", []),
+    }
+
+
+def main() -> int:
+    if len(sys.argv) < 4:
+        print(f"usage: {sys.argv[0]} THRESHOLDS.json OBSERVATION.json PERIOD [FORECAST.json]", file=sys.stderr)
+        return 2
+    config = json.loads(Path(sys.argv[1]).read_text())
+    observation = json.loads(Path(sys.argv[2]).read_text())
+    period = sys.argv[3]
+    forecast = json.loads(Path(sys.argv[4]).read_text()) if len(sys.argv) > 4 else None
+    report = evaluate(config, observation, forecast, period)
+    print(json.dumps(report, indent=2))
+    # A breach is a non-zero exit so the monthly cadence can gate on it.
+    return 1 if report["breaches"] else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/variance.py

@@ -26,6 +26,16 @@ def compare(forecast: dict, actual: dict) -> dict:
         for metric in METRICS:
             planned = expected[period][metric]
             measured = observed[metric]
+            # An uninvoiced period has no cost to compare against. Treating a
+            # missing amount as zero would report a 100% favourable variance.
+            if planned is None or measured is None:
+                metrics[metric] = {
+                    "forecast": planned,
+                    "actual": measured,
+                    "status": "unknown",
+                    "category": "data_quality",
+                }
+                continue
             error = measured - planned
             metrics[metric] = {
                 "forecast": planned,