#!/usr/bin/env python3 """source_ledger.py — deterministic Source Ledger renderer (PROV-1, #282). A thin, boring projection of already-persisted state: accepted OUTWARD artifact + SOURCE_READ/TOOL_FAILED receipts + persisted run metadata (runner, model) → pure deterministic render → fixed Markdown template → idempotent Gitea issue comment (caller's responsibility) Hard invariants: - READ is impossible without a successful machine receipt (SOURCE_READ with matching URL). BLOCKED/FAILED sources may appear in the ledger but can never support accepted evidence. - runner/model come from persisted factual fields only; UNKNOWN if absent. Never reconstructed from session memory or timing records. - requested_url and resolved_url both retained. - claim relationships use explicit IDs when present; for pre-#282 artifacts the nesting of claims under sources_read is used and is labeled "derived (legacy nesting)" in the rendered output. - renderer failure raises visibly — provenance publication failing loudly must not silently falsify the ledger. - No trust scores. Factual source classification only. """ from __future__ import annotations import hashlib import json from urllib.parse import urlparse STATUS_READ = "READ" STATUS_BLOCKED = "BLOCKED" STATUS_FAILED = "FAILED" STATUS_NOT_SELECTED = "NOT_SELECTED" def _domain(url: str) -> str: try: return urlparse(url).netloc or "unknown" except (ValueError, AttributeError): # Malformed URL — domain is unknown, the row still renders return "unknown" def _receipt_id(stage: str, tool: str, url: str, ts, seq: int) -> str: """Deterministic receipt identity: hash of stage + tool + canonical URL + timestamp + event sequence (NOT ts+url alone).""" canonical = url.rstrip("/") raw = f"{stage}|{tool}|{canonical}|{ts}|{seq}" return "R-" + hashlib.sha256(raw.encode()).hexdigest()[:12] def _norm(url: str) -> str: return (url or "").rstrip("/").lower() def build_records(outward: dict, stage: str = "OUTWARD") -> list[dict]: """Pure projection: accepted OUTWARD artifact + receipts → ledger records. Raises on structurally invalid input (renderer failure is visible).""" if not outward or not isinstance(outward, dict): raise ValueError("SourceLedger: no accepted OUTWARD artifact") discovery = outward.get("discovery", {}) sources = discovery.get("sources_discovered", []) reading = outward.get("reading", {}).get("sources_read", []) receipts = outward.get("tool_receipts", []) or outward.get("reading", {}).get("tool_receipts", []) # Receipt join keys: successful reads by normalized URL read_urls: dict[str, dict] = {} failed_urls: dict[str, dict] = {} for i, r in enumerate(receipts): url = _norm(r.get("url", "")) if not url: continue rid = _receipt_id(stage, r.get("tool", "?"), r.get("url", ""), r.get("ts", 0), i) if r.get("kind") == "SOURCE_READ": read_urls[url] = {"receipt_id": rid, "tool": r.get("tool", "?"), "ts": r.get("ts")} elif r.get("kind") == "TOOL_FAILED": failed_urls[url] = {"receipt_id": rid, "tool": r.get("tool", "?"), "ts": r.get("ts")} # Claim → source: explicit if artifact carries claim source_ids, else # legacy nesting (claims are children of sources_read entries). claims_by_source: dict[str, list[str]] = {} legacy = False for s in reading: sid = s.get("id") for c in s.get("claims", []): claims_by_source.setdefault(sid, []).append(c["id"]) legacy = True # Contradiction links: explicit contradicts_source_ids if present; # otherwise no deterministic derivation (labeled honestly). contradictions_by_source: dict[str, list[str]] = {} for i, c in enumerate(outward.get("contradictory_evidence", outward.get("contradictions", []))): if isinstance(c, dict): for sid in c.get("contradicts_source_ids", []): contradictions_by_source.setdefault(sid, []).append(c.get("id", f"C{i + 1}")) # plain-string contradictions are not linked (labeled in render) records = [] for s in sources: sid = s.get("id", "?") url = s.get("url", "") nurl = _norm(url) # Status precedence: machine receipt is authoritative. A source the # artifact CLAIMS was read but with no SOURCE_READ receipt cannot # render as READ. if nurl in read_urls: status, receipt = STATUS_READ, read_urls[nurl] elif nurl in failed_urls: status, receipt = STATUS_FAILED, failed_urls[nurl] elif s.get("read_receipt"): # Artifact asserts read but no machine receipt matched — honest # downgrade with provenance note rather than silent READ. status, receipt = STATUS_FAILED, None else: # Never attempted: not selected for reading (distinct from BLOCKED, # which implies an attempted fetch that was refused) status, receipt = STATUS_NOT_SELECTED, None # Upgrade precedence fix is above; this block is intentionally after # status resolution so the downgrade applies regardless of note text. resolved = s.get("resolved_url", url) # explicit field if present records.append( { "receipt_id": (receipt or {}).get("receipt_id", ""), "source_id": sid, "requested_url": url, "resolved_url": resolved, "domain": _domain(resolved), "status": status, "tool": (receipt or {}).get("tool", s.get("tool", "UNKNOWN_NOT_EXPOSED")), "timestamp": (receipt or {}).get("ts", "UNKNOWN_NOT_EXPOSED"), "source_type": s.get("class", s.get("source_class", "UNKNOWN_NOT_EXPOSED")), "claim_ids": claims_by_source.get(sid, []), "contradiction_ids": contradictions_by_source.get(sid, []), "blocked_reason": ( s.get("note", "") if status == STATUS_BLOCKED else ( receipt.get("kind") if status == STATUS_FAILED and receipt else ( "no machine SOURCE_READ receipt matched (legacy artifact)" if status == STATUS_FAILED else "" ) ) ), "supports_evidence": status == STATUS_READ, } ) return records def render_markdown( records: list[dict], outward: dict, runner: dict | None, model: str, artifact_hash: str, legacy_claims: bool = False ) -> str: """Fixed Markdown template. Pure function of records + metadata.""" d = outward.get("discovery", {}) n_read = sum(1 for r in records if r["status"] == STATUS_READ) n_blocked = sum(1 for r in records if r["status"] == STATUS_BLOCKED) n_failed = sum(1 for r in records if r["status"] == STATUS_FAILED) n_claims = sum(len(r["claim_ids"]) for r in records) n_contra = len(outward.get("contradictory_evidence", outward.get("contradictions", []))) runner_name = (runner or {}).get("runner_name", "UNKNOWN") runner_host = (runner or {}).get("runner_hostname", "UNKNOWN") prov_note = "" if legacy_claims: prov_note = ( "\n*Claim↔source links for this legacy artifact are derived from " "claim nesting (pre-#282 format); future artifacts carry explicit " "supports_source_ids/contradicts_source_ids.*" ) lines = [ "## OUTWARD — ACCEPTED", f"- artifact hash: `{artifact_hash}`", f"- runner: {runner_name} ({runner_host})", f"- model: {model}", f"- queries: {len(d.get('queries', []))}", f"- sources discovered: {len(records)}", f"- sources actually read (machine receipt): {n_read}", f"- blocked: {n_blocked} | failed fetch: {n_failed}", f"- evidence claims: {n_claims} | contradictions: {n_contra}", "", "## SOURCE LEDGER", "", "| ID | source/domain | type | status | E-claims | contradictions | caveat |", "|---|---|---|---|---|---|---|", ] for r in records: caveat = r["blocked_reason"] or "" if r["status"] != STATUS_READ: caveat = (caveat + " — backs ZERO accepted claims").strip(" —") link = f"[{r['domain']}]({r['requested_url']})" if r["requested_url"].startswith("http") else r["domain"] lines.append( f"| {r['source_id']} | {link} | {r['source_type']} | {r['status']} " f"| {len(r['claim_ids'])} | {len(r['contradiction_ids'])} | {caveat[:60]} |" ) lines.append(prov_note) return "\n".join(lines)