rune/tests/test_source_ledger.py

"""Tests: Source Ledger deterministic renderer (#282 / PROV-1).

Invariants:
- READ impossible without a successful machine receipt
- BLOCKED/FAILED appear but back zero accepted claims
- deterministic receipt_id (same inputs → same ID)
- runner/model from persisted fields only; UNKNOWN if absent
- requested_url + resolved_url both retained
- legacy nesting labeled; explicit IDs used when present
- render is a pure function (same input → same output)
"""

from __future__ import annotations

import importlib.util
import json
import sys

import pytest
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
spec = importlib.util.spec_from_file_location("source_ledger", ROOT / "scripts" / "source_ledger.py")
sl = importlib.util.module_from_spec(spec)
sys.modules["source_ledger"] = sl
spec.loader.exec_module(sl)


def _artifact(**overrides):
    a = {
        "discovery": {
            "queries": [{"id": "Q1", "query": "q", "receipt": "ok"}],
            "sources_discovered": [
                {"id": "O1", "url": "https://example.org/read-me", "class": "official docs"},
                {"id": "O2", "url": "https://blocked.example/x", "class": "forum", "note": "BLOCKED: bot challenge"},
                {"id": "O3", "url": "https://failed.example/y", "class": "blog"},
            ],
            "selected_for_reading": ["O1"],
        },
        "reading": {
            "sources_read": [{"id": "O1", "url": "https://example.org/read-me", "claims": [{"id": "E1", "claim": "x"}]}]
        },
        "tool_receipts": [
            {"kind": "SOURCE_READ", "tool": "webfetch", "url": "https://example.org/read-me", "ts": 100},
            {"kind": "TOOL_FAILED", "tool": "webfetch", "url": "https://failed.example/y", "ts": 101},
        ],
        "contradictory_evidence": [
            {"id": "C1", "topic": "t", "resolution": "r", "contradicts_source_ids": ["O1"]},
        ],
    }
    a.update(overrides)
    return a


def test_read_requires_machine_receipt():
    """THE safety invariant: READ only with a SOURCE_READ receipt."""
    records = sl.build_records(_artifact())
    by_id = {r["source_id"]: r for r in records}
    assert by_id["O1"]["status"] == "READ"
    assert by_id["O1"]["receipt_id"].startswith("R-")
    assert by_id["O1"]["supports_evidence"] is True


def test_artifact_claimed_read_without_receipt_downgrades():
    """A source claiming read with no matching machine receipt → FAILED,
    never silently READ."""
    a = _artifact(tool_receipts=[])  # no receipts at all
    # Artifact-asserts-read lives on the discovery source entry (real shape)
    a["discovery"]["sources_discovered"][0]["read_receipt"] = True
    records = sl.build_records(a)
    by_id = {r["source_id"]: r for r in records}
    assert by_id["O1"]["status"] == "FAILED"
    assert by_id["O1"]["supports_evidence"] is False


def test_blocked_and_failed_back_zero_claims():
    records = sl.build_records(_artifact())
    by_id = {r["source_id"]: r for r in records}
    assert by_id["O2"]["status"] == "NOT_SELECTED"
    assert by_id["O3"]["status"] == "FAILED"
    for r in records:
        if r["status"] != "READ":
            assert r["claim_ids"] == []
            assert r["supports_evidence"] is False


def test_receipt_id_deterministic():
    r1 = sl._receipt_id("OUTWARD", "webfetch", "https://x.example/", 100, 0)
    r2 = sl._receipt_id("OUTWARD", "webfetch", "https://x.example/", 100, 0)
    r3 = sl._receipt_id("OUTWARD", "webfetch", "https://x.example/", 100, 1)
    assert r1 == r2 and r1 != r3


def test_requested_and_resolved_urls_retained():
    a = _artifact()
    a["discovery"]["sources_discovered"][0]["resolved_url"] = "https://resolved.example/read-me"
    records = sl.build_records(a)
    by_id = {r["source_id"]: r for r in records}
    assert by_id["O1"]["requested_url"] == "https://example.org/read-me"
    assert by_id["O1"]["resolved_url"] == "https://resolved.example/read-me"
    assert by_id["O1"]["domain"] == "resolved.example"


def test_claim_and_contradiction_ids():
    records = sl.build_records(_artifact())
    by_id = {r["source_id"]: r for r in records}
    assert by_id["O1"]["claim_ids"] == ["E1"]
    assert by_id["O1"]["contradiction_ids"] == ["C1"]


def test_legacy_nesting_flagged_in_render():
    md = sl.render_markdown(
        sl.build_records(_artifact()),
        _artifact(),
        {"runner_name": "r", "runner_hostname": "h"},
        "test-model",
        "abc123",
        legacy_claims=True,
    )
    assert "legacy artifact" in md or "pre-#282" in md


def test_render_pure_function():
    a = _artifact()
    recs = sl.build_records(a)
    md1 = sl.render_markdown(
        recs, a, {"runner_name": "rune-desktop", "runner_hostname": "nick-desktop"}, "glm-5.3-flash", "hash1"
    )
    md2 = sl.render_markdown(
        recs, a, {"runner_name": "rune-desktop", "runner_hostname": "nick-desktop"}, "glm-5.3-flash", "hash1"
    )
    assert md1 == md2


def test_unknown_metadata_not_reconstructed():
    md = sl.render_markdown(sl.build_records(_artifact()), _artifact(), None, "UNKNOWN", "hash1")
    assert "UNKNOWN" in md
    # runner None → UNKNOWN, never guessed
    assert "rune-desktop" not in md


def test_empty_artifact_raises_visibly():
    try:
        sl.build_records({})
        raise AssertionError("should have raised")
    except ValueError:
        pass


def test_full_render_smoke():
    md = sl.render_markdown(
        sl.build_records(_artifact()),
        _artifact(),
        {"runner_name": "rune-desktop", "runner_hostname": "nick-desktop"},
        "glm-5.3-flash",
        "abc123",
    )
    assert "OUTWARD — ACCEPTED" in md
    assert "SOURCE LEDGER" in md
    assert "READ" in md and "NOT_SELECTED" in md and "FAILED" in md
    assert "backs ZERO accepted claims" in md


# ── Idempotent publication (#282 constraint) ────────────────────────────────


def _publish_mod():
    spec = importlib.util.spec_from_file_location(
        "publish_source_ledger", ROOT / "scripts" / "publish_source_ledger.py"
    )
    m = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(m)
    return m


def test_idempotent_publish_updates_not_duplicates(tmp_path, monkeypatch):
    """Same artifact hash → second publish UPDATES the existing comment."""
    plm = _publish_mod()
    a = _artifact()
    (tmp_path / "outward.json").write_text(json.dumps(a))
    calls = {"posted": [], "patched": []}

    class FakeResp:
        def __init__(self, payload, status=200):
            self._p, self.status = payload, status

        def read(self):
            return json.dumps(self._p).encode()

    def fake_api(method, path, payload=None):
        if path.endswith("/comments?limit=50"):
            return 200, [{"id": 1, "body": "<!-- source-ledger hash=abc123 -->\nold body"}]
        if method == "PATCH" and "/comments/" in path:
            calls["patched"].append((path, payload["body"][:40]))
            return 200, {"id": 1}
        if method == "POST" and path.endswith("/comments"):
            calls["posted"].append(payload["body"][:40])
            return 201, {"id": 99}
        return 200, {}

    monkeypatch.setattr(plm, "_api", fake_api)

    r1 = plm.publish(999, str(tmp_path), "abc123")
    assert r1["action"] == "updated" and r1["comment_id"] == 1
    assert not calls["posted"], "must UPDATE, not create a duplicate"
    r2 = plm.publish(999, str(tmp_path), "abc123")
    assert r2["action"] == "updated"
    assert len(calls["posted"]) == 0


def test_new_hash_creates_new_comment(tmp_path, monkeypatch):
    plm = _publish_mod()
    (tmp_path / "outward.json").write_text(json.dumps(_artifact()))
    posted = []

    def fake_api(method, path, payload=None):
        if path.endswith("/comments?limit=50"):
            return 200, []
        if method == "POST" and path.endswith("/comments"):
            posted.append(payload["body"][:50])
            return 201, {"id": 7}
        return 200, {}

    import unittest.mock as mock

    monkeypatch.setattr(plm, "_api", fake_api)
    r = plm.publish(999, str(tmp_path), "newhash")
    assert r["action"] == "created" and len(posted) == 1


def test_publish_failure_raises_visibly(tmp_path, monkeypatch):
    """Provenance publication failure must be VISIBLE, not swallowed."""
    plm = _publish_mod()
    (tmp_path / "outward.json").write_text(json.dumps(_artifact()))

    def fake_api(method, path, payload=None):
        if path.endswith("/comments?limit=50"):
            return 500, "server error"
        return 200, {}

    import unittest.mock as mock

    monkeypatch.setattr(plm, "_api", fake_api)
    with pytest.raises(RuntimeError, match="cannot list comments"):
        plm.publish(999, str(tmp_path), "hash1")