atlas-iac/testing/tests/test_hermes_hux_suggestions_backend.py

"""Focused HUX-09 server-authoritative suggestion tests."""

from __future__ import annotations

import json
import sys
from datetime import datetime, timedelta, timezone
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 contracts, identity, privacy, store, suggestions
from hux.errors import Invalid
from hux.server import build_router

HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"}
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
START = datetime(2026, 8, 24, 12, 0, tzinfo=timezone.utc)


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


def tenant(router):
    return store.TenantStore(router.data_root, identity.resolve(HEADERS, {"HUX_ROUTER_KEY": "rk"}))


@pytest.fixture
def setup(tmp_path, monkeypatch):
    moment = {"value": START}
    monkeypatch.setattr(suggestions, "clock", lambda: moment["value"])
    monkeypatch.setattr(suggestions, "now_iso", lambda: moment["value"].strftime("%Y-%m-%dT%H:%M:%SZ"))
    router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk"})
    _, project, _ = call(router, "POST", "/hux/v1/projects", {"name": "P"})
    _, conversation, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C", "project_id": project["id"]})
    base = f"/hux/v1/projects/{project['id']}/conversations/{conversation['id']}/suggestions"
    return router, project, conversation, base, moment


def evaluate(setup, context="first_session", key="suggest-key-001", **extra):
    router, _, _, base, _ = setup
    return call(router, "POST", base + "/evaluate", {"context": context, **extra}, {"Idempotency-Key": key})


def test_every_context_returns_a_canonical_server_definition(setup):
    for index, context in enumerate(suggestions.CONTEXTS):
        status, result, headers = evaluate(setup, context, f"context-key-{index:02}")
        assert status == 200 and result["stored"] is True and headers["Cache-Control"] == "no-store"
        suggestion = result["suggestion"]
        assert suggestion["trigger"] == {"surface": "chat", "context": context}
        assert suggestion["suppression"] == {"dismissable": True, "max_shows": 3, "cooldown_seconds": 86400, "never_again_supported": True}
        assert contracts.validate("suggestion.schema.json", suggestion, pointer="/$defs/suggestion") == []
        assert contracts.validate("suggestion.schema.json", result["state"], pointer="/$defs/state") == []


def test_idempotency_cooldown_and_max_show_are_authoritative(setup):
    _, _, _, _, moment = setup
    status, first, _ = evaluate(setup)
    assert status == 200 and first["state"]["shows"] == 1 and first["revision"] == 1
    status, replay, headers = evaluate(setup)
    assert status == 200 and replay == first and headers["HUX-Replayed"] == "true"
    assert evaluate(setup, context="idle")[0] == 409
    status, cooled, _ = evaluate(setup, key="suggest-key-002")
    assert status == 200 and cooled == {"suggestion": None, "reason": "cooldown", "stored": False}
    moment["value"] += timedelta(days=1, seconds=1)
    second = evaluate(setup, key="suggest-key-003")[1]
    assert second["state"]["shows"] == 2 and second["revision"] == 2
    moment["value"] += timedelta(days=1, seconds=1)
    third = evaluate(setup, key="suggest-key-004")[1]
    assert third["state"]["shows"] == 3
    moment["value"] += timedelta(days=1, seconds=1)
    exhausted = evaluate(setup, key="suggest-key-005")[1]
    assert exhausted == {"suggestion": None, "reason": "max_shows", "stored": False}


def test_decisions_require_click_if_match_and_idempotency(setup):
    _, _, _, base, moment = setup
    shown = evaluate(setup)[1]
    suggestion_id = shown["suggestion"]["id"]
    path = base + f"/{suggestion_id}/decisions"
    headers = {"If-Match": "1", "Idempotency-Key": "decision-key-001"}
    status, result, response_headers = call(setup[0], "POST", path, {"decision": "dismissed", "clicked": True}, headers)
    assert status == 200 and result["decision"] == "dismissed" and result["revision"] == 2
    assert response_headers["ETag"] == "2" and "dismissed_at" in result["state"]
    status, replay, replay_headers = call(setup[0], "POST", path, {"decision": "dismissed", "clicked": True}, headers)
    assert status == 200 and replay == result and replay_headers["HUX-Replayed"] == "true"
    assert call(setup[0], "POST", path, {"decision": "acted", "clicked": True}, headers)[0] == 409
    moment["value"] += timedelta(days=1, seconds=1)
    shown_again = evaluate(setup, key="suggest-key-after-dismiss")[1]
    assert shown_again["suggestion"]["id"] == suggestion_id and shown_again["state"]["shows"] == 2
    invalid = [
        ({"decision": "acted", "clicked": False}, {"If-Match": "3", "Idempotency-Key": "decision-key-002"}),
        ({"decision": "silent", "clicked": True}, {"If-Match": "3", "Idempotency-Key": "decision-key-003"}),
        ({"decision": "acted", "clicked": True}, {"Idempotency-Key": "decision-key-004"}),
        ({"decision": "acted", "clicked": True}, {"If-Match": "3"}),
    ]
    for body, bad_headers in invalid:
        assert call(setup[0], "POST", path, body, bad_headers)[0] == 400


def test_never_again_is_explicit_and_permanent(setup):
    _, _, _, base, moment = setup
    shown = evaluate(setup)[1]
    path = base + f"/{shown['suggestion']['id']}/decisions"
    status, result, _ = call(setup[0], "POST", path, {"decision": "never_again", "clicked": True}, {"If-Match": "1", "Idempotency-Key": "never-key-0001"})
    assert status == 200 and result["state"]["never_again"] is True
    moment["value"] += timedelta(days=365)
    blocked = evaluate(setup, key="suggest-key-never")[1]
    assert blocked == {"suggestion": None, "reason": "never_again", "stored": False}


@pytest.mark.parametrize(("gate", "reason"), [
    ("no_store", "client_no_store"), ("sensitive", "client_no_store"), ("restricted", "client_no_store"),
])
def test_client_privacy_gates_do_not_write_state(setup, gate, reason):
    body = {"no_store": True} if gate == "no_store" else {"sensitivity": gate}
    status, result, _ = evaluate(setup, key=f"privacy-{gate}-1", **body)
    assert status == 200 and result == {"suggestion": None, "reason": reason, "stored": False}
    assert tenant(setup[0]).count(suggestions.FAMILY) == 0
    assert tenant(setup[0]).read(suggestions.FAMILY, "idempotency") == []


def test_server_privacy_state_overrides_client_claims(setup):
    router, _, conversation, base, _ = setup
    privacy.mark_topic(tenant(router), conversation["id"], "health")
    result = evaluate(setup, key="privacy-topic-01", sensitivity="public")[1]
    assert result["reason"] == "sensitive_topic" and tenant(router).count(suggestions.FAMILY) == 0
    privacy.set_flag(tenant(router), conversation["id"], "memory_disabled", True)
    assert evaluate(setup, key="privacy-disabled", sensitivity="public")[1]["reason"] == "conversation_no_store"
    privacy.set_flag(tenant(router), conversation["id"], "forgotten", True)
    assert evaluate(setup, key="privacy-forgotten", sensitivity="public")[1]["reason"] == "conversation_no_store"
    privacy.set_flag(tenant(router), conversation["id"], "forgotten", False)
    privacy.set_flag(tenant(router), conversation["id"], "memory_disabled", False)
    doc = tenant(router).get("privacy", privacy.TOPICS_DOC)
    doc["items"][conversation["id"]]["topics"] = []
    tenant(router).put("privacy", doc)
    current = tenant(router).get("conversations", conversation["id"])
    tenant(router).put("conversations", {**current, "mode": "private"}, current["revision"])
    result = call(router, "POST", base + "/evaluate", {"context": "idle", "sensitivity": "public"}, {"Idempotency-Key": "privacy-private1"})[1]
    assert result["reason"] == "private_mode"


def test_list_state_scope_and_limits(setup, monkeypatch):
    router, project, _, base, _ = setup
    shown = evaluate(setup)[1]
    status, states, headers = call(router, "GET", base + "/states")
    assert status == 200 and states["items"][0]["suggestion_id"] == shown["suggestion"]["id"]
    assert states["items"][0]["revision"] == 1 and headers["Cache-Control"] == "no-store"
    assert call(router, "GET", base + "/states", headers=OTHER)[0] == 404
    wrong = base.replace(project["id"], "prj_missing0000")
    assert call(router, "GET", wrong + "/states")[0] == 404
    _, conversation2, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C2", "project_id": project["id"]})
    base2 = f"/hux/v1/projects/{project['id']}/conversations/{conversation2['id']}/suggestions"
    monkeypatch.setattr(suggestions, "MAX_STATES", 0)
    result = call(router, "POST", base2 + "/evaluate", {"context": "idle"}, {"Idempotency-Key": "state-limit-key"})[1]
    assert result == {"suggestion": None, "reason": "state_limit", "stored": False}


def test_bad_context_body_and_unknown_decision_fail_closed(setup):
    router, _, _, base, _ = setup
    for body in ([], {"context": "nope"}, {"context": "idle", "priority": 100}):
        raw = json.dumps(body).encode()
        assert router.dispatch("POST", base + "/evaluate", {**HEADERS, "Idempotency-Key": "bad-context-key"}, raw).status == 400
    assert call(router, "POST", base + "/evaluate", {"context": "idle"})[0] == 400
    missing = base + "/sug_missing0000/decisions"
    assert call(router, "POST", missing, {"decision": "acted", "clicked": True}, {"If-Match": "1", "Idempotency-Key": "missing-decision"})[0] == 404


def test_defensive_validation_stale_act_and_private_inspection(setup):
    router, _, conversation, base, _ = setup
    with pytest.raises(Invalid, match="suggestion failed"):
        suggestions._suggestion("idle", "invalid-surface")
    with pytest.raises(Invalid, match="state failed"):
        suggestions._public({"schema": "hux.suggestion_state.v1", "owner": "raw-user", "suggestion_id": "sug_idle_workflow", "shows": 1, "never_again": False})
    shown = evaluate(setup)[1]
    path = base + f"/{shown['suggestion']['id']}/decisions"
    assert call(router, "POST", path, {"decision": "acted", "clicked": True}, {"If-Match": "9", "Idempotency-Key": "stale-decision1"})[0] == 409
    status, acted, _ = call(router, "POST", path, {"decision": "acted", "clicked": True}, {"If-Match": "1", "Idempotency-Key": "acted-decision1"})
    assert status == 200 and "acted_at" in acted["state"] and "dismissed_at" not in acted["state"]
    current = tenant(router).get("conversations", conversation["id"])
    tenant(router).put("conversations", {**current, "mode": "private"}, current["revision"])
    assert call(router, "POST", path, {"decision": "dismissed", "clicked": True}, {"If-Match": "2", "Idempotency-Key": "private-decision"})[0] == 404
    status, states, _ = call(router, "GET", base + "/states")
    assert status == 200 and states == {"items": [], "next": None}


def test_source_and_test_files_stay_bounded():
    assert len((FOUNDATION / "hux" / "suggestions.py").read_text().splitlines()) <= 500
    assert len(Path(__file__).read_text().splitlines()) <= 500