#!/usr/bin/env python3 """Admit or refuse a platform order from the resource-control consumption-mode signal. RAILIANCE-WP-0017-T01. resource-control publishes the signal; this repo enforces it. Missing or unknown mode is not restricted (facility runbook). Only restricted + a new-order/elastic charge that exceeds the published allowance is refused. Safety paths are admitted and recorded as exceptions. """ from __future__ import annotations import argparse import json import os import sys from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any REPO_DIR = Path(__file__).resolve().parents[1] DEFAULT_SIGNAL = REPO_DIR / "data" / "consumption-mode" / "current.json" ENTITY_RE_PREFIX = "entity:" MODES = {"open", "restricted"} CLASSES = {"new-order", "elastic", "safety", "committed"} RAILIANCE = "entity:railiance" def money(value: Any) -> Decimal | None: if value is None or value == "": return None try: amount = Decimal(str(value)) except (InvalidOperation, ValueError): return None if amount < 0: return None return amount.quantize(Decimal("0.01")) def load_payload(path: Path) -> list[dict[str, Any]]: if not path.is_file(): return [] raw = json.loads(path.read_text(encoding="utf-8")) if isinstance(raw, list): out: list[dict[str, Any]] = [] for item in raw: if not isinstance(item, dict): continue if item.get("record_type") == "settlement_statement": out.append(signal_from_statement(item)) else: out.append(item) return out if isinstance(raw, dict) and raw.get("record_type") == "consumption_mode": return [raw] if isinstance(raw, dict) and isinstance(raw.get("signals"), list): return [item for item in raw["signals"] if isinstance(item, dict)] if isinstance(raw, dict) and raw.get("record_type") == "settlement_statement": return [signal_from_statement(raw)] return [] def signal_from_statement(statement: dict[str, Any]) -> dict[str, Any]: return { "schema_version": "0.1", "record_type": "consumption_mode", "financial_entity_id": statement.get("financial_entity_id"), "period": statement.get("period"), "consumption_mode": statement.get("consumption_mode"), "new_transfer_charges_allowed_eur": statement.get("next_month_allowance_eur"), "terms_version": statement.get("terms_version"), } def signal_for(signals: list[dict[str, Any]], entity_id: str) -> dict[str, Any] | None: matches = [ item for item in signals if item.get("financial_entity_id") == entity_id and item.get("record_type", "consumption_mode") == "consumption_mode" ] return matches[-1] if matches else None def decide( *, entity_id: str, order_class: str, estimate_eur: Decimal | None, signal: dict[str, Any] | None, ) -> dict[str, Any]: if entity_id == RAILIANCE: return _result("admit", "railiance-self-use", "open", None, order_class) if signal is None: return _result("admit", "no-signal-not-restricted", None, None, order_class) mode = signal.get("consumption_mode") if mode not in MODES: return _result("admit", "unknown-mode-not-restricted", None, None, order_class) allowance = money(signal.get("new_transfer_charges_allowed_eur")) if mode == "open": return _result("admit", "open", mode, allowance, order_class) if order_class == "safety": return _result( "admit", "safety-exception", mode, allowance, order_class, exception=True, ) if order_class == "committed": return _result( "admit", "committed-flagged", mode, allowance, order_class, exception=True, ) if allowance is None: return _result("refuse", "restricted-without-allowance", mode, allowance, order_class) if estimate_eur is None: return _result("refuse", "restricted-estimate-required", mode, allowance, order_class) if estimate_eur > allowance: return _result("refuse", "exceeds-allowance", mode, allowance, order_class) return _result("admit", "within-allowance", mode, allowance, order_class) def _result( decision: str, reason: str, mode: str | None, allowance: Decimal | None, order_class: str, exception: bool = False, ) -> dict[str, Any]: return { "decision": decision, "reason": reason, "consumption_mode": mode, "allowance_eur": None if allowance is None else f"{allowance:.2f}", "order_class": order_class, "exception": exception, } def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__.split("\n\n", 1)[0]) parser.add_argument("command", choices=["check"], help="check an order against the signal") parser.add_argument("--entity", required=True, help="financial_entity_id, e.g. entity:coulomb") parser.add_argument( "--class", dest="order_class", choices=sorted(CLASSES), default="new-order", ) parser.add_argument("--estimate-eur", help="transfer-price estimate for the new order") parser.add_argument( "--signal", type=Path, default=Path(os.environ.get("CONSUMPTION_MODE_FILE", DEFAULT_SIGNAL)), help="JSON signal file (list, one record, or a settlement statement)", ) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) if not args.entity.startswith(ENTITY_RE_PREFIX) or len(args.entity) < 8: print(f"[FAIL] --entity must be entity:: {args.entity}", file=sys.stderr) return 2 estimate = money(args.estimate_eur) if args.estimate_eur and estimate is None: print(f"[FAIL] --estimate-eur is not a non-negative EUR amount", file=sys.stderr) return 2 try: signals = load_payload(args.signal) except json.JSONDecodeError as exc: print(f"[FAIL] signal is not JSON: {exc}", file=sys.stderr) return 2 signal = signal_for(signals, args.entity) result = decide( entity_id=args.entity, order_class=args.order_class, estimate_eur=estimate, signal=signal, ) tag = "OK" if result["decision"] == "admit" else "REFUSE" print( f"[{tag}] {args.entity} {result['order_class']} " f"mode={result['consumption_mode'] or 'unknown'} " f"allowance={result['allowance_eur']} reason={result['reason']}" ) if result["exception"]: print("[EXCEPTION] restricted safety/committed path; record overage on the next statement") return 0 if result["decision"] == "admit" else 2 if __name__ == "__main__": sys.exit(main())