sovereign_benchmark.py

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))