eval/run_eval.py

"""AlphaBrief evaluation harness.

Runs the **full graph** N times against live market data and reports:

* end-to-end success rate,
* post-verification numeric accuracy (must be 100%),
* tool-call error rate,
* average cost and latency per brief.

Plus two **fault-injection controls**, which are the point: a metric that only
ever says 100% is worthless unless you can also show the check fires. The
controls prove the verifier catches a corrupted figure and that a dead ticker
degrades instead of crashing.

Mock-clock safe: no assertion depends on wall-clock time or on a particular
market session, only on internal consistency between the brief and raw state.

    python eval/run_eval.py --runs 20

Writes ``eval/results.md``.
"""

from __future__ import annotations

import argparse
import asyncio
import os
import statistics
import sys
import tempfile
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parents[1]
API_ROOT = REPO_ROOT / "apps" / "api"
if str(API_ROOT) not in sys.path:
    sys.path.insert(0, str(API_ROOT))

DEFAULT_WATCHLIST = ["AAPL", "MSFT", "NVDA"]
RESULTS_PATH = Path(__file__).resolve().parent / "results.md"


@dataclass
class RunOutcome:
    """One evaluated run."""

    index: int
    run_id: str
    status: str
    verified: bool
    claims_checked: int = 0
    claims_matched: int = 0
    claims_mismatched: int = 0
    coverage: float = 0.0
    tool_calls: int = 0
    tool_call_errors: int = 0
    cost_usd: float = 0.0
    latency_ms: float = 0.0
    iterations: int = 0
    partial: bool = False
    error: str | None = None

    @property
    def succeeded(self) -> bool:
        return self.status in ("AWAITING_APPROVAL", "HUMAN_REVIEW") and self.claims_checked > 0


@dataclass
class EvalReport:
    """Aggregate results."""

    runs: list[RunOutcome] = field(default_factory=list)
    controls: dict[str, dict[str, Any]] = field(default_factory=dict)
    started_at: str = ""
    engine: str = ""
    watchlist: list[str] = field(default_factory=list)

    @property
    def success_rate(self) -> float:
        if not self.runs:
            return 0.0
        return sum(1 for r in self.runs if r.succeeded) / len(self.runs)

    @property
    def numeric_accuracy(self) -> float:
        """Fraction of claims in *verified* briefs confirmed by recomputation."""
        verified = [r for r in self.runs if r.verified]
        checked = sum(r.claims_checked for r in verified)
        matched = sum(r.claims_matched for r in verified)
        if checked == 0:
            return 0.0
        return matched / checked

    @property
    def tool_error_rate(self) -> float:
        total = sum(r.tool_calls for r in self.runs)
        errors = sum(r.tool_call_errors for r in self.runs)
        return (errors / total) if total else 0.0

    @property
    def avg_cost(self) -> float:
        return statistics.fmean([r.cost_usd for r in self.runs]) if self.runs else 0.0

    @property
    def avg_latency_ms(self) -> float:
        values = [r.latency_ms for r in self.runs if r.latency_ms > 0]
        return statistics.fmean(values) if values else 0.0

    @property
    def cold_latency_ms(self) -> float:
        """Run 1 — the only run that pays real provider round-trips."""
        first = next((run for run in self.runs if run.index == 1), None)
        return first.latency_ms if first else 0.0

    @property
    def warm_latency_ms(self) -> float:
        """Runs 2..N — orchestration overhead with the provider cache warm."""
        values = [run.latency_ms for run in self.runs if run.index > 1 and run.latency_ms > 0]
        return statistics.fmean(values) if values else 0.0

    @property
    def p95_latency_ms(self) -> float:
        values = sorted(r.latency_ms for r in self.runs if r.latency_ms > 0)
        if not values:
            return 0.0
        index = min(len(values) - 1, round(0.95 * (len(values) - 1)))
        return values[index]


def _configure_environment(database_url: str) -> None:
    """Free, deterministic, network-polite configuration for the eval."""
    os.environ.setdefault("LLM_ENGINE", "auto")
    os.environ["MCP_TRANSPORT"] = "inmemory"
    os.environ["MCP_SHARED_SERVER"] = "true"
    os.environ["DATABASE_URL"] = database_url
    os.environ.setdefault("PROVIDER_MIN_INTERVAL_SECONDS", "0.15")
    os.environ.setdefault("APPROVAL_TOKEN", "eval-local-token")
    os.environ.setdefault("LOG_LEVEL", "WARNING")
    # The eval approves every run to free its watchlist slot, and `Settings`
    # reads `../../.env` — so a working operator credential would make `make
    # eval` mail a real inbox once per scored run. Assigned rather than
    # setdefault: this has to beat the dotenv file.
    os.environ["SMTP_HOST"] = ""
    os.environ["SMTP_USERNAME"] = ""
    os.environ["SMTP_PASSWORD"] = ""
    os.environ["SMTP_FROM"] = ""
    os.environ["SMTP_TO"] = ""


async def _await_gate(service: Any, run_id: str, timeout_s: float = 180.0) -> dict[str, Any]:
    """Poll until the run leaves RUNNING."""
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        record = await service.get_run(run_id)
        if isinstance(record, dict) and record["status"] not in ("RUNNING", "QUEUED"):
            return record
        await asyncio.sleep(0.4)
    record = await service.get_run(run_id)
    fallback: dict[str, Any] = {"status": "TIMEOUT"}
    return record if isinstance(record, dict) else fallback


async def _evaluate_once(service: Any, index: int, tickers: list[str]) -> RunOutcome:
    from app.models.run import RunMode

    started = time.perf_counter()
    try:
        run_id = await service.start_run(tickers=tickers, mode=RunMode.STANDARD, trigger="eval")
    except Exception as exc:  # noqa: BLE001 - an eval must record failures, not die
        return RunOutcome(
            index=index, run_id="-", status="START_FAILED", verified=False, error=str(exc)
        )

    record = await _await_gate(service, run_id)
    gate = await service.gate_payload(run_id)
    report = (gate or {}).get("verification") or {}
    events = await service.bus.history(run_id)
    tool_events = [e for e in events if str(e.kind) == "mcp.tool_call"]
    tool_errors = sum(1 for e in tool_events if not e.payload.get("ok", True))

    return RunOutcome(
        index=index,
        run_id=run_id,
        status=str(record.get("status")),
        verified=bool(report.get("ok")),
        claims_checked=int(report.get("checked_claims", 0)),
        claims_matched=int(report.get("matched", 0)),
        claims_mismatched=int(report.get("mismatched", 0)),
        coverage=float(report.get("coverage", 0.0)),
        tool_calls=len(tool_events),
        tool_call_errors=tool_errors,
        cost_usd=float(record.get("cost_usd", 0.0)),
        latency_ms=float(record.get("latency_ms", (time.perf_counter() - started) * 1000)),
        iterations=int(record.get("iterations", 0)),
        partial=bool(record.get("partial", False)),
    )


async def _run_control(service: Any, mode_name: str, tickers: list[str]) -> dict[str, Any]:
    """Fault-injection control: proves the guarantees are not vacuous."""
    from app.models.run import RunMode

    mode = RunMode(mode_name)
    run_id = await service.start_run(tickers=tickers, mode=mode, trigger="eval-control")
    record = await _await_gate(service, run_id)
    gate = await service.gate_payload(run_id)
    report = (gate or {}).get("verification") or {}
    brief = (gate or {}).get("brief") or {}
    return {
        "mode": mode_name,
        "run_id": run_id,
        "status": str(record.get("status")),
        "verified": bool(report.get("ok")),
        "claims_checked": int(report.get("checked_claims", 0)),
        "claims_mismatched": int(report.get("mismatched", 0)),
        "partial": bool(brief.get("partial", False)),
        "snapshot_rows": len(brief.get("snapshot", [])),
        "errors": int(record.get("error_count", 0)),
        "crashed": str(record.get("status")) == "FAILED",
    }


async def run_eval(count: int, tickers: list[str]) -> EvalReport:
    """Execute `count` scored runs plus the fault-injection controls."""
    from app.core.settings import get_settings
    from app.services.runner import RunService

    settings = get_settings()
    report = EvalReport(
        started_at=datetime.now(UTC).isoformat(timespec="seconds"),
        engine=settings.resolved_engine,
        watchlist=tickers,
    )
    service = RunService()

    for index in range(1, count + 1):
        outcome = await _evaluate_once(service, index, tickers)
        report.runs.append(outcome)
        flag = "ok " if outcome.succeeded and outcome.verified else "FAIL"
        print(
            f"  [{index:>2}/{count}] {flag} {outcome.status:<18} "
            f"claims {outcome.claims_matched}/{outcome.claims_checked} "
            f"tools {outcome.tool_calls} "
            f"{outcome.latency_ms / 1000:.2f}s ${outcome.cost_usd:.4f}"
        )
        # Release the watchlist slot so the next run may start.
        await _release(service, outcome.run_id)

    print("\n  fault-injection controls:")
    for mode_name, control_tickers in (
        ("demo_mismatch", tickers[:1]),
        ("demo_fault", tickers[:1]),
    ):
        control = await _run_control(service, mode_name, control_tickers)
        report.controls[mode_name] = control
        print(
            f"   {mode_name:<14} status={control['status']:<14} "
            f"verified={control['verified']} mismatched={control['claims_mismatched']} "
            f"crashed={control['crashed']}"
        )
        await _release(service, control["run_id"])

    await service.shutdown()
    return report


async def _release(service: Any, run_id: str) -> None:
    """Approve a paused run so its watchlist slot is freed for the next one."""
    from app.models.run import HumanDecision

    try:
        await service.submit_decision(run_id, HumanDecision(action="approve", reviewer="eval"))
    except Exception:  # noqa: BLE001 - already terminal, nothing to release
        return


def render_results(report: EvalReport) -> str:
    """Render `eval/results.md`."""
    accuracy = report.numeric_accuracy
    lines: list[str] = [
        "# AlphaBrief — Evaluation Results",
        "",
        f"- **Generated:** {report.started_at}",
        f"- **Runs scored:** {len(report.runs)}",
        f"- **Watchlist:** {', '.join(report.watchlist)}",
        f"- **Model engine:** `{report.engine}`",
        "- **Data:** live yfinance prices and live Yahoo/Google RSS headlines. "
        "No synthetic fixtures.",
        "",
        "## Headline metrics",
        "",
        "| Metric | Result | Target |",
        "| --- | ---: | ---: |",
        f"| End-to-end success rate | {report.success_rate * 100:.1f}% | ≥ 95% |",
        f"| Post-verification numeric accuracy | {accuracy * 100:.2f}% | 100% |",
        f"| Tool-call error rate | {report.tool_error_rate * 100:.2f}% | ≤ 5% |",
        f"| Average cost per brief | ${report.avg_cost:.4f} | ≤ $0.15 |",
        f"| Cold-start latency (run 1, live provider calls) | {report.cold_latency_ms / 1000:.2f}s | — |",
        f"| Warm latency per brief (runs 2–N, cached providers) | {report.warm_latency_ms / 1000:.2f}s | — |",
        f"| p95 latency per brief | {report.p95_latency_ms / 1000:.2f}s | — |",
        "",
        "### Reading these numbers honestly",
        "",
        "- **Latency.** The harness shares one MCP server, and therefore one provider",
        "  cache, across the whole session so that twenty runs are polite to free",
        "  providers. Run 1 pays the real yfinance and RSS round-trips; runs 2–N measure",
        "  orchestration overhead only. Both are reported rather than blended, because",
        "  the blended average would flatter the system.",
        f"- **Cost.** `${report.avg_cost:.4f}` is the measured spend on the",
        f"  `{report.engine}` engine. On the deterministic engine that is genuinely zero —",
        "  it makes no Anthropic calls. With `ANTHROPIC_API_KEY` set, the same graph runs",
        "  on Haiku 4.5 + Sonnet 4.6 at roughly $0.05–0.15 per full 5-ticker brief;",
        "  the budget guard aborts the run rather than exceed `TOKEN_BUDGET_USD`.",
        "- **Accuracy.** This is not a model score. It is the fraction of published",
        "  figures that a second, independent implementation reproduced from the raw",
        "  price bars. A figure that fails is not published at all.",
        "",
        "**Post-verification numeric accuracy is 100% by construction, not by luck.**",
        "Every figure is minted from a tool-computed metric, then independently",
        "recomputed from the raw price bars by `app/graph/recompute.py` before the",
        "brief can reach the approval gate. A figure that fails is not published —",
        "it triggers one regeneration and then `HUMAN_REVIEW`.",
        "",
        "## Fault-injection controls",
        "",
        "A verifier that always passes proves nothing. These runs deliberately break",
        "the system to show the checks actually fire.",
        "",
        "| Control | Status | Verified | Mismatches caught | Crashed |",
        "| --- | --- | --- | ---: | --- |",
    ]
    for name, control in report.controls.items():
        lines.append(
            f"| `{name}` | {control['status']} | {control['verified']} | "
            f"{control['claims_mismatched']} | {control['crashed']} |"
        )
    lines += [
        "",
        "- `demo_mismatch` corrupts one claim after assembly. Expected: the verifier",
        "  catches it, one regeneration is attempted, and the run lands in",
        "  `HUMAN_REVIEW` rather than being delivered.",
        "- `demo_fault` injects an unresolvable ticker. Expected: the run completes",
        "  with per-ticker errors and a brief explicitly marked partial — never a crash.",
        "",
        "## Per-run detail",
        "",
        "| # | Status | Verified | Claims matched | Tool calls | Iterations | Latency | Cost |",
        "| ---: | --- | --- | ---: | ---: | ---: | ---: | ---: |",
    ]
    for run in report.runs:
        lines.append(
            f"| {run.index} | {run.status} | {run.verified} | "
            f"{run.claims_matched}/{run.claims_checked} | {run.tool_calls} | "
            f"{run.iterations} | {run.latency_ms / 1000:.2f}s | ${run.cost_usd:.4f} |"
        )
    lines += [
        "",
        "---",
        "",
        "Reproduce with `python eval/run_eval.py --runs "
        f"{len(report.runs)}`. Figures move between runs because the market moves;",
        "the invariants (accuracy, graceful degradation, termination) do not.",
        "",
    ]
    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(description="Evaluate the AlphaBrief pipeline.")
    parser.add_argument("--runs", type=int, default=20, help="scored runs (default: 20)")
    parser.add_argument(
        "--tickers",
        type=str,
        default=",".join(DEFAULT_WATCHLIST),
        help="comma-separated watchlist",
    )
    parser.add_argument(
        "--database-url",
        type=str,
        default="",
        help="override the eval database (default: a temporary SQLite file)",
    )
    args = parser.parse_args()

    tickers = [t.strip().upper() for t in args.tickers.split(",") if t.strip()]
    database_url = args.database_url or f"sqlite:///{Path(tempfile.mkdtemp()) / 'eval.db'}"
    _configure_environment(database_url)

    print(f"AlphaBrief eval — {args.runs} runs over {', '.join(tickers)}\n")
    report = asyncio.run(run_eval(args.runs, tickers))

    RESULTS_PATH.write_text(render_results(report), encoding="utf-8")
    accuracy = report.numeric_accuracy

    print(f"\n  success rate            {report.success_rate * 100:.1f}%")
    print(f"  numeric accuracy        {accuracy * 100:.2f}%")
    print(f"  tool-call error rate    {report.tool_error_rate * 100:.2f}%")
    print(f"  avg cost / brief        ${report.avg_cost:.4f} ({report.engine} engine)")
    print(f"  cold-start latency      {report.cold_latency_ms / 1000:.2f}s (live providers)")
    print(f"  warm latency / brief    {report.warm_latency_ms / 1000:.2f}s (cached providers)")
    print(f"\n  wrote {RESULTS_PATH}")

    mismatch_control = report.controls.get("demo_mismatch", {})
    fault_control = report.controls.get("demo_fault", {})
    ok = (
        accuracy >= 1.0
        and report.success_rate >= 0.95
        and mismatch_control.get("claims_mismatched", 0) > 0
        and not mismatch_control.get("verified", True)
        and not fault_control.get("crashed", True)
    )
    print("\n  RESULT:", "PASS" if ok else "FAIL")
    return 0 if ok else 1


if __name__ == "__main__":
    raise SystemExit(main())