atlas-iac/testing/tests/test_hermes_hux_research_citations.py

"""HUX-08 research: sources, passages and citations.

Security obligations exercised: the service stores but never dereferences a
source ``uri`` and only allowlisted schemes are accepted (SO-19); passage and
citation references resolve under the caller's subtree and foreign ids are
404 (SO-33); hashes and dedupe keys are server-computed (SO-30 by analogy);
every served record satisfies citation.schema.json.
"""

from __future__ import annotations

import hashlib
import json
import sys
import types
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
    sys.path.insert(0, str(FOUNDATION))

from hux import audit, contracts, identity, research, store  # noqa: E402
from hux.server import build_router  # noqa: E402

SCHEMAS = contracts.load_all()
OWNER = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat"}
OTHER = {**OWNER, "X-Hux-Subject": "usr_fedcba9876543210"}
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])


def tenant(router, subject="usr_0123456789abcdef") -> store.TenantStore:
    return store.TenantStore(router.data_root, identity.Identity("slot-3", subject, "chat", "router"))


@pytest.fixture
def router(tmp_path):
    return build_router(tmp_path, {"HUX_FLAGS": ALL_ON})


def call(router, method, path, body=None, headers=None, raw=None):
    payload = raw if raw is not None else (json.dumps(body).encode() if body is not None else b"")
    response = router.dispatch(method, path, {**OWNER, **(headers or {})}, payload)
    return response.status, response.body, response.headers


def valid(record) -> bool:
    return contracts.validate_record(record, SCHEMAS) == []


def source(router, uri="https://Example.com/cabinets/#lead", **extra):
    status, record, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": uri, "title": "Cabinet pricing", "classification": "primary", **extra})
    assert status == 201, record
    return record


def passage(router, source_id, text="Standard base cabinets ship in 3-4 weeks."):
    status, record, _ = call(router, "POST", "/hux/v1/passages", {"source_id": source_id, "text": text})
    assert status == 201, record
    return record


def cite(router, passage_ids, message="msg-44", claim="Base cabinets take three to four weeks.", **extra):
    status, record, _ = call(router, "POST", f"/hux/v1/messages/{message}/citations", {"claim": claim, "passage_ids": passage_ids, "support": "supports", **extra})
    assert status == 201, record
    return record


@pytest.fixture
def events(monkeypatch):
    calls = []
    module = types.ModuleType("hux.events")
    module.emit = lambda store, identity, conversation_id, kind, summary, detail=None, evidence=None, sensitivity="personal", run_id=None, turn=None, correlation_id=None: calls.append((conversation_id, kind, detail, evidence))
    monkeypatch.setitem(sys.modules, "hux.events", module)
    return calls


# --- sources -----------------------------------------------------------------------

def test_source_is_valid_and_deduped_on_normalised_uri(router):
    first = source(router, publisher="Example Co", published_at="2026-01-01T00:00:00Z", content_hash="sha256:" + "a" * 64, conversation_id="conv_0001abcd")
    assert valid(first) and first["provenance"]["actor"] == {"type": "user", "id": "usr_0123456789abcdef"}
    assert first["provenance"]["conversation_id"] == "conv_0001abcd" and first["publisher"] == "Example Co"
    expected = "sha256:" + hashlib.sha256(b"uri\nhttps://example.com/cabinets").hexdigest()
    assert first["dedupe_key"] == expected
    status, again, headers = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "HTTPS://example.com/cabinets", "title": "Different title"})
    assert (status, again["id"], headers["HUX-Replayed"]) == (200, first["id"], "true")
    assert tenant(router).count("sources") == 1
    status, record, _ = call(router, "GET", f"/hux/v1/sources/{first['id']}")
    assert status == 200 and record == first
    assert call(router, "GET", "/hux/v1/sources/src_missing00001")[0] == 404
    assert call(router, "GET", "/hux/v1/sources/..")[0] == 404


def test_source_without_uri_dedupes_on_kind_and_title(router):
    status, memory, _ = call(router, "POST", "/hux/v1/sources", {"kind": "memory", "title": "  Remembered fact "})
    status, again, _ = call(router, "POST", "/hux/v1/sources", {"kind": "memory", "title": "remembered fact"})
    assert status == 200 and again["id"] == memory["id"] and "uri" not in memory and memory["classification"] == "unknown"
    assert research.normalise_uri("artifact://art_0001aaaa@2/") == "artifact://art_0001aaaa@2/"
    assert research.normalise_uri("file:///opt/data/x/") == "file:///opt/data/x"
    assert research.normalise_uri("https://h.example/") == "https://h.example/"


@pytest.mark.parametrize("bad", [
    {"uri": "javascript:alert(1)"}, {"uri": "ftp://x"}, {"uri": "/relative"}, {"kind": "rumour"}, {"kind": None},
    {"classification": "gospel"}, {"title": ""}, {"conversation_id": "x"}, {"content_hash": "md5:abc"}, {"published_at": "yesterday"},
])
def test_source_rejects_bad_bodies(router, bad):
    status, error, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "https://ok.example/", "title": "t", **bad})
    assert status == 400 and valid(error), bad
    assert call(router, "POST", "/hux/v1/sources", raw=b'"str"')[0] == 400


def test_source_idempotency_and_cap(router, monkeypatch):
    status, one, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "https://a.example/1", "title": "t"}, {"Idempotency-Key": "src-key-0001"})
    status, two, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "https://a.example/2", "title": "t"}, {"Idempotency-Key": "src-key-0001"})
    assert status == 200 and two["id"] == one["id"]
    monkeypatch.setattr(research, "MAX_SOURCES", 1)
    status, error, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "https://a.example/3", "title": "t"})
    assert (status, error["code"]) == (413, "too_large")


def test_service_has_no_outbound_client():
    import hux.artifacts
    import hux.research
    for name in ("urllib.request", "http.client", "socket"):
        assert name not in sys.modules or not any(name in getattr(m, "__dict__", {}) for m in (hux.research, hux.artifacts))
    text = (FOUNDATION / "hux" / "research.py").read_text()
    assert "urlopen" not in text and "http.client" not in text


# --- passages ----------------------------------------------------------------------

def test_passage_hash_is_server_computed_and_deduped(router, monkeypatch):
    src = source(router)
    first = passage(router, src["id"])
    assert valid(first) and first["hash"] == "sha256:" + hashlib.sha256(b"Standard base cabinets ship in 3-4 weeks.").hexdigest()
    status, again, headers = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "Standard base cabinets ship in 3-4 weeks.", "hash": "sha256:" + "0" * 64})
    assert (status, again["id"], headers["HUX-Replayed"]) == (200, first["id"], "true") and again["hash"] == first["hash"]
    status, located, _ = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "Other", "locator": {"page": 2, "selector": "#x"}}, {"Idempotency-Key": "psg-key-00001"})
    assert status == 201 and valid(located) and located["locator"] == {"page": 2, "selector": "#x"}
    status, replay, _ = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "Ignored"}, {"Idempotency-Key": "psg-key-00001"})
    assert status == 200 and replay["id"] == located["id"]
    status, error, _ = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "Bad", "locator": {"page": 0}})
    assert status == 400 and error["details"]
    assert call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "x" * 4001})[0] == 400
    assert call(router, "POST", "/hux/v1/passages", {"source_id": "src_missing00001", "text": "x"})[0] == 404
    assert call(router, "POST", "/hux/v1/passages", {"text": "x"})[0] == 404
    assert call(router, "POST", "/hux/v1/passages", raw=b"[]")[0] == 400
    monkeypatch.setattr(research, "MAX_PASSAGES", 2)
    assert call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "third"})[0] == 413


def test_passage_cannot_name_another_subjects_source(router):
    src = source(router)
    status, error, _ = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "steal"}, OTHER)
    assert (status, error["code"]) == (404, "not_found")
    assert call(router, "GET", f"/hux/v1/sources/{src['id']}", headers=OTHER)[0] == 404
    rows = audit.recent(tenant(router, "usr_fedcba9876543210"))
    assert [r["outcome"] for r in rows] == ["not_found", "not_found"]


# --- citations ---------------------------------------------------------------------

def test_citation_attach_dedupes_and_lists_with_passages_and_sources(router, events):
    src = source(router)
    one = passage(router, src["id"])
    two = passage(router, src["id"], "Delivery excludes installation.")
    cit = cite(router, [two["id"], one["id"]], conversation_id="conv_0001abcd", note="checked")
    assert valid(cit) and cit["passage_ids"] == [two["id"], one["id"]] and cit["note"] == "checked"
    assert cit["dedupe_key"] == "sha256:" + hashlib.sha256(("\n".join(["citation", "msg-44", "Base cabinets take three to four weeks.", *sorted([one["id"], two["id"]])])).encode()).hexdigest()
    assert events == [("conv_0001abcd", "citation.attached", {"message_id": "msg-44", "citation_id": cit["id"], "support": "supports"}, [{"kind": "passage", "id": two["id"], "hash": two["hash"]}, {"kind": "passage", "id": one["id"], "hash": one["hash"]}])]
    status, again, headers = call(router, "POST", "/hux/v1/messages/msg-44/citations", {"claim": "Base cabinets take three to four weeks.", "passage_ids": [one["id"], two["id"], one["id"]], "support": "contradicts"})
    assert (status, again["id"], headers["HUX-Replayed"]) == (200, cit["id"], "true") and again["support"] == "supports"
    second = cite(router, [one["id"]], claim="Second claim")
    status, body, _ = call(router, "GET", "/hux/v1/messages/msg-44/citations")
    assert status == 200 and [i["citation"]["id"] for i in body["items"]] == [cit["id"], second["id"]]
    strip = body["items"][0]
    assert [p["id"] for p in strip["passages"]] == [two["id"], one["id"]] and strip["sources"] == [src]
    assert all(valid(r) for item in body["items"] for r in (item["citation"], *item["passages"], *item["sources"]))
    assert call(router, "GET", "/hux/v1/messages/msg-none/citations")[1]["items"] == []
    assert len(events) == 1  # the second citation carried no conversation_id, so nothing to emit against


@pytest.mark.parametrize("bad", [
    {"claim": ""}, {"support": "maybe"}, {"passage_ids": "psg_x"}, {"passage_ids": []}, {"note": "n" * 501}, {"conversation_id": "nope"},
])
def test_citation_rejects_bad_bodies(router, bad):
    src = source(router)
    psg = passage(router, src["id"])
    status, error, _ = call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "c", "passage_ids": [psg["id"]], "support": "supports", **bad})
    assert status == 400 and valid(error), bad


def test_citation_reference_rules(router):
    src = source(router)
    psg = passage(router, src["id"])
    assert call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "c", "passage_ids": ["psg_missing00001"]})[0] == 404
    assert call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "c", "passage_ids": [psg["id"]]}, OTHER)[0] == 404
    status, error, _ = call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "c", "passage_ids": [f"psg_{i:012d}" for i in range(33)]})
    assert status == 413
    assert call(router, "POST", f"/hux/v1/messages/{'m' * 121}/citations", {"claim": "c", "passage_ids": [psg["id"]]})[0] == 400
    assert call(router, "POST", "/hux/v1/messages/msg-1/citations", raw=b"null")[0] == 400
    status, cit, _ = call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "c", "passage_ids": [psg["id"]]}, {"Idempotency-Key": "cit-key-00001"})
    status, replay, _ = call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "other", "passage_ids": [psg["id"]]}, {"Idempotency-Key": "cit-key-00001"})
    assert status == 200 and replay["id"] == cit["id"] and replay["support"] == "unverified"
    assert call(router, "GET", "/hux/v1/messages/msg-1/citations", headers=OTHER)[1]["items"] == []


def test_validate_citations_reports_integrity_problems(router, monkeypatch):
    src = source(router)
    psg = passage(router, src["id"])
    good = cite(router, [psg["id"]])
    contradiction = cite(router, [psg["id"]], claim="Never", support="contradicts")
    noted = cite(router, [psg["id"]], claim="Never again", support="contradicts", note="supplier changed policy")
    tenant_store = tenant(router)
    assert research.validate_citations(tenant_store, [good["id"], noted["id"]]) == []
    assert research.validate_citations(tenant_store, [contradiction["id"]]) == [f"{contradiction['id']}: contradicts without a note"]
    tenant_store.delete("sources", src["id"])
    assert research.validate_citations(tenant_store, [good["id"]]) == [f"{good['id']}: passage {psg['id']} names missing source {src['id']}"]
    tenant_store.delete("passages", psg["id"])
    problems = research.validate_citations(tenant_store, [good["id"], "cit_missing00001", "bad id"])
    assert problems == [f"{good['id']}: passage {psg['id']} does not resolve", "cit_missing00001: citation does not resolve", "bad id: citation does not resolve"]
    monkeypatch.setitem(sys.modules, "hux.events", None)
    assert cite(router, [passage(router, source(router, "https://b.example/")["id"])["id"]], claim="silent", conversation_id="conv_0001abcd")["support"] == "supports"


def test_research_modules_stay_under_500_lines():
    assert len((FOUNDATION / "hux" / "research.py").read_text().splitlines()) <= 500


def test_worker_trust_records_a_system_actor(tmp_path):
    router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_WORKER_KEY": "wk"})
    headers = {"X-Hux-Trust": "worker", "X-Hux-Surface": "worker", "X-Hux-Relay-Key": "wk"}
    status, record, _ = call(router, "POST", "/hux/v1/sources", {"kind": "tool_output", "title": "grep"}, headers)
    assert status == 201 and record["provenance"] == {"surface": "worker", "actor": {"type": "system", "id": "worker"}, "recorded_at": record["retrieved_at"]}