atlas-iac/testing/tests/test_hermes_hux_research_notebook.py

"""HUX-08 research notebooks: state machine, references and concurrency.

Security obligations exercised: every id added to a notebook resolves under
the caller's subtree, so a second subject cannot attach or read another
tenant's records (SO-33); PATCH requires If-Match and a stale revision is a
409 (SO-44); status only moves open -> answered | abandoned; every served
record satisfies citation.schema.json#/$defs/notebook.
"""

from __future__ import annotations

import json
import sys
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, 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"])


@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) == []


@pytest.fixture
def evidence(router):
    """One source, passage and citation owned by the default subject."""
    _, src, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "https://example.com/a", "title": "A"})
    _, psg, _ = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "A says B."})
    _, cit, _ = call(router, "POST", "/hux/v1/messages/msg-9/citations", {"claim": "B", "passage_ids": [psg["id"]], "support": "supports"})
    return src, psg, cit


def notebook(router, **extra):
    status, record, headers = call(router, "POST", "/hux/v1/notebooks", {"conversation_id": "conv_0001abcd", "question": "Which supplier?", **extra})
    assert status == 201, record
    assert headers["ETag"] == "1"
    return record


def test_create_and_get_are_contract_valid(router):
    record = notebook(router, assumptions=["Budget is fixed"], unresolved_questions=["Installation included?"])
    assert valid(record) and record["status"] == "open" and record["revision"] == 1
    assert record["assumptions"] == ["Budget is fixed"] and record["notes"] == []
    status, fetched, headers = call(router, "GET", f"/hux/v1/notebooks/{record['id']}")
    assert status == 200 and fetched == record and headers["ETag"] == "1"
    status, replay, headers = call(router, "POST", "/hux/v1/notebooks", {"conversation_id": "conv_0001abcd", "question": "again"}, {"Idempotency-Key": "nb-key-000001"})
    status, same, headers = call(router, "POST", "/hux/v1/notebooks", {"conversation_id": "conv_0001abcd", "question": "again"}, {"Idempotency-Key": "nb-key-000001"})
    assert (status, same["id"], headers["HUX-Replayed"]) == (200, replay["id"], "true")


@pytest.mark.parametrize("bad", [
    {"conversation_id": "nope"}, {"conversation_id": None}, {"question": ""}, {"question": "q" * 1001},
    {"assumptions": "one"}, {"assumptions": [""]}, {"unresolved_questions": [1]}, {"assumptions": ["a"] * 65},
])
def test_create_rejects_bad_bodies(router, bad):
    status, error, _ = call(router, "POST", "/hux/v1/notebooks", {"conversation_id": "conv_0001abcd", "question": "q", **bad})
    assert status == 400 and valid(error), bad
    assert call(router, "POST", "/hux/v1/notebooks", raw=b"[]")[0] == 400


def test_patch_adds_references_notes_and_lists(router, evidence):
    src, psg, cit = evidence
    record = notebook(router)
    path = f"/hux/v1/notebooks/{record['id']}"
    patch = {
        "add_source_ids": [src["id"], src["id"]], "add_passage_ids": [psg["id"]], "add_citation_ids": [cit["id"]],
        "add_notes": [{"text": "Lead times exclude installation.", "source_id": src["id"]}, {"text": "Plain note"}],
        "assumptions": ["Budget cap 12k"], "unresolved_questions": [],
    }
    status, updated, headers = call(router, "PATCH", path, patch, {"If-Match": "1"})
    assert status == 200 and valid(updated) and headers["ETag"] == "2"
    assert updated["source_ids"] == [src["id"]] and updated["passage_ids"] == [psg["id"]] and updated["citation_ids"] == [cit["id"]]
    assert [n["text"] for n in updated["notes"]] == ["Lead times exclude installation.", "Plain note"]
    assert updated["notes"][0]["source_id"] == src["id"] and "source_id" not in updated["notes"][1]
    assert updated["assumptions"] == ["Budget cap 12k"] and updated["unresolved_questions"] == []
    status, again, _ = call(router, "PATCH", path, {"add_source_ids": [src["id"]]}, {"If-Match": "2"})
    assert again["source_ids"] == [src["id"]] and again["revision"] == 3
    status, error, _ = call(router, "PATCH", path, {"add_notes": [{"text": "x", "source_id": "src_missing00001"}]}, {"If-Match": "3"})
    assert status == 404
    for bad in ({"add_notes": "note"}, {"add_notes": [{"text": ""}]}, {"add_source_ids": "src_x"}, {"assumptions": [""]}):
        status, error, _ = call(router, "PATCH", path, bad, {"If-Match": "3"})
        assert status == 400 and valid(error), bad
    assert call(router, "PATCH", path, raw=b"[]", headers={"If-Match": "3"})[0] == 400
    status, error, _ = call(router, "PATCH", path, {"add_source_ids": [f"src_{i:012d}" for i in range(501)]}, {"If-Match": "3"})
    assert status == 413


def test_patch_requires_matching_if_match(router):
    record = notebook(router)
    path = f"/hux/v1/notebooks/{record['id']}"
    status, error, _ = call(router, "PATCH", path, {"assumptions": ["x"]})
    assert (status, error["code"]) == (400, "invalid")
    status, error, _ = call(router, "PATCH", path, {"assumptions": ["x"]}, {"If-Match": "5"})
    assert (status, error["code"], error["details"]) == (409, "conflict", ["1"])
    assert call(router, "PATCH", path, {"assumptions": ["x"]}, {"If-Match": "abc"})[0] == 400
    assert call(router, "GET", path)[1]["assumptions"] == []
    rows = audit.recent(store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router")))
    assert [r["outcome"] for r in rows if r["action"] == "research.notebook_patch"] == ["deny", "conflict", "deny"]


@pytest.mark.parametrize(("target", "then", "expected"), [
    ("answered", "abandoned", 409), ("abandoned", "answered", 409), ("answered", "open", 409), ("abandoned", "abandoned", 200),
])
def test_status_moves_only_from_open(router, target, then, expected):
    record = notebook(router)
    path = f"/hux/v1/notebooks/{record['id']}"
    status, moved, _ = call(router, "PATCH", path, {"status": target}, {"If-Match": "1"})
    assert status == 200 and moved["status"] == target and valid(moved)
    status, body, _ = call(router, "PATCH", path, {"status": then}, {"If-Match": "2"})
    assert status == expected
    if expected == 409:
        assert body["code"] == "conflict" and call(router, "GET", path)[1]["status"] == target
    assert call(router, "PATCH", path, {"status": "done"}, {"If-Match": str(3 if expected == 200 else 2)})[0] == 409
    assert call(router, "PATCH", path, {"status": target}, {"If-Match": str(3 if expected == 200 else 2)})[0] == 200


def test_second_subject_cannot_read_patch_or_attach(router, evidence):
    src, psg, cit = evidence
    record = notebook(router)
    path = f"/hux/v1/notebooks/{record['id']}"
    assert call(router, "GET", path, headers=OTHER)[0] == 404
    status, error, _ = call(router, "PATCH", path, {"status": "abandoned"}, {**OTHER, "If-Match": "1"})
    assert (status, error["code"]) == (404, "not_found") and valid(error)
    theirs = call(router, "POST", "/hux/v1/notebooks", {"conversation_id": "conv_0001abcd", "question": "spy"}, OTHER)[1]
    for key, value in (("add_source_ids", src["id"]), ("add_passage_ids", psg["id"]), ("add_citation_ids", cit["id"])):
        status, error, _ = call(router, "PATCH", f"/hux/v1/notebooks/{theirs['id']}", {key: [value]}, {**OTHER, "If-Match": "1"})
        assert status == 404, key
    status, mine, _ = call(router, "GET", path)
    assert mine["status"] == "open" and mine["revision"] == 1
    other_rows = audit.recent(store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_fedcba9876543210", "chat", "router")))
    assert other_rows[-1]["outcome"] == "not_found" and other_rows[-1]["action"] == "research.notebook_patch"


def test_notebook_paths_are_tenant_scoped(router):
    record = notebook(router)
    tenant = store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router"))
    assert (tenant.root / "notebooks" / f"{record['id']}.json").exists()
    assert call(router, "GET", "/hux/v1/notebooks/nb_missing000001")[0] == 404
    assert call(router, "GET", "/hux/v1/notebooks/NB")[0] == 404


def test_f9_notebook_free_text_is_scrubbed(router, evidence):
    """F9 / SO-07: question, assumptions, unresolved questions and notes are secret-scrubbed on create and patch."""
    token = "ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
    record = notebook(router, question=f"q {token}", assumptions=[f"a {token}"], unresolved_questions=[f"u {token}"])
    assert record["question"] == "q [redacted:github_token]" and record["assumptions"] == ["a [redacted:github_token]"]
    assert record["unresolved_questions"] == ["u [redacted:github_token]"]
    status, patched, _ = call(router, "PATCH", f"/hux/v1/notebooks/{record['id']}", {"add_notes": [{"text": f"n {token}"}], "assumptions": [f"b {token}"]}, {"If-Match": "1"})
    assert status == 200 and patched["notes"][0]["text"] == "n [redacted:github_token]" and patched["assumptions"] == ["b [redacted:github_token]"]
    leaked = [path for path in Path(router.data_root).rglob("*.json*") if token in path.read_text()]
    assert leaked == []