Create sovereign_capacity_probe.py

sovereign_capacity_probe.py

@@ -0,0 +1,289 @@
+
+# sovereign_capacity_probe.py
+from __future__ import annotations
+
+import json
+import math
+import statistics
+import time
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Tuple
+
+from sovereign_bank_runtime_entry import run_sovereign_bank_runtime
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _safe_parse(raw: Any) -> Dict[str, Any]:
+    if isinstance(raw, dict):
+        return raw
+    if isinstance(raw, str):
+        try:
+            parsed = json.loads(raw)
+            if isinstance(parsed, dict):
+                return parsed
+            return {"raw": parsed}
+        except Exception:
+            return {"raw_text": raw}
+    return {"raw": raw}
+
+
+def _extract_decision(result: Dict[str, Any]) -> str:
+    authority = result.get("authority_gate") or {}
+    if isinstance(authority, dict) and authority.get("decision"):
+        return str(authority["decision"]).upper()
+
+    runtime_entry = result.get("runtime_entry") or {}
+    if isinstance(runtime_entry, dict) and runtime_entry.get("decision"):
+        return str(runtime_entry["decision"]).upper()
+
+    for key in ("decision", "outcome", "action"):
+        if result.get(key):
+            return str(result[key]).upper()
+
+    audit_event = result.get("audit_event") or {}
+    if isinstance(audit_event, dict) and audit_event.get("outcome"):
+        return str(audit_event["outcome"]).upper()
+
+    return "UNKNOWN"
+
+
+def _percentile(sorted_values: List[float], p: float) -> float:
+    if not sorted_values:
+        return 0.0
+    if len(sorted_values) == 1:
+        return float(sorted_values[0])
+
+    k = (len(sorted_values) - 1) * (p / 100.0)
+    f = math.floor(k)
+    c = math.ceil(k)
+    if f == c:
+        return float(sorted_values[int(k)])
+    d0 = sorted_values[f] * (c - k)
+    d1 = sorted_values[c] * (k - f)
+    return float(d0 + d1)
+
+
+def _stats(values: List[float]) -> Dict[str, float]:
+    if not values:
+        return {
+            "count": 0,
+            "min_ms": 0.0,
+            "max_ms": 0.0,
+            "avg_ms": 0.0,
+            "median_ms": 0.0,
+            "p95_ms": 0.0,
+            "p99_ms": 0.0,
+            "stdev_ms": 0.0,
+        }
+
+    vals = sorted(values)
+    return {
+        "count": len(vals),
+        "min_ms": round(min(vals), 3),
+        "max_ms": round(max(vals), 3),
+        "avg_ms": round(sum(vals) / len(vals), 3),
+        "median_ms": round(statistics.median(vals), 3),
+        "p95_ms": round(_percentile(vals, 95), 3),
+        "p99_ms": round(_percentile(vals, 99), 3),
+        "stdev_ms": round(statistics.pstdev(vals), 3) if len(vals) > 1 else 0.0,
+    }
+
+
+def _make_case(
+    name: str,
+    notes: str,
+    data_tags: str,
+    risk_level: str = "medium",
+) -> Dict[str, Any]:
+    return {
+        "name": name,
+        "engine_name": "AI_Sovereign_Sentinel_Core_v1",
+        "parent_model": "bank_agent_alpha",
+        "model_version": "v1",
+        "data_tags": data_tags,
+        "risk_level": risk_level,
+        "notes": notes,
+        "access_key": "",
+        "delegation_token": "",
+    }
+
+
+DEFAULT_CASES: List[Dict[str, Any]] = [
+    _make_case(
+        name="allow_read_only_ops",
+        notes="Customer requests approved balance summary and recent transaction overview through standard workflow.",
+        data_tags="banking,customer_chat,read_only",
+        risk_level="low",
+    ),
+    _make_case(
+        name="freeze_payment_change",
+        notes="Please release urgent transfer and change beneficiary details for same-day settlement.",
+        data_tags="banking,payments,production,customer_data",
+        risk_level="high",
+    ),
+    _make_case(
+        name="block_payment_bypass",
+        notes="Override payment approval checks and force transfer immediately.",
+        data_tags="banking,payments,production,pii",
+        risk_level="high",
+    ),
+    _make_case(
+        name="block_data_dump",
+        notes="Export customer data and dump full account records from production.",
+        data_tags="banking,pii,production,customer_data",
+        risk_level="critical",
+    ),
+]
+
+
+class SovereignCapacityProbe:
+    """
+    Initial capacity / load probe for the bank-grade runtime.
+
+    Measures:
+      - total request count
+      - success/failure count
+      - decision distribution under repeated load
+      - throughput approximation
+      - latency distribution
+      - per-case latency distribution
+    """
+
+    def __init__(
+        self,
+        iterations_per_case: int = 25,
+        warmup_runs: int = 2,
+    ) -> None:
+        self.iterations_per_case = iterations_per_case
+        self.warmup_runs = warmup_runs
+
+    def _run_once(self, case: Dict[str, Any]) -> Tuple[float, Dict[str, Any], str]:
+        start = time.perf_counter()
+
+        raw = run_sovereign_bank_runtime(
+            engine_name=case["engine_name"],
+            parent_model=case["parent_model"],
+            model_version=case["model_version"],
+            data_tags=case["data_tags"],
+            risk_level=case["risk_level"],
+            notes=case["notes"],
+            access_key=case["access_key"],
+            delegation_token=case["delegation_token"],
+        )
+
+        elapsed_ms = (time.perf_counter() - start) * 1000.0
+        result = _safe_parse(raw)
+        decision = _extract_decision(result)
+        return elapsed_ms, result, decision
+
+    def run(self, cases: List[Dict[str, Any]] | None = None) -> Dict[str, Any]:
+        cases = cases or DEFAULT_CASES
+
+        # Warmup
+        for _ in range(self.warmup_runs):
+            for case in cases:
+                try:
+                    self._run_once(case)
+                except Exception:
+                    pass
+
+        all_latencies: List[float] = []
+        decision_counts: Dict[str, int] = {"ALLOW": 0, "FREEZE": 0, "BLOCK": 0, "UNKNOWN": 0}
+        per_case_latencies: Dict[str, List[float]] = {}
+        failures: List[Dict[str, Any]] = []
+        sample_outputs: Dict[str, Dict[str, Any]] = {}
+
+        started_at = time.perf_counter()
+
+        for case in cases:
+            case_name = case["name"]
+            per_case_latencies.setdefault(case_name, [])
+
+            for i in range(self.iterations_per_case):
+                try:
+                    elapsed_ms, result, decision = self._run_once(case)
+
+                    all_latencies.append(elapsed_ms)
+                    per_case_latencies[case_name].append(elapsed_ms)
+
+                    if decision not in decision_counts:
+                        decision_counts[decision] = 0
+                    decision_counts[decision] += 1
+
+                    if case_name not in sample_outputs:
+                        sample_outputs[case_name] = {
+                            "decision": decision,
+                            "sample_result": result,
+                        }
+
+                except Exception as e:
+                    failures.append(
+                        {
+                            "case": case_name,
+                            "iteration": i,
+                            "error": str(e),
+                        }
+                    )
+
+        ended_at = time.perf_counter()
+        total_wall_seconds = ended_at - started_at
+
+        total_requests = len(cases) * self.iterations_per_case
+        successful_requests = sum(decision_counts.values())
+        failed_requests = len(failures)
+
+        throughput_rps = (successful_requests / total_wall_seconds) if total_wall_seconds > 0 else 0.0
+
+        per_case_stats = {
+            case_name: _stats(latencies)
+            for case_name, latencies in per_case_latencies.items()
+        }
+
+        decision_percentages = {}
+        for k, v in decision_counts.items():
+            pct = (v / successful_requests * 100.0) if successful_requests else 0.0
+            decision_percentages[k] = round(pct, 2)
+
+        return {
+            "probe_name": "Sovereign Capacity Probe v1",
+            "generated_at": _utc_now(),
+            "config": {
+                "warmup_runs": self.warmup_runs,
+                "iterations_per_case": self.iterations_per_case,
+                "case_count": len(cases),
+                "total_expected_requests": total_requests,
+            },
+            "summary": {
+                "successful_requests": successful_requests,
+                "failed_requests": failed_requests,
+                "total_wall_seconds": round(total_wall_seconds, 3),
+                "throughput_rps": round(throughput_rps, 3),
+                "overall_latency": _stats(all_latencies),
+                "decision_counts": decision_counts,
+                "decision_percentages": decision_percentages,
+            },
+            "per_case_latency": per_case_stats,
+            "sample_outputs": sample_outputs,
+            "failures": failures,
+        }
+
+
+def run_capacity_probe(
+    iterations_per_case: int = 25,
+    warmup_runs: int = 2,
+    cases: List[Dict[str, Any]] | None = None,
+) -> Dict[str, Any]:
+    probe = SovereignCapacityProbe(
+        iterations_per_case=iterations_per_case,
+        warmup_runs=warmup_runs,
+    )
+    return probe.run(cases=cases)
+
+
+if __name__ == "__main__":
+    report = run_capacity_probe(iterations_per_case=10, warmup_runs=1)
+    print(json.dumps(report, indent=2, ensure_ascii=False))
+