sovereign_capacity_report.py

# sovereign_capacity_report.py
from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

from sovereign_capacity_probe import run_capacity_probe


CAPACITY_OUTPUT_DIR = "sovereign_capacity_outputs"


def _utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _ensure_dir(path: str) -> None:
    os.makedirs(path, exist_ok=True)


def _safe(v: Any, default: Any = "") -> Any:
    return v if v is not None else default


def _pct(v: Any) -> str:
    try:
        return f"{float(v):.2f}%"
    except Exception:
        return "0.00%"


def _ms(v: Any) -> str:
    try:
        return f"{float(v):.3f} ms"
    except Exception:
        return "0.000 ms"


def _rps(v: Any) -> str:
    try:
        return f"{float(v):.3f} rps"
    except Exception:
        return "0.000 rps"


def _render_summary(report: Dict[str, Any]) -> str:
    summary = report.get("summary", {})
    overall = summary.get("overall_latency", {})
    counts = summary.get("decision_counts", {})
    pcts = summary.get("decision_percentages", {})
    cfg = report.get("config", {})

    return f"""# Sovereign Capacity Report

Generated at: {_safe(report.get("generated_at"), _utc_now())}

## 1. Probe Identity
- Probe Name: {_safe(report.get("probe_name"), "Sovereign Capacity Probe")}
- Case Count: {_safe(cfg.get("case_count"), 0)}
- Warmup Runs: {_safe(cfg.get("warmup_runs"), 0)}
- Iterations Per Case: {_safe(cfg.get("iterations_per_case"), 0)}
- Total Expected Requests: {_safe(cfg.get("total_expected_requests"), 0)}

## 2. Executive Summary
- Successful Requests: {_safe(summary.get("successful_requests"), 0)}
- Failed Requests: {_safe(summary.get("failed_requests"), 0)}
- Total Wall Time: {_safe(summary.get("total_wall_seconds"), 0)} sec
- Approx Throughput: {_rps(summary.get("throughput_rps", 0))}
- Overall Min Latency: {_ms(overall.get("min_ms"))}
- Overall Avg Latency: {_ms(overall.get("avg_ms"))}
- Overall Median Latency: {_ms(overall.get("median_ms"))}
- Overall P95 Latency: {_ms(overall.get("p95_ms"))}
- Overall P99 Latency: {_ms(overall.get("p99_ms"))}
- Overall Max Latency: {_ms(overall.get("max_ms"))}
- Overall Standard Deviation: {_ms(overall.get("stdev_ms"))}

## 3. Decision Distribution Under Load
- ALLOW Count: {_safe(counts.get("ALLOW"), 0)} ({_pct(pcts.get("ALLOW", 0))})
- FREEZE Count: {_safe(counts.get("FREEZE"), 0)} ({_pct(pcts.get("FREEZE", 0))})
- BLOCK Count: {_safe(counts.get("BLOCK"), 0)} ({_pct(pcts.get("BLOCK", 0))})
- UNKNOWN Count: {_safe(counts.get("UNKNOWN"), 0)} ({_pct(pcts.get("UNKNOWN", 0))})
"""


def _render_case_latency(report: Dict[str, Any]) -> str:
    blocks = ["## 4. Per-Scenario Latency Under Load"]
    per_case = report.get("per_case_latency", {})

    if not per_case:
        blocks.append("- No per-case data available.")
        return "\n".join(blocks)

    for case_name, stats in per_case.items():
        blocks.append(
            f"""
### {case_name}
- Count: {_safe(stats.get("count"), 0)}
- Min: {_ms(stats.get("min_ms"))}
- Avg: {_ms(stats.get("avg_ms"))}
- Median: {_ms(stats.get("median_ms"))}
- P95: {_ms(stats.get("p95_ms"))}
- P99: {_ms(stats.get("p99_ms"))}
- Max: {_ms(stats.get("max_ms"))}
- Stdev: {_ms(stats.get("stdev_ms"))}
""".rstrip()
        )

    return "\n\n".join(blocks)


def _render_failures(report: Dict[str, Any]) -> str:
    failures = report.get("failures", [])
    lines = ["## 5. Failures"]

    if not failures:
        lines.append("- No capacity-probe failures were recorded.")
        return "\n".join(lines)

    lines.append(f"- Failure Count: {len(failures)}")
    for item in failures[:25]:
        lines.append(
            f"- Case: {item.get('case')} | Iteration: {item.get('iteration')} | Error: {item.get('error')}"
        )
    if len(failures) > 25:
        lines.append(f"- Additional failures omitted: {len(failures) - 25}")

    return "\n".join(lines)


def _render_samples(report: Dict[str, Any]) -> str:
    sample_outputs = report.get("sample_outputs", {})
    lines = ["## 6. Sample Outputs Under Load"]

    if not sample_outputs:
        lines.append("- No sample outputs available.")
        return "\n".join(lines)

    for case_name, sample in sample_outputs.items():
        lines.append(f"### {case_name}")
        lines.append(f"- Decision: {_safe(sample.get('decision'), 'UNKNOWN')}")
        lines.append("```json")
        lines.append(json.dumps(sample.get("sample_result", {}), ensure_ascii=False, indent=2))
        lines.append("```")

    return "\n".join(lines)


def _render_verdict(report: Dict[str, Any]) -> str:
    summary = report.get("summary", {})
    overall = summary.get("overall_latency", {})
    failures = int(summary.get("failed_requests", 0) or 0)
    p95 = float(overall.get("p95_ms", 0.0) or 0.0)
    p99 = float(overall.get("p99_ms", 0.0) or 0.0)
    throughput = float(summary.get("throughput_rps", 0.0) or 0.0)

    if failures == 0 and p95 <= 100 and p99 <= 200:
        verdict = "Capacity posture is stable for current probe volume, with acceptable runtime behavior under repeated load."
    elif failures == 0:
        verdict = "Capacity posture is operationally viable, but latency under load should be further optimized before strong bank-facing performance claims."
    else:
        verdict = "Capacity probe indicates additional hardening or stability work is required before bank-facing performance commitments."

    return f"""## 7. Preliminary Capacity Verdict
{verdict}

## 8. Notes
- This report reflects current repeated-request behavior, not final production-scale distributed load testing.
- Throughput in this report is an approximate runtime throughput for the current execution path.
- Bank-facing performance claims should be based on repeated probes, tuning iterations, and later higher-intensity validation.
- This report is suitable as a technical capacity artifact, not yet as a final production SLA statement.
"""


def build_capacity_markdown(report: Dict[str, Any]) -> str:
    parts = [
        _render_summary(report),
        _render_case_latency(report),
        _render_failures(report),
        _render_samples(report),
        _render_verdict(report),
    ]
    return "\n\n".join(parts).strip() + "\n"


def save_capacity_report(
    report: Dict[str, Any],
    output_dir: str = CAPACITY_OUTPUT_DIR,
    base_name: Optional[str] = None,
) -> Dict[str, str]:
    _ensure_dir(output_dir)

    ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    base = base_name or f"sovereign_capacity_report_{ts}"

    json_path = os.path.join(output_dir, f"{base}.json")
    md_path = os.path.join(output_dir, f"{base}.md")

    with open(json_path, "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)

    markdown = build_capacity_markdown(report)
    with open(md_path, "w", encoding="utf-8") as f:
        f.write(markdown)

    return {
        "json_report": json_path,
        "markdown_report": md_path,
    }


def generate_capacity_report(
    iterations_per_case: int = 25,
    warmup_runs: int = 2,
    output_dir: str = CAPACITY_OUTPUT_DIR,
    base_name: Optional[str] = None,
) -> Dict[str, Any]:
    report = run_capacity_probe(
        iterations_per_case=iterations_per_case,
        warmup_runs=warmup_runs,
    )

    files = save_capacity_report(
        report=report,
        output_dir=output_dir,
        base_name=base_name,
    )

    return {
        "ok": True,
        "generated_at": _utc_now(),
        "report": report,
        "files": files,
    }


if __name__ == "__main__":
    out = generate_capacity_report(iterations_per_case=10, warmup_runs=1)
    print(json.dumps(out, indent=2, ensure_ascii=False))