"""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"(?[$€£])?\s*(?P[+-]?(?:\d{1,3}(?:,\d{3})+|\d+)(?:\.\d+)?)" r"\s*(?P%|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