#!/usr/bin/env python3 """Verify answer citations against independently verified accepted-node records. Optional quote repair copies text from the owning claim's evidence. It never changes answer prose, identities or locators and does not assess semantic support. The original model output must remain separate from repaired artifacts. """ from __future__ import annotations import argparse import copy import hashlib import json from pathlib import Path from typing import Any def verify( answers: dict[str, Any], records: dict[str, Any], *, repair_quotes: bool = False, governing_rules: dict[str, str] | None = None, governing_rule_answer_map: dict[str, list[str]] | None = None, ) -> tuple[dict[str, Any], dict[str, Any]]: validated = copy.deepcopy(answers) repairs: list[dict[str, str]] = [] for answer in validated.get("answers", []): if not isinstance(answer, dict) or answer.get("citations") or answer.get("answer_basis"): continue rule_ids = (governing_rule_answer_map or {}).get(answer.get("id", ""), []) if not rule_ids: continue rules = governing_rules or {} if any(rule_id not in rules for rule_id in rule_ids): raise ValueError("Frozen answer-rule map references an unknown governing rule") answer["answer_basis"] = { "kind": "governing_rule", "rules": [{"rule_id": rule_id, "quote": rules[rule_id]} for rule_id in rule_ids], } repairs.append( { "answer_id": answer.get("id", ""), "node_id": "", "code": "ANSWER_BASIS_ATTACHED_FROM_FROZEN_RUBRIC", } ) shape_issues = _validate_answer_shape(validated, governing_rules or {}) nodes = {node["node_id"]: node for node in records["node_audit"]} issues: list[dict[str, str]] = list(shape_issues) checked = 0 answer_rows = validated.get("answers", []) for answer in answer_rows if isinstance(answer_rows, list) else []: if not isinstance(answer, dict) or not isinstance(answer.get("citations"), list): continue for citation in answer["citations"]: if not isinstance(citation, dict): continue checked += 1 identity = {"answer_id": answer["id"], "node_id": citation.get("node_id", "")} node = nodes.get(citation.get("node_id")) claim = ( next( ( item for item in node["claims"] if item["claim_id"] == citation.get("claim_id") ), None, ) if node else None ) expected = None locator = citation.get("locator", {}) for evidence in claim["evidence"] if claim else (): parent = evidence["locator"] same_binding = all( citation.get(key) == evidence[key] for key in ("source_revision_id", "extraction_id") ) and all( locator.get(key) == parent.get(key) for key in ("kind", "block_id", "page", "time_start_ms", "time_end_ms") ) start, end = locator.get("start"), locator.get("end") if ( same_binding and type(start) is int and type(end) is int and parent["start"] <= start < end <= parent["end"] ): expected = evidence["quote"][start - parent["start"] : end - parent["start"]] break if expected is None: issues.append({**identity, "code": "UNOWNED_EVIDENCE_SPAN"}) elif citation.get("quote") != expected: if repair_quotes: citation["quote"] = expected repairs.append({**identity, "code": "QUOTE_RESTORED_FROM_ACCEPTED_EVIDENCE"}) else: issues.append({**identity, "code": "QUOTE_SPAN_MISMATCH"}) report = { "status": "invalid" if issues else "repaired" if repairs else "valid", "checked": checked, "issues": issues, "repairs": repairs, "semantic_support": "not assessed by this binding verifier", } return validated, report def _validate_answer_shape( answers: dict[str, Any], governing_rules: dict[str, str] ) -> list[dict[str, str]]: """Reject malformed or uncited substantive answers before binding checks.""" issues: list[dict[str, str]] = [] rows = answers.get("answers") if not isinstance(rows, list) or not rows: return [{"answer_id": "", "node_id": "", "code": "ANSWERS_REQUIRED"}] seen_ids: set[str] = set() for index, answer in enumerate(rows): if not isinstance(answer, dict): issues.append({"answer_id": str(index), "node_id": "", "code": "ANSWER_INVALID"}) continue answer_id = answer.get("id") identity = answer_id if isinstance(answer_id, str) else str(index) if not isinstance(answer_id, str) or not answer_id.strip() or answer_id in seen_ids: issues.append({"answer_id": identity, "node_id": "", "code": "ANSWER_ID_INVALID"}) else: seen_ids.add(answer_id) if not isinstance(answer.get("answer"), str) or not answer["answer"].strip(): issues.append({"answer_id": identity, "node_id": "", "code": "ANSWER_TEXT_REQUIRED"}) if "evidence_gap" in answer and not isinstance(answer["evidence_gap"], bool): issues.append({"answer_id": identity, "node_id": "", "code": "EVIDENCE_GAP_INVALID"}) citations = answer.get("citations") if not isinstance(citations, list): issues.append({"answer_id": identity, "node_id": "", "code": "CITATIONS_INVALID"}) continue basis = answer.get("answer_basis") if not citations and answer.get("evidence_gap") is not True: if not isinstance(basis, dict) or basis.get("kind") != "governing_rule": issues.append( {"answer_id": identity, "node_id": "", "code": "ANSWER_BASIS_REQUIRED"} ) else: rule_refs = basis.get("rules") if not isinstance(rule_refs, list) or not rule_refs: issues.append( { "answer_id": identity, "node_id": "", "code": "GOVERNING_RULE_REFERENCE_REQUIRED", } ) else: for rule_ref in rule_refs: if not isinstance(rule_ref, dict): issues.append( { "answer_id": identity, "node_id": "", "code": "GOVERNING_RULE_REFERENCE_INVALID", } ) continue rule_id, quote = rule_ref.get("rule_id"), rule_ref.get("quote") if not isinstance(rule_id, str) or rule_id not in governing_rules: issues.append( { "answer_id": identity, "node_id": "", "code": "GOVERNING_RULE_UNKNOWN", } ) elif quote != governing_rules[rule_id]: issues.append( { "answer_id": identity, "node_id": "", "code": "GOVERNING_RULE_QUOTE_MISMATCH", } ) elif basis is not None and ( not isinstance(basis, dict) or basis.get("kind") != "accepted_evidence" ): issues.append({"answer_id": identity, "node_id": "", "code": "ANSWER_BASIS_INVALID"}) for citation in citations: if not isinstance(citation, dict): issues.append({"answer_id": identity, "node_id": "", "code": "CITATION_INVALID"}) continue locator = citation.get("locator") if ( not isinstance(citation.get("node_id"), str) or not isinstance(citation.get("claim_id"), str) or not isinstance(citation.get("source_revision_id"), str) or not isinstance(citation.get("extraction_id"), str) or not isinstance(citation.get("quote"), str) or not isinstance(locator, dict) ): issues.append( { "answer_id": identity, "node_id": str(citation.get("node_id", "")), "code": "CITATION_FIELDS_REQUIRED", } ) return issues def load_governing_rules(path: Path, repo_root: Path) -> dict[str, str]: """Load only rules whose exact source file and line match the pinned catalog.""" catalog = json.loads(path.read_text(encoding="utf-8")) if catalog.get("schema_version") != 1 or not isinstance(catalog.get("rules"), list): raise ValueError("Governing-rule catalog has an unsupported shape") result: dict[str, str] = {} for rule in catalog["rules"]: rule_id, source_path, line, quote = ( rule.get("rule_id"), rule.get("source_path"), rule.get("line"), rule.get("quote"), ) if ( not isinstance(rule_id, str) or not isinstance(source_path, str) or type(line) is not int or line < 1 or not isinstance(quote, str) ): raise ValueError("Governing-rule catalog entry is malformed") source = (repo_root / source_path).resolve() if not source.is_relative_to(repo_root.resolve()): raise ValueError("Governing-rule source escapes the repository") if hashlib.sha256(source.read_bytes()).hexdigest() != rule.get("source_sha256"): raise ValueError(f"Governing-rule source digest changed: {rule_id}") lines = source.read_text(encoding="utf-8").splitlines() if line > len(lines) or quote not in lines[line - 1]: raise ValueError(f"Governing-rule source text changed: {rule_id}") if rule_id in result: raise ValueError(f"Duplicate governing-rule ID: {rule_id}") result[rule_id] = quote return result def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--answers", type=Path, required=True) parser.add_argument("--records", type=Path, required=True) parser.add_argument( "--governing-rules", type=Path, default=Path(__file__).with_name("governing_answer_rules.json"), ) parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parent.parent) parser.add_argument("--repair-quotes", action="store_true") parser.add_argument("--output", type=Path) parser.add_argument( "--governing-rule-answer-map", type=Path, help="Frozen question ID to governing-rule IDs for a separate verified artifact", ) args = parser.parse_args() if args.repair_quotes and args.output is None: parser.error("--repair-quotes requires a separate --output artifact") validated, report = verify( json.loads(args.answers.read_text()), json.loads(args.records.read_text()), repair_quotes=args.repair_quotes, governing_rules=load_governing_rules(args.governing_rules, args.repo_root), governing_rule_answer_map=( json.loads(args.governing_rule_answer_map.read_text(encoding="utf-8")) if args.governing_rule_answer_map else None ), ) print(json.dumps(report, indent=2)) if report["issues"]: raise SystemExit(1) if args.output is not None: with args.output.open("x", encoding="utf-8") as output: output.write(json.dumps(validated, ensure_ascii=False, indent=2) + "\n") if __name__ == "__main__": main()