server/app.py

from __future__ import annotations

import json
import os
from pathlib import Path
import importlib
import sys
from typing import Any

if __package__ in {None, ""}:
    repo_root = str(Path(__file__).resolve().parents[1])
    if repo_root not in sys.path:
        sys.path.insert(0, repo_root)

import uvicorn

from models import LedgerShieldAction, LedgerShieldObservation
from openenv_compat import create_fastapi_app

if __package__ in {None, ""}:
    from server.environment import LedgerShieldEnvironment
    from server.certify import build_certify_report
    from server.visualization import build_controlbench_visualization
else:
    from .environment import LedgerShieldEnvironment
    from .certify import build_certify_report
    from .visualization import build_controlbench_visualization


def _load_benchmark_report_module():
    try:
        return importlib.import_module("benchmark_report")
    except ModuleNotFoundError:
        return None


def build_app():
    env = LedgerShieldEnvironment()
    app = create_fastapi_app(env, LedgerShieldAction, LedgerShieldObservation)
    runtime_report_cache: dict[str, Any] | None = None

    def _runtime_report_preview(benchmark_report: Any) -> dict[str, Any]:
        """Build a small in-memory report when packaged artifacts are absent."""
        nonlocal runtime_report_cache
        if runtime_report_cache is not None:
            return runtime_report_cache

        controlbench_length = max(
            1,
            int(os.getenv("LEDGERSHIELD_RUNTIME_REPORT_CONTROLBENCH_CASES", "12") or 12),
        )
        holdout_seed = int((benchmark_report.DEFAULT_HOLDOUT_SEEDS or [2026])[0])
        report = benchmark_report.build_report(
            holdout_seeds=[holdout_seed],
            variants_per_case=1,
            pass_threshold=benchmark_report.DEFAULT_PASS_THRESHOLD,
            pass_k=benchmark_report.DEFAULT_PASS_K,
            temperature=benchmark_report.DEFAULT_TEMPERATURE,
            client=None,
            model_name="",
            controlbench_sequence_length=controlbench_length,
        )
        report["runtime_preview"] = True
        report["artifact_note"] = (
            "Generated in memory because no benchmark report artifact was found. "
            "Run benchmark_report.py to write the full artifact set."
        )
        report.setdefault("evaluation_protocol", {})["runtime_preview"] = True
        runtime_report_cache = report
        return report

    def _latest_report() -> dict[str, Any]:
        benchmark_report = _load_benchmark_report_module()
        if benchmark_report is None:
            return {}
        report_path = benchmark_report.DEFAULT_REPORT_PATH
        if report_path.exists():
            return json.loads(report_path.read_text(encoding="utf-8"))
        try:
            return _runtime_report_preview(benchmark_report)
        except Exception as exc:  # pragma: no cover - defensive runtime fallback
            return {
                "benchmark": "ledgershield-controlbench-v1",
                "generated_at": None,
                "runtime_preview": False,
                "note": f"No benchmark report artifact found and runtime preview generation failed: {exc}",
            }

    @app.get("/leaderboard")
    def leaderboard() -> dict[str, Any]:
        benchmark_report = _load_benchmark_report_module()
        if benchmark_report is None:
            return {
                "benchmark": "ledgershield-controlbench-v1",
                "generated_at": None,
                "note": "benchmark_report.py is unavailable in this runtime image.",
                "entries": [],
            }
        leaderboard_path = benchmark_report.DEFAULT_LEADERBOARD_PATH
        if leaderboard_path.exists():
            return benchmark_report.load_leaderboard_payload()

        report = _latest_report()
        if isinstance(report.get("public_benchmark"), dict):
            protocol = report.get("evaluation_protocol", {}) or {}
            entry = benchmark_report.build_leaderboard_entry(
                report,
                model_name=protocol.get("model_name", benchmark_report.DETERMINISTIC_BASELINE_MODEL),
                agent_type=protocol.get("agent_type", "deterministic-policy"),
            )
            return {
                "benchmark": report.get("benchmark", "ledgershield-controlbench-v1"),
                "generated_at": report.get("generated_at"),
                "entries": [entry],
                "runtime_preview": bool(report.get("runtime_preview")),
                "note": report.get("artifact_note", "Leaderboard derived from the runtime benchmark preview."),
            }

        return {
            "benchmark": "ledgershield-controlbench-v1",
            "generated_at": None,
            "entries": [],
            "note": report.get("note", "No leaderboard artifact or runtime benchmark report is available."),
        }

    @app.get("/benchmark-report")
    def latest_benchmark_report() -> dict[str, Any]:
        benchmark_report = _load_benchmark_report_module()
        if benchmark_report is None:
            return {
                "benchmark": "ledgershield-controlbench-v1",
                "generated_at": None,
                "note": "benchmark_report.py is unavailable in this runtime image.",
            }
        report_path = benchmark_report.DEFAULT_REPORT_PATH
        if report_path.exists():
            return json.loads(report_path.read_text(encoding="utf-8"))
        return _latest_report()

    @app.post("/certify")
    def certify(payload: dict[str, Any] | None = None) -> dict[str, Any]:
        return build_certify_report(
            payload or {},
            benchmark_report=_latest_report(),
            institutional_memory=env.institutional_memory(),
        )

    @app.get("/certify-summary")
    def certify_summary() -> dict[str, Any]:
        return build_certify_report(
            {},
            benchmark_report=_latest_report(),
            institutional_memory=env.institutional_memory(),
        )

    @app.get("/controlbench-visualization")
    def controlbench_visualization() -> dict[str, Any]:
        report = _latest_report()
        if report and isinstance(report.get("controlbench_visualization"), dict):
            return report["controlbench_visualization"]
        return build_controlbench_visualization(report, institutional_memory=env.institutional_memory())

    @app.get("/controlbench-summary")
    def controlbench_summary() -> dict[str, Any]:
        benchmark_report = _load_benchmark_report_module()
        if benchmark_report is not None:
            controlbench_report_path = getattr(benchmark_report, "DEFAULT_CONTROLBENCH_REPORT_PATH", None)
            if isinstance(controlbench_report_path, Path) and controlbench_report_path.exists():
                return json.loads(controlbench_report_path.read_text(encoding="utf-8"))
            report_path = benchmark_report.DEFAULT_REPORT_PATH
            if report_path.exists():
                report = json.loads(report_path.read_text(encoding="utf-8"))
                if isinstance(report.get("controlbench_quarter"), dict):
                    return report["controlbench_quarter"]
        memory = env.institutional_memory()
        return {
            "benchmark": "ledgershield-controlbench-v1",
            "note": "No ControlBench report artifact found; returning live institutional memory summary.",
            "controlbench_summary": memory.get("controlbench_summary", {}),
            "loss_surface": (memory.get("loss_ledger", {}) or {}).get("loss_surface", {}),
            "calibration_gate": memory.get("calibration_gate", {}),
        }

    @app.get("/human-baseline-summary")
    def human_baseline_summary() -> dict[str, Any]:
        benchmark_report = _load_benchmark_report_module()
        if benchmark_report is not None:
            report_path = benchmark_report.DEFAULT_REPORT_PATH
            if report_path.exists():
                report = json.loads(report_path.read_text(encoding="utf-8"))
                if isinstance(report.get("human_baseline_track"), dict):
                    return report["human_baseline_track"]
        try:
            if __package__ in {None, ""}:
                from server.human_baseline import load_human_baseline_summary
            else:
                from .human_baseline import load_human_baseline_summary
        except ModuleNotFoundError:
            return {
                "track": "human_baseline",
                "note": "Human baseline loader is unavailable in this runtime image.",
            }
        return load_human_baseline_summary()

    @app.get("/institutional-memory")
    def institutional_memory() -> dict[str, Any]:
        return env.institutional_memory()

    @app.post("/institutional-reset")
    def institutional_reset() -> dict[str, Any]:
        return env.reset_institutional_memory()

    return app


app = build_app()


def main() -> None:
    host = os.getenv("HOST", "0.0.0.0")
    port = int(os.getenv("PORT", "8000"))
    uvicorn.run("server.app:app", host=host, port=port, reload=False)


if __name__ == "__main__":
    main()