import time import json import statistics from typing import Any, Dict, List from sovereign_core import run_sentinel def _percentile(sorted_values: List[float], p: float) -> float: if not sorted_values: return 0.0 if p <= 0: return sorted_values[0] if p >= 100: return sorted_values[-1] k = (len(sorted_values) - 1) * (p / 100.0) f = int(k) c = min(f + 1, len(sorted_values) - 1) if f == c: return sorted_values[f] d0 = sorted_values[f] * (c - k) d1 = sorted_values[c] * (k - f) return d0 + d1 def _single_run() -> float: start = time.perf_counter() run_sentinel( engine_name="benchmark_engine", parent_model="test_agent", model_version="v1", data_tags="pii", risk_level="medium", notes="benchmark_run", access_key="", delegation_token="", ) end = time.perf_counter() return (end - start) * 1000.0 def run_test(iterations: int = 200, warmup: int = 10) -> Dict[str, Any]: iterations = max(10, int(iterations)) warmup = max(0, int(warmup)) warmup_latencies: List[float] = [] measured_latencies: List[float] = [] for _ in range(warmup): warmup_latencies.append(_single_run()) for _ in range(iterations): measured_latencies.append(_single_run()) measured_latencies.sort() result = { "benchmark_name": "sovereign_runtime_latency", "iterations": iterations, "warmup_runs": warmup, "unit": "ms", "summary": { "min_ms": round(min(measured_latencies), 4), "avg_ms": round(statistics.mean(measured_latencies), 4), "median_ms": round(statistics.median(measured_latencies), 4), "p95_ms": round(_percentile(measured_latencies, 95), 4), "p99_ms": round(_percentile(measured_latencies, 99), 4), "max_ms": round(max(measured_latencies), 4), }, "warmup_summary": { "avg_ms": round(statistics.mean(warmup_latencies), 4) if warmup_latencies else 0.0, "max_ms": round(max(warmup_latencies), 4) if warmup_latencies else 0.0, }, } return result if __name__ == "__main__": result = run_test(iterations=200, warmup=10) print(json.dumps(result, indent=2, ensure_ascii=False))