vasily
78033bb830
CI / quality (push) Canceled after 0s
Adds claim-bound online evidence records and citation verification scripts, node removal transport/adapter, retrieval benchmark and owner-review tooling, GraphCanvas/SourceDetail knowledge browser UI, and the ingest handoff helper. Bumps the agents/gewu pack to the online-verification branch.
287 lines
12 KiB
Python
287 lines
12 KiB
Python
#!/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()
|