apps/api/app/graph/verify.py

"""The verification node — deterministic, never an LLM.

Every numeric claim in the brief is recomputed from the raw price bars in state
by :mod:`app.graph.recompute`, every quoted headline is matched verbatim against
the news actually retrieved, and the brief's structure is checked against the
run. Numbers match to the cent or nothing ships.

Outcomes:

* all checks pass → the gate
* any mismatch, first time → exactly one writer regeneration
* any mismatch, second time → ``HUMAN_REVIEW`` (flagged, never auto-delivered)
"""

from __future__ import annotations

import re
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field

from app.core.events import EventKind
from app.graph.context import current_context
from app.graph.recompute import recompute
from app.graph.state import RunState, is_partial, usable_tickers
from app.models.brief import Brief
from app.models.market import METRIC_UNITS, MetricUnit
from app.models.run import RunError, RunStatus

#: Absolute tolerance per unit. USD is "to the cent"; percentages are checked far
#: tighter than the 4-decimal rounding the tool applies on the way out.
TOLERANCES: dict[MetricUnit, float] = {
    "usd": 0.005,
    "percent": 0.001,
    "ratio": 0.001,
    "score": 0.000001,
}

_WHITESPACE = re.compile(r"\s+")

ClaimStatus = Literal["match", "mismatch", "unverifiable"]


def _normalise_quote(text: str) -> str:
    return _WHITESPACE.sub(" ", (text or "").strip()).casefold()


class ClaimCheck(BaseModel):
    """One recomputed numeric claim."""

    model_config = ConfigDict(extra="forbid")

    claim_id: str
    ticker: str
    metric: str
    unit: MetricUnit
    claimed: float
    recomputed: float | None
    delta: float | None
    tolerance: float
    status: ClaimStatus
    detail: str


class QuoteCheck(BaseModel):
    """One verbatim third-party quotation."""

    model_config = ConfigDict(extra="forbid")

    ticker: str
    field: str
    text: str
    status: Literal["match", "mismatch"]
    detail: str


class VerificationReport(BaseModel):
    """The full verification pass, rendered claim-by-claim in the UI."""

    model_config = ConfigDict(extra="forbid")

    ok: bool
    checked_claims: int = 0
    matched: int = 0
    mismatched: int = 0
    unverifiable: int = 0
    coverage: float = 0.0
    claim_checks: list[ClaimCheck] = Field(default_factory=list)
    quote_checks: list[QuoteCheck] = Field(default_factory=list)
    structural_issues: list[str] = Field(default_factory=list)
    structural_warnings: list[str] = Field(default_factory=list)

    @property
    def failures(self) -> list[str]:
        out = [
            f"{c.claim_id} ({c.ticker}.{c.metric}): {c.detail}"
            for c in self.claim_checks
            if c.status != "match"
        ]
        out += [
            f"{q.ticker}.{q.field}: {q.detail}" for q in self.quote_checks if q.status != "match"
        ]
        out += self.structural_issues
        return out


def verify_brief(brief: Brief, state: RunState) -> VerificationReport:
    """Pure verification: recompute every claim from raw state.

    Importable and testable in isolation — no graph, no network, no model.
    """
    prices = state.get("prices", {})
    fundamentals = state.get("fundamentals", {})
    sentiment = state.get("sentiment", {})
    news = state.get("news", {})
    watchlist = set(state.get("tickers", []))

    structural: list[str] = []
    warnings: list[str] = []

    # ---------------------------------------------------------- structure ---
    expected_date = state.get("session_date")
    if expected_date and brief.generated_for != expected_date:
        structural.append(
            f"brief.generated_for '{brief.generated_for}' does not match the run's "
            f"session date '{expected_date}'"
        )
    if set(brief.watchlist) != watchlist:
        structural.append(
            f"brief.watchlist {sorted(brief.watchlist)} does not match the run watchlist "
            f"{sorted(watchlist)}"
        )

    defined = brief.claims_by_id()
    if len(defined) != len(brief.claims):
        structural.append("duplicate claim_id values in brief.claims")

    referenced = brief.referenced_claim_ids()
    missing_refs = sorted(referenced - defined.keys())
    if missing_refs:
        structural.append(f"narrative references undefined claims: {missing_refs}")
    orphans = sorted(defined.keys() - referenced)
    if orphans:
        warnings.append(f"claims defined but never cited: {orphans}")

    expected_rows = set(usable_tickers(state))
    row_tickers = [row.ticker for row in brief.snapshot]
    if len(row_tickers) != len(set(row_tickers)):
        structural.append("duplicate tickers in the snapshot table")
    unexpected = sorted(set(row_tickers) - watchlist)
    if unexpected:
        structural.append(f"snapshot contains tickers outside the watchlist: {unexpected}")
    dropped = sorted(expected_rows - set(row_tickers))
    if dropped:
        structural.append(f"snapshot omits tickers that have market data: {dropped}")

    for row in brief.snapshot:
        for claim_id in row.claim_ids():
            claim = defined.get(claim_id)
            if claim and claim.ticker != row.ticker:
                structural.append(
                    f"snapshot row {row.ticker} cites {claim_id}, which belongs to {claim.ticker}"
                )

    if brief.partial != is_partial(state):
        structural.append(
            f"brief.partial={brief.partial} but the run "
            f"{'has' if is_partial(state) else 'has no'} incomplete tickers"
        )

    # ------------------------------------------------------------- claims ---
    claim_checks: list[ClaimCheck] = []
    for claim in brief.claims:
        unit = METRIC_UNITS.get(claim.metric, claim.unit)
        tolerance = TOLERANCES[unit]
        if claim.ticker not in watchlist:
            claim_checks.append(
                ClaimCheck(
                    claim_id=claim.claim_id,
                    ticker=claim.ticker,
                    metric=claim.metric,
                    unit=unit,
                    claimed=claim.value,
                    recomputed=None,
                    delta=None,
                    tolerance=tolerance,
                    status="mismatch",
                    detail=f"ticker '{claim.ticker}' is not in the run watchlist",
                )
            )
            continue
        if claim.unit != unit:
            structural.append(
                f"{claim.claim_id} declares unit '{claim.unit}' but "
                f"{claim.metric} is measured in '{unit}'"
            )

        recomputed = recompute(
            claim.metric,
            history=prices.get(claim.ticker),
            fundamentals=fundamentals.get(claim.ticker),
            sentiment=sentiment.get(claim.ticker),
        )
        if recomputed is None:
            claim_checks.append(
                ClaimCheck(
                    claim_id=claim.claim_id,
                    ticker=claim.ticker,
                    metric=claim.metric,
                    unit=unit,
                    claimed=claim.value,
                    recomputed=None,
                    delta=None,
                    tolerance=tolerance,
                    status="unverifiable",
                    detail="raw state cannot support recomputing this figure",
                )
            )
            continue

        delta = abs(claim.value - recomputed)
        matched = delta <= tolerance
        claim_checks.append(
            ClaimCheck(
                claim_id=claim.claim_id,
                ticker=claim.ticker,
                metric=claim.metric,
                unit=unit,
                claimed=claim.value,
                recomputed=round(recomputed, 6),
                delta=round(delta, 8),
                tolerance=tolerance,
                status="match" if matched else "mismatch",
                detail=(
                    "recomputed from raw price bars"
                    if matched
                    else f"claimed {claim.value} vs recomputed {recomputed:.6f} (Δ {delta:.6f})"
                ),
            )
        )

    # ------------------------------------------------------------- quotes ---
    quote_checks: list[QuoteCheck] = []
    for ticker, field_name, text in brief.quoted_texts():
        feed = news.get(ticker)
        available = {_normalise_quote(item.title) for item in (feed.items if feed else [])}
        matched = _normalise_quote(text) in available
        quote_checks.append(
            QuoteCheck(
                ticker=ticker,
                field=field_name,
                text=text[:200],
                status="match" if matched else "mismatch",
                detail=(
                    "verbatim match against retrieved headlines"
                    if matched
                    else "quoted text does not appear in the headlines retrieved for this ticker"
                ),
            )
        )

    matched_count = sum(1 for c in claim_checks if c.status == "match")
    mismatched = sum(1 for c in claim_checks if c.status == "mismatch")
    unverifiable = sum(1 for c in claim_checks if c.status == "unverifiable")
    quote_failures = sum(1 for q in quote_checks if q.status == "mismatch")

    coverage = 1.0 if not claim_checks else matched_count / len(claim_checks)
    ok = mismatched == 0 and unverifiable == 0 and quote_failures == 0 and not structural

    return VerificationReport(
        ok=ok,
        checked_claims=len(claim_checks),
        matched=matched_count,
        mismatched=mismatched,
        unverifiable=unverifiable,
        coverage=round(coverage, 4),
        claim_checks=claim_checks,
        quote_checks=quote_checks,
        structural_issues=structural,
        structural_warnings=warnings,
    )


async def verify_node(state: RunState) -> dict[str, Any]:
    """Graph node: run verification and stream each claim to the UI."""
    ctx = current_context()
    brief = state.get("brief")

    if brief is None:
        await ctx.emit(EventKind.VERIFY_COMPLETED, "No brief to verify", {"ok": False})
        return {
            "verification": VerificationReport(
                ok=False, structural_issues=["writer produced no brief"]
            ).model_dump(),
            "errors": [RunError(stage="verify", message="writer produced no brief")],
            "status": RunStatus.HUMAN_REVIEW,
        }

    await ctx.emit(
        EventKind.VERIFY_STARTED,
        f"Recomputing {len(brief.claims)} numeric claims from raw state",
        {"claims": len(brief.claims)},
    )

    with ctx.tracer.step(
        "verify",
        run_id=ctx.run_id,
        input_data={"claims": len(brief.claims)},
    ) as span:
        report = verify_brief(brief, state)
        span.update(output={"ok": report.ok, "coverage": report.coverage})

    # Stream tick-by-tick so the verification screen animates green (or red).
    for check in report.claim_checks:
        await ctx.emit(
            EventKind.VERIFY_CLAIM,
            f"{check.ticker} · {check.metric}",
            check.model_dump(),
        )
    for quote in report.quote_checks:
        await ctx.emit(
            EventKind.VERIFY_CLAIM,
            f"{quote.ticker} · quoted headline",
            {**quote.model_dump(), "metric": "quoted_headline", "unit": "text"},
        )

    await ctx.emit(
        EventKind.VERIFY_COMPLETED,
        (
            f"All {report.matched} claims verified"
            if report.ok
            else f"{report.mismatched + report.unverifiable} claim(s) failed verification"
        ),
        report.model_dump(),
    )

    update: dict[str, Any] = {"verification": report.model_dump()}
    if report.ok:
        return update

    failures = report.failures
    update["errors"] = [
        RunError(stage="verify", message=detail, severity="error") for detail in failures[:10]
    ]
    if state.get("regenerations", 0) >= ctx.settings.max_regenerations:
        update["status"] = RunStatus.HUMAN_REVIEW
    return update


def route_after_verify(state: RunState) -> Literal["writer", "gate"]:
    """Exactly one regeneration on mismatch, then straight to the human gate.

    The cap lives in state (`max_regenerations`, default 1 per spec) so a
    mismatch can never ping-pong between writer and verifier.
    """
    report = state.get("verification") or {}
    if report.get("ok"):
        return "gate"
    if int(state.get("regenerations", 0)) < int(state.get("max_regenerations", 1)):
        return "writer"
    return "gate"