agent/evidence.py

"""Deterministic evidence records, envelopes, and brief verification.

Tool outputs use the versioned ``evidence.v1`` JSON envelope.  Evidence IDs are
content-addressed and stable: the same source/document/chunk/content tuple always
produces the same ID, independently of retrieval order or dictionary ordering.
"""
from __future__ import annotations

import hashlib
import json
import re
import unicodedata
from decimal import Decimal, InvalidOperation
from typing import Any, Iterable, Mapping, Sequence

from agent.schemas import EvidenceRecord, EvidenceRef, EvidenceSource

EVIDENCE_SCHEMA = "evidence.v1"
NO_VERIFIED_SYNTHESIS_MESSAGE = (
    "No AI-generated claim in this run passed deterministic evidence verification, "
    "so AI commentary is withheld. Deterministic figures from the ingested SEC "
    "filings (revenue, EPS, margins) are still available in the Financials view — "
    "regenerate the brief to retry the AI synthesis."
)
ENVELOPE_STATUSES = {"OK", "EMPTY", "ERROR"}
TOOL_EVIDENCE_SOURCES = {
    "get_financial_metrics": {"metrics"},
    "search_filing": {"10-K", "10-Q"},
    "search_transcript": {"transcript"},
    "search_news": {"news"},
    "get_analyst_expectations": {"analyst"},
}


def normalize_content(value: str) -> str:
    """Return the canonical representation used for hashing and storage."""
    value = unicodedata.normalize("NFKC", str(value or ""))
    value = value.replace("\r\n", "\n").replace("\r", "\n")
    return "\n".join(line.rstrip() for line in value.split("\n")).strip()


def content_hash(content: str) -> str:
    return hashlib.sha256(normalize_content(content).encode("utf-8")).hexdigest()


def _canonical_json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def stable_evidence_id(
    source: EvidenceSource,
    document_id: str,
    chunk_id: str | None,
    digest: str,
) -> str:
    identity = {
        "chunk_id": chunk_id or "",
        "content_hash": digest,
        "document_id": document_id,
        "source": source,
    }
    return "ev_" + hashlib.sha256(_canonical_json(identity).encode("utf-8")).hexdigest()[:24]


def make_evidence_record(
    *,
    source: EvidenceSource,
    content: str,
    document_id: str = "",
    chunk_id: str | None = None,
    source_url: str | None = None,
    as_of: str | None = None,
    metadata: Mapping[str, Any] | None = None,
) -> EvidenceRecord:
    """Build a validated, content-addressed evidence record."""
    canonical = normalize_content(content)
    digest = content_hash(canonical)
    document_id = str(document_id or f"document:{digest[:16]}")
    ref = EvidenceRef(
        evidence_id=stable_evidence_id(source, document_id, chunk_id, digest),
        source=source,
        content_hash=digest,
        document_id=document_id,
        chunk_id=str(chunk_id) if chunk_id else None,
        source_url=str(source_url) if source_url else None,
        as_of=str(as_of) if as_of else None,
    )
    return EvidenceRecord(ref=ref, content=canonical, metadata=dict(metadata or {}))


def is_valid_evidence_record(record: EvidenceRecord) -> bool:
    """Validate both the content digest and the content-addressed identifier."""
    digest = content_hash(record.content)
    if digest != record.ref.content_hash:
        return False
    return record.ref.evidence_id == stable_evidence_id(
        record.ref.source,
        record.ref.document_id,
        record.ref.chunk_id,
        digest,
    )


def evidence_envelope(
    *,
    tool: str,
    records: Sequence[EvidenceRecord] | None = None,
    query: Mapping[str, Any] | None = None,
    status: str | None = None,
    message: str = "",
    error_code: str | None = None,
) -> str:
    """Serialize a stable ``evidence.v1`` envelope.

    ``OK`` always contains at least one record, ``EMPTY`` contains none, and
    ``ERROR`` contains a machine-readable error code.  Invalid combinations are
    rejected locally instead of leaking an ambiguous payload to the model.
    """
    items = list(records or [])
    if any(not is_valid_evidence_record(item) for item in items):
        raise ValueError("Evidence envelope contains a record with an invalid hash or ID")
    allowed_sources = TOOL_EVIDENCE_SOURCES.get(str(tool))
    if allowed_sources is not None and any(
        item.ref.source not in allowed_sources for item in items
    ):
        raise ValueError(f"Evidence source is inconsistent with tool {tool!r}")
    resolved = status or ("OK" if items else "EMPTY")
    if resolved not in ENVELOPE_STATUSES:
        raise ValueError(f"Unsupported evidence envelope status: {resolved}")
    if resolved == "OK" and not items:
        raise ValueError("An OK evidence envelope requires at least one record")
    if resolved != "OK" and items:
        raise ValueError(f"A {resolved} evidence envelope cannot contain records")
    payload = {
        "schema": EVIDENCE_SCHEMA,
        "status": resolved,
        "tool": str(tool),
        "query": dict(query or {}),
        "records": [item.model_dump(mode="json") for item in items],
        "message": str(message or ""),
        "error": (
            {"code": str(error_code or "TOOL_ERROR"), "message": str(message or "")}
            if resolved == "ERROR" else None
        ),
    }
    return _canonical_json(payload)


def parse_evidence_envelope(payload: Any) -> dict[str, Any] | None:
    """Parse an envelope from a string, dict, or LangChain message-like object."""
    if hasattr(payload, "content"):
        payload = payload.content
    if isinstance(payload, str):
        try:
            payload = json.loads(payload)
        except (TypeError, json.JSONDecodeError):
            return None
    if not isinstance(payload, dict) or payload.get("schema") != EVIDENCE_SCHEMA:
        return None
    if payload.get("status") not in ENVELOPE_STATUSES:
        return None
    if not isinstance(payload.get("records"), list):
        return None
    if payload["status"] == "OK" and not payload["records"]:
        return None
    if payload["status"] != "OK" and payload["records"]:
        return None
    if payload["status"] == "ERROR" and not isinstance(payload.get("error"), dict):
        return None
    return payload


def evidence_records_from(payloads: Any) -> list[EvidenceRecord]:
    """Extract and validate all records from one payload or an iterable of payloads."""
    if payloads is None:
        return []
    if isinstance(payloads, (str, bytes, dict)) or hasattr(payloads, "content"):
        candidates: Iterable[Any] = [payloads]
    else:
        try:
            candidates = iter(payloads)
        except TypeError:
            candidates = [payloads]

    records: list[EvidenceRecord] = []
    seen: set[str] = set()
    for candidate in candidates:
        envelope = parse_evidence_envelope(candidate)
        if not envelope or envelope.get("status") != "OK":
            continue
        allowed_sources = TOOL_EVIDENCE_SOURCES.get(str(envelope.get("tool") or ""))
        for raw in envelope["records"]:
            try:
                record = EvidenceRecord.model_validate(raw)
            except Exception:
                continue
            if (
                is_valid_evidence_record(record)
                and (allowed_sources is None or record.ref.source in allowed_sources)
                and record.ref.evidence_id not in seen
            ):
                records.append(record)
                seen.add(record.ref.evidence_id)
    return records


def is_usable_evidence_payload(payload: Any) -> bool:
    """Return True only for an OK envelope containing a valid evidence record."""
    envelope = parse_evidence_envelope(payload)
    return bool(envelope and envelope.get("status") == "OK" and evidence_records_from(envelope))


def _normalize_for_match(value: str) -> str:
    value = unicodedata.normalize("NFKC", str(value or "")).casefold()
    value = value.translate(str.maketrans({"’": "'", "‘": "'", "“": '"', "”": '"', "–": "-", "—": "-"}))
    return " ".join(value.split())


_MIN_SNIPPET_RUN_WORDS = 6
_SNIPPET_RUN_RATIO = 0.65


def _longest_common_run(a: list[str], b: list[str]) -> int:
    """Length of the longest contiguous word sequence shared by both lists."""
    if not a or not b:
        return 0
    prev = [0] * (len(b) + 1)
    best = 0
    for word_a in a:
        curr = [0] * (len(b) + 1)
        for j, word_b in enumerate(b, start=1):
            if word_a == word_b:
                curr[j] = prev[j - 1] + 1
                if curr[j] > best:
                    best = curr[j]
        prev = curr
    return best


def _snippet_supported(snippet: str, content: str) -> bool:
    """True if the snippet is drawn from the record rather than fabricated.

    A synthesis model routinely trims a verbatim sentence to fit the prompt's
    word-count cap, or drops a connective ("the Company's"), which breaks a
    full substring match even though the quote is genuinely sourced from the
    record. Requiring the longest contiguous word run shared with the record
    to cover most of the snippet still rejects fabricated or unrelated text
    (which shares no long run with the record) while tolerating that trimming.
    """
    norm_snippet = _normalize_for_match(snippet)
    norm_content = _normalize_for_match(content)
    if not norm_snippet:
        return False
    if norm_snippet in norm_content:
        return True
    snippet_words = norm_snippet.split()
    if len(snippet_words) < _MIN_SNIPPET_RUN_WORDS:
        return False
    run = _longest_common_run(snippet_words, norm_content.split())
    return run >= max(_MIN_SNIPPET_RUN_WORDS, round(len(snippet_words) * _SNIPPET_RUN_RATIO))


_NUMBER_RE = re.compile(
    r"(?<![\w])(?P<currency>[$€£])?\s*(?P<number>[+-]?(?:\d{1,3}(?:,\d{3})+|\d+)(?:\.\d+)?)"
    r"\s*(?P<suffix>%|percent|percentage\s+points?|bps?|basis\s+points?|k|m|mm|mn|million|b|bn|billion|t|tn|trillion)?(?![\w])",
    re.IGNORECASE,
)


def _numbers(value: str) -> list[tuple[Decimal, str]]:
    parsed: list[tuple[Decimal, str]] = []
    multipliers = {
        "k": Decimal("1e3"),
        "m": Decimal("1e6"), "mm": Decimal("1e6"), "mn": Decimal("1e6"), "million": Decimal("1e6"),
        "b": Decimal("1e9"), "bn": Decimal("1e9"), "billion": Decimal("1e9"),
        "t": Decimal("1e12"), "tn": Decimal("1e12"), "trillion": Decimal("1e12"),
    }
    for match in _NUMBER_RE.finditer(value or ""):
        try:
            number = Decimal(match.group("number").replace(",", ""))
        except InvalidOperation:
            continue
        suffix = (match.group("suffix") or "").lower()
        if suffix in multipliers:
            parsed.append((number * multipliers[suffix], "currency" if match.group("currency") else "number"))
        elif suffix in {"%", "percent", "percentage point", "percentage points"}:
            parsed.append((number, "percent"))
        elif suffix in {"bp", "bps", "basis point", "basis points"}:
            parsed.append((number / Decimal(100), "percent"))
        else:
            parsed.append((number, "currency" if match.group("currency") else "number"))
    return parsed


def _number_close(left: tuple[Decimal, str], right: tuple[Decimal, str]) -> bool:
    a, a_kind = left
    b, b_kind = right
    if a_kind != b_kind and {a_kind, b_kind} != {"currency", "number"}:
        return False
    tolerance = max(Decimal("0.0001"), abs(a) * Decimal("0.005"))
    return abs(a - b) <= tolerance


def _numbers_supported(claim: str, evidence: str) -> bool:
    claim_numbers = _numbers(claim)
    if not claim_numbers:
        return True
    evidence_numbers = _numbers(evidence)
    return all(any(_number_close(item, candidate) for candidate in evidence_numbers) for item in claim_numbers)


def verify_fact(fact: dict[str, Any], records: Sequence[EvidenceRecord]) -> dict[str, Any]:
    """Verify one fact in place using only deterministic comparisons."""
    raw_ref = fact.get("evidence_ref")
    if isinstance(raw_ref, EvidenceRef):
        supplied = raw_ref
    else:
        try:
            supplied = EvidenceRef.model_validate(raw_ref) if raw_ref else None
        except Exception:
            supplied = None

    def fail(status: str, reason: str) -> dict[str, Any]:
        fact["verification_status"] = status
        fact["verification_reason"] = reason
        fact["reliability"] = "LOW"
        return fact

    if supplied is None:
        return fail("UNVERIFIED", "missing_evidence_ref")

    by_id = {record.ref.evidence_id: record for record in records}
    record = by_id.get(supplied.evidence_id)
    if record is None:
        return fail("FAILED", "evidence_id_not_retrieved")
    actual_digest = content_hash(record.content)
    if actual_digest != record.ref.content_hash:
        return fail("FAILED", "record_content_hash_mismatch")
    if not is_valid_evidence_record(record):
        return fail("FAILED", "record_evidence_id_mismatch")
    if supplied.source != record.ref.source or fact.get("source") != record.ref.source:
        return fail("FAILED", "source_mismatch")
    if supplied.content_hash != record.ref.content_hash:
        return fail("FAILED", "content_hash_mismatch")
    if (
        supplied.document_id != record.ref.document_id
        or supplied.chunk_id != record.ref.chunk_id
        or supplied.source_url != record.ref.source_url
        or supplied.as_of != record.ref.as_of
    ):
        return fail("FAILED", "locator_mismatch")

    snippet = str(fact.get("evidence_snippet") or "").strip()
    if not snippet:
        return fail("FAILED", "missing_evidence_snippet")
    # Metrics records are a structured one-field-per-line table, not prose: a
    # "verbatim quote" spanning two fields (e.g. revenue + margin) can never be
    # a contiguous substring. The per-number check below is the correct
    # integrity guarantee for this source, as it already is for `analyst`.
    if record.ref.source != "metrics" and not _snippet_supported(snippet, record.content):
        return fail("FAILED", "snippet_not_found")
    claim = " ".join(str(fact.get(key) or "") for key in (
        "text", "summary", "rationale", "observation", "reading", "implication",
    ))
    # Only hashed record content may support a numeric claim. Metadata and
    # locators are intentionally excluded because they are not part of the
    # content digest and could otherwise be altered after record creation.
    numeric_context = record.content
    if not _numbers_supported(claim, numeric_context):
        return fail("FAILED", "unsupported_numeric_claim")

    fact["evidence_ref"] = record.ref.model_dump(mode="json")
    fact["verification_status"] = "VERIFIED"
    fact["verification_reason"] = "id_source_hash_snippet_numbers_match"
    return fact


_MARKET_NUMERIC_FIELDS = (
    "consensus_eps_est",
    "consensus_rev_est_bn",
    "revision_30d_pct",
    "d1_price_reaction_pct",
    "d5_price_reaction_pct",
    "since_release_price_reaction_pct",
)


def _market_record_fields(content: str) -> dict[str, str]:
    """Parse the line-oriented analyst tool payload produced by agent.tools."""
    fields: dict[str, str] = {}
    for raw_line in normalize_content(content).splitlines():
        line = raw_line.strip()
        if ":" not in line:
            continue
        key, value = line.split(":", 1)
        key = key.strip()
        if key in {
            *_MARKET_NUMERIC_FIELDS,
            "target_period", "provider_period_codes", "period_aligned",
            "comparison_allowed", "alignment_status", "event_date",
            "event_kind", "event_timing", "event_aligned",
            "event_comparison_allowed", "price_alignment_status",
        }:
            fields[key] = value.split("#", 1)[0].strip()
    return fields


def _optional_float(value: str | None) -> float | None:
    if value is None or value.casefold() in {"", "null", "none", "n/a"}:
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _quarantine_market(market: dict[str, Any], status: str, reason: str) -> str:
    for field in _MARKET_NUMERIC_FIELDS:
        market[field] = None
    market.update({
        "period_aligned": False,
        "comparison_allowed": False,
        "event_aligned": False,
        "event_comparison_allowed": False,
        "alignment_status": "UNVERIFIED_EVIDENCE",
        "price_alignment_status": "UNVERIFIED_EVIDENCE",
        "verification_status": status,
        "verification_reason": reason,
        "rationale": (
            "Market expectations and event reaction are unavailable pending "
            "verified period and event alignment."
        ),
    })
    return status


def _verify_market_expectations(
    market: dict[str, Any], records: Sequence[EvidenceRecord]
) -> str:
    """Replace model-authored market fields with an exact analyst record copy."""
    raw_ref = market.get("evidence_ref")
    try:
        supplied = EvidenceRef.model_validate(raw_ref) if raw_ref else None
    except Exception:
        supplied = None
    if supplied is None:
        return _quarantine_market(market, "UNVERIFIED", "missing_analyst_evidence_ref")

    record = next(
        (item for item in records if item.ref.evidence_id == supplied.evidence_id),
        None,
    )
    if record is None:
        return _quarantine_market(market, "FAILED", "analyst_evidence_id_not_retrieved")
    if record.ref.source != "analyst" or supplied.source != "analyst":
        return _quarantine_market(market, "FAILED", "analyst_source_mismatch")
    if not is_valid_evidence_record(record) or supplied.content_hash != record.ref.content_hash:
        return _quarantine_market(market, "FAILED", "analyst_content_hash_mismatch")
    if (
        supplied.document_id != record.ref.document_id
        or supplied.chunk_id != record.ref.chunk_id
        or supplied.source_url != record.ref.source_url
        or supplied.as_of != record.ref.as_of
    ):
        return _quarantine_market(market, "FAILED", "analyst_locator_mismatch")

    fields = _market_record_fields(record.content)
    period_aligned = fields.get("period_aligned", "false").casefold() == "true"
    comparison_allowed = (
        period_aligned
        and fields.get("comparison_allowed", "false").casefold() == "true"
    )
    event_aligned = fields.get("event_aligned", "false").casefold() == "true"
    event_allowed = (
        event_aligned
        and fields.get("event_comparison_allowed", "false").casefold() == "true"
    )

    market.update({
        "target_period": None if fields.get("target_period") in {None, "null"} else fields.get("target_period"),
        "as_of": record.ref.as_of,
        "period_aligned": period_aligned,
        "comparison_allowed": comparison_allowed,
        "alignment_status": fields.get("alignment_status") or "UNVERIFIED",
        "event_date": None if fields.get("event_date") in {None, "null"} else fields.get("event_date"),
        "event_kind": fields.get("event_kind") or "unknown",
        "event_timing": fields.get("event_timing") or "unknown",
        "event_aligned": event_aligned,
        "event_comparison_allowed": event_allowed,
        "price_alignment_status": fields.get("price_alignment_status") or "UNVERIFIED",
        "evidence_ref": record.ref.model_dump(mode="json"),
        "verification_status": "VERIFIED",
        "verification_reason": "exact_analyst_record_copy",
    })
    for field in _MARKET_NUMERIC_FIELDS[:3]:
        market[field] = _optional_float(fields.get(field)) if comparison_allowed else None
    for field in _MARKET_NUMERIC_FIELDS[3:]:
        market[field] = _optional_float(fields.get(field)) if event_allowed else None

    available = []
    if comparison_allowed:
        available.append(f"consensus aligned to {market.get('target_period') or 'the target period'}")
    if event_allowed:
        available.append("event-aligned price reaction")
    market["rationale"] = (
        "Verified analyst record: " + "; ".join(available) + "."
        if available
        else "Verified analyst record contains no displayable aligned comparison."
    )
    return "VERIFIED"


def _iter_fact_dicts(value: Any) -> Iterable[dict[str, Any]]:
    if isinstance(value, dict):
        if "source" in value and "reliability" in value and (
            "evidence_snippet" in value or "evidence_ref" in value
        ):
            yield value
        for nested in value.values():
            yield from _iter_fact_dicts(nested)
    elif isinstance(value, list):
        for nested in value:
            yield from _iter_fact_dicts(nested)


def verify_brief_evidence(brief: dict[str, Any], payloads: Any) -> dict[str, Any]:
    """Verify every evidence-bearing fact and attach a compact coverage summary."""
    if not isinstance(brief, dict):
        return brief
    records = evidence_records_from(payloads)
    counts = {"VERIFIED": 0, "UNVERIFIED": 0, "FAILED": 0}
    for fact in _iter_fact_dicts(brief):
        verify_fact(fact, records)
        status = fact.get("verification_status", "UNVERIFIED")
        counts[status if status in counts else "UNVERIFIED"] += 1
    market_status = None
    market = brief.get("market_expectations")
    if isinstance(market, dict):
        market_status = _verify_market_expectations(market, records)
        counts[market_status] += 1
    total = sum(counts.values())
    removed = _filter_unverified_claims(brief)
    if market_status in {"UNVERIFIED", "FAILED"}:
        removed += 1
    brief["evidence_coverage"] = {
        "status": "VERIFIED" if total and counts["VERIFIED"] == total else "INCOMPLETE",
        "verified": counts["VERIFIED"],
        "unverified": counts["UNVERIFIED"],
        "failed": counts["FAILED"],
        "total": total,
    }
    brief["verification_report"] = {
        "verified": counts["VERIFIED"],
        "unverified": counts["UNVERIFIED"],
        "failed": counts["FAILED"],
        "removed": removed,
    }
    if removed or not total or counts["VERIFIED"] != total:
        brief["status"] = "PARTIAL"
    else:
        brief["status"] = "COMPLETE"
    return brief


def _is_verified(value: Any) -> bool:
    return isinstance(value, dict) and value.get("verification_status") == "VERIFIED"


def _filter_unverified_claims(brief: dict[str, Any]) -> int:
    """Remove unsupported claims from every user-visible factual section."""
    removed = 0
    verified_count = 0

    for key in (
        "what_changed", "bull_points", "bear_points", "risks_categorized",
        "management_commentary", "guidance_history",
    ):
        values = brief.get(key)
        if not isinstance(values, list):
            continue
        kept = [item for item in values if _is_verified(item)]
        removed += len(values) - len(kept)
        verified_count += len(kept)
        brief[key] = kept

    # These are cross-period conclusions, not properties of the single record
    # cited by each item. Keep the disclosure but suppress model-authored
    # novelty and beat/miss labels until a deterministic comparator supplies them.
    for risk in brief.get("risks_categorized") or []:
        if isinstance(risk, dict):
            risk["is_new_this_filing"] = False
    for guidance in brief.get("guidance_history") or []:
        if isinstance(guidance, dict):
            guidance["actual_result"] = None
            guidance["verdict"] = None

    for key in ("standout_number",):
        value = brief.get(key)
        if isinstance(value, dict):
            if _is_verified(value):
                verified_count += 1
            else:
                brief[key] = None
                removed += 1

    mda = brief.get("mda_summary")
    if isinstance(mda, dict):
        for key in ("drivers", "headwinds"):
            values = mda.get(key)
            if isinstance(values, list):
                kept = [item for item in values if _is_verified(item)]
                removed += len(values) - len(kept)
                verified_count += len(kept)
                mda[key] = kept
        quote = mda.get("key_quote")
        if isinstance(quote, dict):
            if _is_verified(quote):
                verified_count += 1
            else:
                mda["key_quote"] = None
                removed += 1

    tensions = brief.get("analytical_tensions")
    if isinstance(tensions, list):
        kept_tensions = []
        for item in tensions:
            if not isinstance(item, dict):
                continue
            bull = item.get("bullish_evidence")
            bear = item.get("bearish_evidence")
            bull_id = ((bull or {}).get("evidence_ref") or {}).get("evidence_id") if isinstance(bull, dict) else None
            bear_id = ((bear or {}).get("evidence_ref") or {}).get("evidence_id") if isinstance(bear, dict) else None
            if _is_verified(bull) and _is_verified(bear) and bull_id and bear_id and bull_id != bear_id:
                kept_tensions.append(item)
                verified_count += 2
            else:
                removed += 1
        brief["analytical_tensions"] = kept_tensions

    for section in ("earnings_quality_signals", "between_the_lines"):
        values = brief.get(section)
        if not isinstance(values, list):
            continue
        kept = [
            item for item in values
            if isinstance(item, dict) and _is_verified(item.get("evidence"))
        ]
        removed += len(values) - len(kept)
        verified_count += len(kept)
        brief[section] = kept

    # Trend points have no per-row locator in the legacy schema.  Do not show
    # LLM-copied numbers as verified time-series data; Financials reads the DB.
    if brief.get("trends"):
        brief["trends"] = []
        removed += 1

    brief["sentiment"] = None
    if verified_count < 2:
        brief["non_obvious_takeaway"] = ""
    if verified_count == 0:
        brief["what_matters_most"] = NO_VERIFIED_SYNTHESIS_MESSAGE
    return removed