rune/scripts/source_ledger.py

#!/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)