atlas-iac/testing/tests/test_hermes_hux_context_bootstrap.py

"""HUX-11 authenticated deterministic context bootstrap contract."""

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 contracts, foundation, identity, organization, store
from hux.server import build_router

ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
SUBJECT = "usr_0123456789abcdef"
BASE = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": SUBJECT}
RELAY = {**BASE, "X-Hux-Surface": "telegram", "X-Hux-Trust": "relay", "X-Hux-Relay-Key": "rk"}
WORKER = {**BASE, "X-Hux-Surface": "worker", "X-Hux-Trust": "worker", "X-Hux-Relay-Key": "wk"}
SCHEMAS = contracts.load_all()


def context_key(tmp_path: Path, value: bytes = b"k" * 32, mode: int = 0o600) -> Path:
    path = tmp_path / "context-key"
    path.write_bytes(value)
    path.chmod(mode)
    return path


def router_for(tmp_path: Path, key_path: Path | None = None):
    environ = {
        "HUX_FLAGS": ALL_ON,
        "HUX_RELAY_KEY": "rk",
        "HUX_WORKER_KEY": "wk",
        "HUX_ROUTER_KEY": "bk",
    }
    if key_path is not None:
        environ["HUX_CONTEXT_KEY_FILE"] = str(key_path)
    return build_router(tmp_path / "data", environ)


def payload(raw: str = "session-123", source: str = "home", who: identity.Identity | None = None) -> dict:
    who = who or identity.Identity("slot-3", SUBJECT, "worker", "worker")
    key = b"k" * 32
    return {
        "raw_session_id": raw,
        "project_source": source,
        "session_id": foundation.derive_context_id(key, "session", who, raw),
        "conversation_id": foundation.derive_context_id(key, "conversation", who, raw),
        "project_id": foundation.derive_context_id(key, "project", who, source),
    }


def call(router, headers, body, idem: str = "bootstrap:0001"):
    response = router.dispatch(
        "POST", "/hux/v1/context/bootstrap", {**headers, "Idempotency-Key": idem}, json.dumps(body).encode()
    )
    return response.status, response.body


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


def test_relay_creates_and_worker_replays_without_persisting_sources(tmp_path):
    router = router_for(tmp_path, context_key(tmp_path))
    body = payload()
    status, created = call(router, RELAY, body)
    assert status == 201
    assert created == {
        "schema": "hux.context_bootstrap.v1",
        "contract_version": "1.1.0",
        "identity": {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "telegram", "trust": "relay"},
        "session_id": body["session_id"],
        "conversation_id": body["conversation_id"],
        "project_id": body["project_id"],
        "created": {"project": True, "conversation": True},
        "revisions": {"project": 1, "conversation": 1},
    }
    assert contracts.validate_record(created, SCHEMAS) == []
    status, replay = call(router, WORKER, body)
    assert status == 200 and replay["created"] == {"project": False, "conversation": False}
    assert replay["identity"]["trust"] == "worker"
    written = " ".join(path.read_text() for path in (tmp_path / "data").rglob("*") if path.is_file())
    assert body["raw_session_id"] not in written and body["project_source"] not in written


def test_bootstrap_makes_conversation_event_appendable(tmp_path):
    router = router_for(tmp_path, context_key(tmp_path))
    body = payload()
    assert call(router, WORKER, body)[0] == 201
    response = router.dispatch(
        "POST",
        f"/hux/v1/conversations/{body['conversation_id']}/events",
        {**WORKER, "Idempotency-Key": "event:00000001"},
        json.dumps({"kind": "message.user", "summary": "hello"}).encode(),
    )
    assert response.status == 201 and response.body["seq"] == 1


def test_only_relay_or_worker_may_bootstrap_and_identities_do_not_cross(tmp_path):
    router = router_for(tmp_path, context_key(tmp_path))
    browser = {**BASE, "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "bk"}
    assert call(router, browser, payload())[0] == 403
    other = identity.Identity("slot-3", "usr_fedcba9876543210", "worker", "worker")
    headers = {**WORKER, "X-Hux-Subject": other.subject}
    assert call(router, headers, payload(), "bootstrap:other")[0] == 400
    assert call(router, headers, payload(who=other), "bootstrap:other2")[0] == 201


def test_id_proof_fields_and_idempotency_are_strict(tmp_path):
    router = router_for(tmp_path, context_key(tmp_path))
    good = payload()
    for name, bad in (
        ("session_id", "ses_" + "0" * 32),
        ("conversation_id", "prj_" + "0" * 32),
        ("project_id", "prj_short"),
        ("raw_session_id", "x" * 241),
        ("project_source", "bad value"),
    ):
        assert call(router, WORKER, {**good, name: bad}, f"bad:{name}:0001")[0] == 400
    assert call(router, WORKER, good, "")[0] == 400
    assert call(router, WORKER, {**good, "extra": True})[0] == 400
    assert call(router, WORKER, good, "bootstrap:same")[0] == 201
    assert call(router, WORKER, payload("session-456"), "bootstrap:same")[0] == 409
    response = router.dispatch(
        "POST", "/hux/v1/context/bootstrap", {**WORKER, "Idempotency-Key": "bootstrap:notobject"}, b"[]"
    )
    assert response.status == 400


def test_existing_owner_and_linkage_mismatches_are_never_overwritten(tmp_path):
    router = router_for(tmp_path, context_key(tmp_path))
    body = payload()
    assert call(router, WORKER, body)[0] == 201
    scoped = tenant(router)
    conversation = scoped.get("conversations", body["conversation_id"])
    scoped.put("conversations", {**conversation, "project_id": "prj_" + "f" * 32})
    assert call(router, WORKER, body, "bootstrap:link2")[0] == 409
    assert scoped.get("conversations", body["conversation_id"])["project_id"] == "prj_" + "f" * 32
    binding = scoped.get(foundation.CONTEXT_FAMILY, body["session_id"])
    scoped.put(foundation.CONTEXT_FAMILY, {**binding, "owner": "usr_fedcba9876543210"})
    assert call(router, WORKER, body, "bootstrap:owner2")[0] == 409


def test_existing_project_owner_and_contract_damage_fail_closed(tmp_path):
    router = router_for(tmp_path, context_key(tmp_path))
    body = payload()
    assert call(router, WORKER, body)[0] == 201
    scoped = tenant(router)
    project = scoped.get("projects", body["project_id"])
    scoped.put("projects", {**project, "owner": "usr_fedcba9876543210"})
    assert call(router, WORKER, body, "bootstrap:badowner")[0] == 409
    scoped.put("projects", {**project, "name": ""})
    assert call(router, WORKER, body, "bootstrap:badschema")[0] == 409


def test_bootstrap_honors_organization_caps_without_partial_creation(tmp_path, monkeypatch):
    router = router_for(tmp_path, context_key(tmp_path))
    monkeypatch.setattr(organization, "MAX_PROJECTS", 0)
    assert call(router, WORKER, payload())[0] == 409
    monkeypatch.setattr(organization, "MAX_PROJECTS", 200)
    monkeypatch.setattr(organization, "MAX_CONVERSATIONS", 0)
    assert call(router, WORKER, payload(), "bootstrap:convcap")[0] == 409
    assert tenant(router).count("projects") == 0


@pytest.mark.parametrize("kind", ["missing", "mode", "length", "symlink"])
def test_context_key_must_be_exact_owner_regular_0600_file(tmp_path, kind):
    key_path = tmp_path / "context-key"
    if kind == "mode":
        key_path = context_key(tmp_path, mode=0o640)
    elif kind == "length":
        key_path = context_key(tmp_path, b"short")
    elif kind == "symlink":
        target = context_key(tmp_path)
        key_path = tmp_path / "context-link"
        key_path.symlink_to(target)
    router = router_for(tmp_path, None if kind == "missing" else key_path)
    status, error = call(router, WORKER, payload())
    assert status == 403 and error["code"] == "forbidden"
    assert "kkkk" not in json.dumps(error)


def test_context_key_io_failures_are_sanitized(tmp_path, monkeypatch):
    path = context_key(tmp_path)
    monkeypatch.setattr(Path, "lstat", lambda self: (_ for _ in ()).throw(OSError("secret lstat detail")))
    with pytest.raises(Exception, match="unavailable") as denied:
        foundation._context_key({"HUX_CONTEXT_KEY_FILE": str(path)})
    assert "secret lstat" not in str(denied.value)
    monkeypatch.undo()
    monkeypatch.setattr(Path, "read_bytes", lambda self: (_ for _ in ()).throw(OSError("secret read detail")))
    with pytest.raises(Exception, match="unavailable") as denied:
        foundation._context_key({"HUX_CONTEXT_KEY_FILE": str(path)})
    assert "secret read" not in str(denied.value)


def test_foundation_capabilities_and_manifest_still_work(tmp_path):
    router = router_for(tmp_path, context_key(tmp_path))
    headers = {**BASE, "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "bk"}
    capabilities = router.dispatch("GET", "/hux/v1/capabilities", headers, b"")
    manifest = router.dispatch("GET", "/hux/v1/manifest", headers, b"")
    assert capabilities.status == 200 and capabilities.body["schema"] == "hux.capabilities.v1"
    assert manifest.status == 200 and manifest.body["schema"] == "hux.manifest.v1"