atlas-iac/testing/tests/test_hermes_hux_memory_retrieval.py

"""HUX-02 retrieval hook: what the agent may read back from the memory ledger.

Security obligations exercised: SO-22 and SO-23 (tombstones and memory_disabled
consulted before the index; only active, retrievable entries are served; no_store,
forgotten, rejected and expired never surface), SO-27 (lazy expiry at read).
"""

from __future__ import annotations

import json
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

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 contracts, events, identity, memory, privacy, store  # noqa: E402
from hux.server import build_router  # noqa: E402

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


def ident() -> identity.Identity:
    return identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router")


def router_for(tmp_path):
    return build_router(tmp_path, {"HUX_FLAGS": ALL_ON})


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


def tenant(tmp_path) -> store.TenantStore:
    return store.TenantStore(tmp_path, ident())


def remember(router, content, conversation_id=CONV, **fields):
    body = {"kind": "fact", "content": content, "reason": "user said so", "proposed_by": "user", "conversation_id": conversation_id, **fields}
    status, record = call(router, "POST", "/hux/v1/memory", HEADERS, body)
    assert status in (201, 202), record
    return record


def ids(rows):
    return [row["id"] for row in rows]


def test_do_not_remember_never_persists_nor_retrieves(tmp_path):
    router = router_for(tmp_path)
    kept = remember(router, "Likes espresso in the morning.")
    declined = remember(router, "Do not remember: I take espresso with sugar.", approval_mode="no_store")
    s = tenant(tmp_path)
    assert declined["status"] == "no_store" and declined["id"] in memory.tombstoned(s)
    assert "sugar" not in json.dumps(list(s.scan(memory.FAMILY))) and "sugar" not in json.dumps(s.read(memory.FAMILY, memory.LEDGER))
    assert ids(memory.retrieve(s, ["espresso"])) == [kept["id"]]
    assert ids(memory.retrieve(s, ["sugar"])) == []
    for record in memory.retrieve(s, []):
        assert contracts.validate_record(record, SCHEMAS) == []


def test_tombstone_wins_even_when_the_document_says_active(tmp_path):
    router = router_for(tmp_path)
    kept = remember(router, "Team standup is at nine.")
    s = tenant(tmp_path)
    s.append(memory.FAMILY, memory.TOMBSTONES, {"memory_id": kept["id"], "at": "2026-08-23T00:00:00Z", "reason": "test", "purged": False})
    assert memory.retrieve(s, ["standup"]) == []
    assert call(router, "GET", "/hux/v1/memory/export")[1]["items"] == []


def test_only_active_and_retrievable_entries_surface(tmp_path):
    router = router_for(tmp_path)
    active = remember(router, "alpha fact")
    proposed = call(router, "POST", "/hux/v1/memory", HEADERS, {"kind": "fact", "content": "beta fact", "reason": "r"})[1]
    rejected = call(router, "POST", "/hux/v1/memory", HEADERS, {"kind": "fact", "content": "gamma fact", "reason": "r"})[1]
    call(router, "POST", f"/hux/v1/memory/{rejected['id']}/reject")
    forgotten = remember(router, "delta fact")
    call(router, "POST", f"/hux/v1/memory/{forgotten['id']}/forget")
    hidden = remember(router, "epsilon fact")
    call(router, "POST", f"/hux/v1/memory/{hidden['id']}/remove_retrieval")
    s = tenant(tmp_path)
    assert ids(memory.retrieve(s, ["fact"])) == [active["id"]]
    assert ids(memory.retrieve(s, [])) == [active["id"]]
    assert proposed["status"] == "proposed"


def test_expired_entries_are_never_served_as_active(tmp_path):
    router = router_for(tmp_path)
    fresh = remember(router, "still valid", ttl={"policy": "expires_at", "expires_at": "2999-01-01T00:00:00Z"})
    stale = remember(router, "already gone", ttl={"policy": "expires_at", "expires_at": "2020-01-01T00:00:00Z"})
    forever = remember(router, "kept forever", ttl={"policy": "never"})
    s = tenant(tmp_path)
    assert ids(memory.retrieve(s, [])) == [forever["id"], fresh["id"]] or set(ids(memory.retrieve(s, []))) == {forever["id"], fresh["id"]}
    expired = s.get(memory.FAMILY, stale["id"])
    assert expired["status"] == "expired" and expired["content"] == "" and expired["audit"][-1]["actor"] == {"type": "system", "id": "retention"}
    assert contracts.validate_record(expired, SCHEMAS) == []
    status, body = call(router, "GET", f"/hux/v1/memory/{stale['id']}")
    assert body["status"] == "expired"
    assert memory.effective_expiry(forever) is None
    decayed = remember(router, "decays", ttl={"policy": "decay", "decay_days": 1})
    far = datetime.now(timezone.utc) + timedelta(days=2)
    assert memory.load(s, decayed["id"], far)["status"] == "expired"
    assert [m["status"] for m in call(router, "GET", "/hux/v1/memory?status=expired")[1]["items"]] == ["expired", "expired"]


def test_scope_and_query_filters(tmp_path):
    router = router_for(tmp_path)
    everywhere = remember(router, "global note about coffee")
    project = remember(router, "project note about coffee", scope={"level": "project", "scope_id": "prj_0001aaaa"})
    other = remember(router, "other project coffee", scope={"level": "project", "scope_id": "prj_0002aaaa"})
    s = tenant(tmp_path)
    assert set(ids(memory.retrieve(s, ["coffee"], {"level": "project", "scope_id": "prj_0001aaaa"}))) == {everywhere["id"], project["id"]}
    assert set(ids(memory.retrieve(s, ["Coffee", ""], None))) == {everywhere["id"], project["id"], other["id"]}
    assert ids(memory.retrieve(s, ["tea"])) == []


def test_blocked_conversations_contribute_and_receive_nothing(tmp_path):
    router = router_for(tmp_path)
    scoped = remember(router, "said in the disabled chat", "conv_0002abcd")
    elsewhere = remember(router, "said elsewhere")
    s = tenant(tmp_path)
    privacy.set_flag(s, "conv_0002abcd", "memory_disabled", True)
    assert ids(memory.retrieve(s, ["said"])) == [elsewhere["id"]]
    assert memory.retrieve(s, ["said"], {"level": "conversation", "scope_id": "conv_0002abcd"}) == []
    assert scoped["status"] == "active"
    later = call(router, "POST", "/hux/v1/memory", HEADERS, {"kind": "fact", "content": "after disable", "reason": "r", "proposed_by": "user", "conversation_id": "conv_0002abcd"})
    assert later[0] == 202 and later[1]["status"] == "no_store" and later[1]["audit"][0]["note"].startswith("memory_disabled")


def test_forgetting_a_conversation_removes_its_memory_from_retrieval(tmp_path):
    router = router_for(tmp_path)
    doomed = remember(router, "doomed detail", "conv_0003abcd")
    pending = call(router, "POST", "/hux/v1/memory", HEADERS, {"kind": "fact", "content": "pending detail", "reason": "r", "conversation_id": "conv_0003abcd"})[1]
    survivor = remember(router, "survivor detail")
    status, body = call(router, "POST", "/hux/v1/conversations/conv_0003abcd/forget")
    assert status == 200 and body["memory_forgotten"] == 2 and body["forgotten"] is True
    s = tenant(tmp_path)
    assert all(r["redaction"]["level"] == "full" for r in s.read(events.FAMILY, "conv_0003abcd"))
    assert ids(memory.retrieve(s, ["detail"])) == [survivor["id"]]
    assert s.get(memory.FAMILY, doomed["id"])["status"] == "forgotten" and s.get(memory.FAMILY, pending["id"])["status"] == "rejected"
    assert {doomed["id"], pending["id"]} <= memory.tombstoned(s)
    assert call(router, "POST", "/hux/v1/memory", HEADERS, {"kind": "fact", "content": "late", "reason": "r", "proposed_by": "user", "conversation_id": "conv_0003abcd"})[1]["status"] == "no_store"