# Build Evidence Ledger - Object ID: `em:research-note:sha256:f5562c8c9ff362229c692ee21c25fc1b30039f74204017019138921a7be7d444` - Kind: `research-note` - Repository path: [`research/how-we-know/agent-citation-lineage/build_evidence_ledger.py`](https://github.com/yoheinakajima/epistemedia/blob/f92846570180dfa4511263f8ba98ecd18f7772c9/research/how-we-know/agent-citation-lineage/build_evidence_ledger.py) - Content digest: `c8c6bfa98dd1a922f692659e3b168b4c1e0e719315b7b2562e468d7c818b3efc` **Also filed under:** [Research Program](https://epistemedia.org/topics/research-program/) ## Source content """Build and verify the EM-0026 source, span, warrant, and dependence ledger. Network retrieval is deliberately outside deterministic validation. The two capture subcommands turn already-downloaded public artifacts into byte-bound receipts. The default build consumes only committed receipts and raw answer artifacts. """ from __future__ import annotations import argparse import hashlib import json import re import unicodedata from pathlib import Path from typing import Any PACKET = Path(__file__).resolve().parent REPO = PACKET.parents[2] ANSWERS = PACKET / "answers-v2" TRACES = PACKET / "traces-v2" CONFIG = PACKET / "source-normalization-v1.json" READBACKS = PACKET / "source-readbacks-v1.json" SPAN_READBACKS = PACKET / "span-readbacks-v1.json" LEDGER = PACKET / "evidence-ledger-v1.json" def load(path: Path) -> Any: return json.loads(path.read_text()) def digest_bytes(payload: bytes) -> str: return hashlib.sha256(payload).hexdigest() def stable_id(prefix: str, *parts: str) -> str: payload = "\0".join(parts).encode() return f"{prefix}:sha256:{digest_bytes(payload)}" def normalize(value: str) -> str: value = unicodedata.normalize("NFKC", value) value = value.translate( str.maketrans( { "“": '"', "”": '"', "‘": "'", "’": "'", "‐": "-", "‑": "-", "–": "-", "—": "-", "−": "-", "…": "...", "\u00a0": " ", } ) ) return " ".join(value.split()).strip() def quote_body(value: str) -> str: value = normalize(value).strip() while len(value) >= 2 and value[0] == value[-1] == '"': value = value[1:-1].strip() return value def ordered_fragments(quote: str, source: str) -> tuple[bool, list[str]]: fragments = [ quote_body(item).strip(" .;:") for item in re.split(r"(?:\.{3}|\"\s*;\s*\"|\"\s*;|;\s*\")", quote) ] fragments = [item for item in fragments if len(item) >= 12] if len(fragments) < 2: return False, fragments cursor = 0 for fragment in fragments: found = source.find(fragment, cursor) if found < 0: return False, fragments cursor = found + len(fragment) return True, fragments def answers() -> list[tuple[str, dict[str, Any]]]: return [(path.stem, load(path)) for path in sorted(ANSWERS.glob("*.json"))] def capture_readbacks(capture_root: Path, output: Path) -> None: config = load(CONFIG) raw_urls = {source["url"] for _, answer in answers() for source in answer["sources"]} mappings = {item["requested_url"]: item for item in config["url_mappings"]} if raw_urls != set(mappings): missing = sorted(raw_urls - set(mappings)) extra = sorted(set(mappings) - raw_urls) raise SystemExit(f"URL mapping coverage drift; missing={missing}, extra={extra}") records = [] for url in sorted(mappings): mapping = mappings[url] body_path = capture_root / mapping["capture_body"] if not body_path.is_file(): raise SystemExit(f"capture body missing: {body_path}") payload = body_path.read_bytes() status = int(mapping["http_status"]) usable = bool(mapping.get("usable", 200 <= status < 300)) records.append( { "requested_url": url, "resolved_url": url, "redirect_chain": [], "retrieval_status": "retrieved" if usable else "inaccessible", "http_status": status, "media_type": mapping["media_type"], "captured_bytes": len(payload), "captured_sha256": digest_bytes(payload), "edition_id": mapping["edition_id"], } ) document = { "schema": "https://epistemedia.org/research/source-readbacks-v1.json", "task_id": "EM-0026", "captured_at": config["readback_captured_at"], "capture_method": ( "credential-free curl GET with redirects enabled; response bodies " "remained outside the repository" ), "records": records, "limitations": [ "A successful HTTP read proves only that this carrier returned bytes at capture time.", "Wiley and OpenReview returned HTTP 403; PubMed returned HTTP 203 " "with a cookie-interstitial body. Alternate authoritative carriers " "were not substituted for their URL-level status.", "Mendeley landing pages returned HTTP 200, but its credential-free " "file API returned HTTP 401 and no supplementary file bytes were captured.", ], } output.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n") def capture_span_readbacks(capture_root: Path, output: Path) -> None: config = load(CONFIG) editions = {item["edition_id"]: item for item in config["editions"]} url_map = {item["requested_url"]: item for item in config["url_mappings"]} readbacks = {item["requested_url"]: item for item in load(READBACKS)["records"]} source_texts: dict[str, tuple[str, str]] = {} for edition_id, edition in editions.items(): text_path = capture_root / edition["source_text_file"] if not text_path.is_file(): raise SystemExit(f"source text missing: {text_path}") payload = text_path.read_bytes() source_texts[edition_id] = ( normalize(payload.decode(errors="replace")), digest_bytes(payload), ) records = [] for run_id, answer in answers(): for source in answer["sources"]: mapping = url_map[source["url"]] edition_id = mapping["edition_id"] source_text, text_digest = source_texts[edition_id] citation_accessible = readbacks[source["url"]]["retrieval_status"] == "retrieved" for span in source["exact_spans"]: quote = quote_body(span["quote"]) exact = bool(quote) and quote in source_text fragments_match, fragments = ordered_fragments(quote, source_text) if exact: match_status = "exact-normalized-match" elif fragments_match: match_status = "ordered-fragment-match" else: match_status = "unresolved" occurrence_id = f"{run_id}:{source['source_id']}:{span['span_id']}" records.append( { "span_occurrence_id": occurrence_id, "run_id": run_id, "source_id": source["source_id"], "span_id": span["span_id"], "edition_id": edition_id, "locator": span["locator"], "quote_sha256": digest_bytes(span["quote"].encode()), "span_root_id": stable_id( "span", edition_id, normalize(span["locator"]), normalize(span["quote"]), ), "match_status": match_status, "citation_carrier_accessible": citation_accessible, "source_text_sha256": text_digest, "matched_fragments": fragments if fragments_match and not exact else [], } ) document = { "schema": "https://epistemedia.org/research/span-readbacks-v1.json", "task_id": "EM-0026", "captured_at": config["readback_captured_at"], "normalization": ( "Unicode NFKC, typographic punctuation normalization, and whitespace collapse" ), "credit_rule": ( "Only exact-normalized-match and ordered-fragment-match create exact-span " "roots; semantic warrant remains a separate review." ), "records": records, "limitations": [ "Text extraction can reorder tables, formulas, and multi-column PDF content.", "An ordered-fragment match confirms each quoted fragment in sequence, " "not a contiguous sentence.", "A text match does not by itself establish that the source warrants " "an agent proposition.", ], } output.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n") def build_ledger() -> dict[str, Any]: config = load(CONFIG) readback_document = load(READBACKS) span_document = load(SPAN_READBACKS) readbacks = {item["requested_url"]: item for item in readback_document["records"]} span_readbacks = {item["span_occurrence_id"]: item for item in span_document["records"]} editions = {item["edition_id"]: item for item in config["editions"]} works = {item["work_id"]: item for item in config["works"]} group_for_result = { occurrence: group for group in config["claim_groups"] for occurrence in group["result_occurrences"] } corrections_by_occurrence: dict[str, list[str]] = {} for correction in config["corrections"]: for occurrence in correction["raw_occurrences"]: corrections_by_occurrence.setdefault(occurrence, []).append(correction["correction_id"]) report_records = [] citation_records = [] span_records = [] claim_records = [] dependence_edges = list(config["dependence_edges"]) for group in config["claim_groups"]: dependence_edges.extend( [ { "dimension": "data", "from": group["warrant_id"], "to": group["data_root"], "status": "declared-reviewed-lineage", }, { "dimension": "method", "from": group["warrant_id"], "to": group["method_root"], "status": "declared-reviewed-lineage", }, { "dimension": "derivation", "from": group["warrant_id"], "to": group["derivation_root"], "status": "declared-reviewed-lineage", }, ] ) seen_result_occurrences: set[str] = set() for run_id, answer in answers(): answer_path = ANSWERS / f"{run_id}.json" trace = load(TRACES / f"{run_id}.json") report_records.append( { "run_id": run_id, "answer_path": str(answer_path.relative_to(REPO)), "answer_sha256": digest_bytes(answer_path.read_bytes()), "answer_bytes": len(answer_path.read_bytes()), "trace_path": str((TRACES / f"{run_id}.json").relative_to(REPO)), "requested_model_profile": trace["requested_model_profile"], "reported_model_identity": trace["reported_model_identity"], "prompt_sha256": trace["prompt"]["sha256"], "retrieval_infrastructure": trace["tool_configuration"]["reported"], "status": trace["status"], } ) dependence_edges.extend( [ { "dimension": "model_profile", "from": f"report:{run_id}", "to": f"model-profile:{trace['requested_model_profile']}", "status": "requested-not-independent", }, { "dimension": "prompt", "from": f"report:{run_id}", "to": f"prompt:sha256:{trace['prompt']['sha256']}", "status": "shared-exact-bytes", }, { "dimension": "retrieval_infrastructure", "from": f"report:{run_id}", "to": "retrieval:codex-public-web-implementation-unknown", "status": "unknown-or-shared", }, ] ) source_by_id = {item["source_id"]: item for item in answer["sources"]} span_to_source: dict[str, str] = {} for source in answer["sources"]: if source["url"] not in readbacks: raise SystemExit(f"missing readback for {source['url']}") readback = readbacks[source["url"]] edition = editions[readback["edition_id"]] occurrence_id = f"{run_id}:{source['source_id']}" source_span_occurrences = [] for span in source["exact_spans"]: span_occurrence_id = f"{occurrence_id}:{span['span_id']}" if span_occurrence_id not in span_readbacks: raise SystemExit(f"missing span readback for {span_occurrence_id}") review = span_readbacks[span_occurrence_id] span_to_source[span["span_id"]] = source["source_id"] source_span_occurrences.append(span_occurrence_id) span_records.append( { **review, "requested_url": source["url"], "source_work_id": edition["work_id"], "quote": span["quote"], "supports_declared_by_raw_answer": span["supports"], "license_treatment": edition["license_treatment"], } ) dependence_edges.append( { "dimension": "exact_span", "from": f"citation:{occurrence_id}", "to": review["span_root_id"], "status": review["match_status"], } ) unresolved = ( readback["retrieval_status"] != "retrieved" or any( span_readbacks[item]["match_status"] == "unresolved" for item in source_span_occurrences ) or any( correction_id == "correction:mendeley-file-readback" for correction_id in corrections_by_occurrence.get(occurrence_id, []) ) ) citation_records.append( { "citation_occurrence_id": occurrence_id, "run_id": run_id, "raw_source_id": source["source_id"], "raw_title": source["title"], "requested_url": source["url"], "readback": readback, "source_work_id": edition["work_id"], "edition_id": edition["edition_id"], "span_occurrence_ids": source_span_occurrences, "license": edition["license"], "license_treatment": edition["license_treatment"], "correction_ids": corrections_by_occurrence.get(occurrence_id, []), "resolution_status": ( "unresolved" if unresolved else "resolved-and-span-matched" ), } ) dependence_edges.extend( [ { "dimension": "requested_url", "from": f"citation:{occurrence_id}", "to": source["url"], "status": readback["retrieval_status"], }, { "dimension": "source_work", "from": f"citation:{occurrence_id}", "to": edition["work_id"], "status": "normalized", }, { "dimension": "edition", "from": f"citation:{occurrence_id}", "to": edition["edition_id"], "status": "examined", }, ] ) for result in answer["results"]: occurrence_id = f"{run_id}:{result['result_id']}" seen_result_occurrences.add(occurrence_id) group = group_for_result.get(occurrence_id) if group is None: raise SystemExit(f"unreviewed result occurrence: {occurrence_id}") missing_sources = sorted(set(result["source_ids"]) - set(source_by_id)) missing_spans = sorted(set(result["exact_span_ids"]) - set(span_to_source)) if missing_sources or missing_spans: raise SystemExit( f"broken raw result linkage for {occurrence_id}: " f"sources={missing_sources}, spans={missing_spans}" ) linked_citations = [f"{run_id}:{source_id}" for source_id in result["source_ids"]] linked_spans = [ f"{run_id}:{span_to_source[span_id]}:{span_id}" for span_id in result["exact_span_ids"] ] linked_citation_records = [ next( item for item in citation_records if item["citation_occurrence_id"] == citation_id ) for citation_id in linked_citations ] source_unresolved = any( item["resolution_status"] == "unresolved" for item in linked_citation_records ) span_unresolved = any( span_readbacks[span_id]["match_status"] == "unresolved" for span_id in linked_spans ) correction_ids = corrections_by_occurrence.get(occurrence_id, []) semantic_no_credit = any( item["correction_id"] in correction_ids and item["kind"] == "semantic-warrant" for item in config["corrections"] ) no_credit_correction = any( item["correction_id"] in correction_ids and ("no credit" in item["effect"] or "withheld" in item["effect"]) for item in config["corrections"] ) if source_unresolved or span_unresolved: status = "unresolved-no-warrant-credit" elif semantic_no_credit: status = "independent-review-no-credit-insufficient-span-semantics" elif no_credit_correction: status = "qualified-no-credit-for-corrected-component" else: status = "author-candidate-supported-pending-independent-review" claim_records.append( { "claim_occurrence_id": occurrence_id, "warrant_id": group["warrant_id"], "raw_proposition": result["proposition"], "canonical_proposition": group["canonical_proposition"], "source_citation_occurrence_ids": linked_citations, "span_occurrence_ids": linked_spans, "correction_ids": correction_ids, "review_status": status, "warrant_dimensions": config["warrant_dimension_defaults"], } ) for citation_id, citation_record in zip( linked_citations, linked_citation_records, strict=True ): dependence_edges.append( { "dimension": "upstream_citation", "from": f"claim:{occurrence_id}", "to": f"citation:{citation_id}", "status": ( "declared-and-resolved" if citation_record["resolution_status"] == "resolved-and-span-matched" else "declared-but-citation-unresolved" ), } ) if seen_result_occurrences != set(group_for_result): raise SystemExit( "claim-group coverage drift; extra=" f"{sorted(set(group_for_result) - seen_result_occurrences)}" ) match_statuses = {"exact-normalized-match", "ordered-fragment-match"} exact_span_roots = { item["span_root_id"] for item in span_records if item["match_status"] in match_statuses } claims_by_warrant: dict[str, list[dict[str, Any]]] = {} for item in claim_records: claims_by_warrant.setdefault(item["warrant_id"], []).append(item) candidate_warrants = { group["warrant_id"] for group in config["claim_groups"] if group["review_status"].startswith("candidate-supported") and any( span_readbacks[span_id]["match_status"] in match_statuses for claim in claims_by_warrant[group["warrant_id"]] for span_id in claim["span_occurrence_ids"] ) } cited_work_ids = {item["source_work_id"] for item in citation_records} cited_edition_ids = {item["edition_id"] for item in citation_records} inaccessible = [ item for item in citation_records if item["readback"]["retrieval_status"] == "inaccessible" ] unresolved = [item for item in citation_records if item["resolution_status"] == "unresolved"] corrected_claims = { occurrence for correction in config["corrections"] if correction["kind"] in { "scope", "internal-source-conflict", "dependence", "edition", "semantic-warrant", } for occurrence in correction["raw_occurrences"] if occurrence in seen_result_occurrences or occurrence.endswith(":answer") } counts = { "captured_reports": len(report_records), "citation_occurrences": len(citation_records), "cited_urls": len({item["requested_url"] for item in citation_records}), "resolving_url_roots": len( { item["requested_url"] for item in citation_records if item["readback"]["retrieval_status"] == "retrieved" } ), "source_work_roots": len(cited_work_ids), "examined_edition_roots": len(cited_edition_ids), "raw_span_occurrences": len(span_records), "exact_span_roots": len(exact_span_roots), "raw_claim_occurrences": len(claim_records), "candidate_warrant_roots": len(candidate_warrants), "independently_confirmed_warrant_roots": 0, "non_resolving_citations": 0, "inaccessible_citations": len(inaccessible), "invalid_citations": 0, "unsupported_or_force_raised_claims": len(corrected_claims), "unresolved_citations": len(unresolved), } return { "schema": "https://epistemedia.org/research/agent-citation-evidence-ledger-v1.json", "task_id": "EM-0026", "protocol_id": "em-0026-agent-citation-trace-v2", "status": "research-only-author-review-complete-independent-review-required", "evidence_cutoff": config["review_cutoff"], "readback_captured_at": config["readback_captured_at"], "count_grammar": config["count_grammar"], "counts": counts, "reports": report_records, "works": [works[item] for item in sorted(cited_work_ids)], "editions": [editions[item] for item in sorted(cited_edition_ids)], "citations": citation_records, "spans": span_records, "claims": claim_records, "candidate_warrants": [ { **{key: value for key, value in group.items() if key != "result_occurrences"}, "independent_review_status": "pending", } for group in config["claim_groups"] if group["warrant_id"] in candidate_warrants ], "dependence_edges": dependence_edges, "corrections": config["corrections"], "limitations": [ "The eight reports share exact prompt bytes, Codex runtime lineage, and " "unknown or shared retrieval infrastructure; they are observations, not " "independent evidence roots.", "Candidate warrant roots are author-side normalization results and remain " "zero independently confirmed roots until a fresh-clone reviewer repeats " "source and count checks.", "HTTP readback, text matching, and semantic warrant are separate gates.", "No universal claim about all agents, current products, or all web sources " "is authorized.", ], } def verify() -> dict[str, Any]: ledger = load(LEDGER) rebuilt = build_ledger() if ledger != rebuilt: raise SystemExit( "evidence ledger is not the deterministic build of raw captures and reviews" ) if ledger["counts"]["captured_reports"] != 8: raise SystemExit("report count drift") if ledger["counts"]["citation_occurrences"] != 48: raise SystemExit("citation occurrence count drift") if ledger["counts"]["raw_span_occurrences"] != 127: raise SystemExit("span occurrence count drift") if ledger["counts"]["raw_claim_occurrences"] != 52: raise SystemExit("claim occurrence count drift") required_dimensions = { "model_profile", "prompt", "retrieval_infrastructure", "requested_url", "source_work", "edition", "exact_span", "upstream_citation", "data", "method", "derivation", } dimensions = {item["dimension"] for item in ledger["dependence_edges"]} missing = required_dimensions - dimensions if missing: raise SystemExit(f"dependence dimensions missing: {sorted(missing)}") if ledger["counts"]["independently_confirmed_warrant_roots"] != 0: raise SystemExit("author packet cannot self-confirm independent warrant roots") citation_status = { f"citation:{item['citation_occurrence_id']}": item["resolution_status"] for item in ledger["citations"] } false_resolved_edges = [ item for item in ledger["dependence_edges"] if item["dimension"] == "upstream_citation" and item["status"] == "declared-and-resolved" and citation_status.get(item["to"]) != "resolved-and-span-matched" ] if false_resolved_edges: raise SystemExit("resolved claim-to-citation edge targets an unresolved citation") return ledger["counts"] def main() -> None: parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="command", required=True) capture_urls = subparsers.add_parser("capture-readbacks") capture_urls.add_argument("--capture-root", type=Path, required=True) capture_urls.add_argument("--output", type=Path, default=READBACKS) capture_spans = subparsers.add_parser("capture-spans") capture_spans.add_argument("--capture-root", type=Path, required=True) capture_spans.add_argument("--output", type=Path, default=SPAN_READBACKS) build = subparsers.add_parser("build") build.add_argument("--output", type=Path, default=LEDGER) subparsers.add_parser("verify") args = parser.parse_args() if args.command == "capture-readbacks": capture_readbacks(args.capture_root, args.output) elif args.command == "capture-spans": capture_span_readbacks(args.capture_root, args.output) elif args.command == "build": args.output.write_text(json.dumps(build_ledger(), indent=2, sort_keys=True) + "\n") else: print(json.dumps(verify(), sort_keys=True)) if __name__ == "__main__": main()