diff --git a/dockerfiles/hermes-hux-foundation/hux/events.py b/dockerfiles/hermes-hux-foundation/hux/events.py new file mode 100644 index 00000000..3df1b728 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/events.py @@ -0,0 +1,283 @@ +"""HUX-01 activity events: the per-conversation append-only log and its readers. + +``emit`` is the one write path every family uses. It runs the redaction +pipeline, allocates ``seq`` under the conversation lock, honours idempotency +keys and keeps a checkpoint document per conversation. Readers page by +``after_seq`` or follow an SSE stream whose ``id:`` is the seq, and every +served record passes serve-time redaction for the caller's surface. +""" + +from __future__ import annotations + +import json +import threading +import time +from typing import Any +from collections.abc import Iterator + +from hux import contracts, redaction +from hux.errors import Invalid, NotFound, TooLarge +from hux.http import Request, Response, Router, page +from hux.identity import Identity +from hux.store import TenantStore, new_id, now_iso + +FAMILY = "events" +SEQ_FAMILY = "events_seq" +IDEM_FAMILY = "events_idem" +PAGE_MAX = 200 +PAGE_DEFAULT = 100 +STREAM_MAX_POLLS = 900 +STREAM_POLL_SECONDS = 1.0 +SCHEMAS = contracts.load_all() +KINDS = frozenset(contracts.load_schema("event.schema.json")["properties"]["kind"]["enum"]) +USER_KINDS = frozenset({"message.user", "approval.resolved", "run.cancelled", "memory.forgotten", "suggestion.dismissed"}) +_streams: dict[str, int] = {} +_streams_guard = threading.Lock() + + +def _seq_id(conversation_id: str) -> str: + return f"seq_{conversation_id}" + + +def is_private(store: TenantStore, conversation_id: str) -> bool: + """True when the conversation document (HUX-03) says the mode is private (SO-28).""" + try: + return store.get("conversations", conversation_id).get("mode") == "private" + except NotFound: + return False + + +def conversation_known(store: TenantStore, conversation_id: str) -> bool: + """Ownership check (SO-18): the conversation exists somewhere in this subject's tree.""" + return ( + store.exists("conversations", conversation_id) + or store.exists(SEQ_FAMILY, _seq_id(conversation_id)) + ) + + +def _actor(identity: Identity, kind: str) -> dict[str, str]: + if kind in USER_KINDS and identity.trust != "worker": + return {"type": "user", "id": identity.subject} + if identity.trust == "worker": + return {"type": "system", "id": "hux-worker"} + return {"type": "assistant", "id": "hermes"} + + +def _find_replay(store: TenantStore, conversation_id: str, key: str) -> dict[str, Any] | None: + for row in store.read(IDEM_FAMILY, conversation_id): + if row.get("idempotency_key") == key: + for event in store.read(FAMILY, conversation_id): + if event.get("id") == row.get("event_id"): + return event + return None + + +def build(identity: Identity, conversation_id: str, kind: str, summary: str, detail: dict | None, evidence: list | None, + sensitivity: str, run_id: str | None, turn: int | None, correlation_id: str | None, idempotency_key: str | None) -> dict[str, Any]: + """Run the redaction pipeline and shape an unsequenced ``hux.event.v1`` record.""" + if kind not in KINDS: + raise Invalid(f"unknown event kind {kind!r}") + if sensitivity not in ("public", "personal", "sensitive", "restricted"): + raise Invalid("unknown sensitivity") + hits: list[str] = [] + summary_clean = redaction.scrub_value(str(summary or "").strip(), hits)[: redaction.SUMMARY_MAX] or "[empty]" + detail_clean = redaction.scrub_value(redaction.filter_detail(kind, detail), hits) + detail_clean, truncated = redaction.cap_detail(detail_clean) + evidence_clean = redaction.scrub_value(redaction.filter_evidence(evidence), hits) + stamp = now_iso() + record: dict[str, Any] = { + "schema": "hux.event.v1", + "id": new_id("evt"), + "seq": 0, + "ts": stamp, + "conversation_id": conversation_id, + "kind": kind, + "summary": summary_clean, + "provenance": {"surface": identity.surface, "actor": _actor(identity, kind), "recorded_at": stamp, "conversation_id": conversation_id}, + "sensitivity": sensitivity, + "redaction": redaction.derive_level(sensitivity, hits, truncated), + "turn": max(0, int(turn or 0)), + "identity": identity.record(), + } + if detail_clean: + record["detail"] = detail_clean + if evidence_clean: + record["evidence"] = evidence_clean + if run_id: + record["run_id"] = str(run_id)[:120] + record["provenance"]["run_id"] = record["run_id"] + if correlation_id: + record["correlation_id"] = str(correlation_id)[:120] + if idempotency_key: + record["idempotency_key"] = idempotency_key + return record + + +def emit_with_status(store: TenantStore, identity: Identity, conversation_id: str, kind: str, summary: str, detail: dict | None = None, + evidence: list | None = None, sensitivity: str = "personal", run_id: str | None = None, turn: int | None = None, + correlation_id: str | None = None, idempotency_key: str | None = None) -> tuple[dict[str, Any] | None, bool]: + """Like ``emit`` but also says whether the record was an idempotent replay. None means private mode (nothing written).""" + if is_private(store, conversation_id): + return None, False + record = build(identity, conversation_id, kind, summary, detail, evidence, sensitivity, run_id, turn, correlation_id, idempotency_key) + with store.lock(f"{FAMILY}:{conversation_id}"): + if idempotency_key: + existing = _find_replay(store, conversation_id, idempotency_key) + if existing is not None: + return existing, True + seq_id = _seq_id(conversation_id) + checkpoint = store.get(SEQ_FAMILY, seq_id) if store.exists(SEQ_FAMILY, seq_id) else {"id": seq_id, "next_seq": 1, "last_event_id": ""} + record["seq"] = int(checkpoint["next_seq"]) + problems = contracts.validate_record(record, SCHEMAS) + if problems: + raise Invalid("event failed contract validation", problems) + line = json.dumps(record, sort_keys=True, separators=(",", ":")).encode() + if len(line) > redaction.LINE_CAP_BYTES: + raise TooLarge("event line exceeds 64 KiB") + store.append(FAMILY, conversation_id, record) + store.put(SEQ_FAMILY, {**checkpoint, "next_seq": record["seq"] + 1, "last_event_id": record["id"], "checkpointed_at": now_iso()}) + if idempotency_key: + store.append(IDEM_FAMILY, conversation_id, {"idempotency_key": idempotency_key, "event_id": record["id"], "seq": record["seq"], "at": record["ts"]}) + return record, False + + +def emit(store: TenantStore, identity: Identity, conversation_id: str, kind: str, summary: str, detail: dict | None = None, + evidence: list | None = None, sensitivity: str = "personal", run_id: str | None = None, turn: int | None = None, + correlation_id: str | None = None, idempotency_key: str | None = None) -> dict[str, Any] | None: + """Append one event and return the stored record (None when the conversation is private).""" + record, _ = emit_with_status(store, identity, conversation_id, kind, summary, detail, evidence, sensitivity, run_id, turn, correlation_id, idempotency_key) + return record + + +def read_after(store: TenantStore, conversation_id: str, after_seq: int, limit: int) -> list[dict[str, Any]]: + """Stored events with ``seq > after_seq``, oldest first, at most ``limit``.""" + rows = [row for row in store.read(FAMILY, conversation_id) if int(row.get("seq", 0)) > after_seq] + return rows[:limit] + + +def rewrite_full(store: TenantStore, conversation_id: str, summary: str, reason: str, match=None) -> int: + """Rewrite matching events of one conversation to ``redaction.level: full`` (SO-24, forget, decay). Returns the count.""" + with store.lock(f"{FAMILY}:{conversation_id}"): + rows = store.read(FAMILY, conversation_id) + changed = 0 + out: list[dict[str, Any]] = [] + for row in rows: + if row.get("redaction", {}).get("level") != "full" and (match is None or match(row)): + row = redaction.full_redaction(row, summary, reason) + changed += 1 + out.append(row) + if changed: + store.rewrite(FAMILY, conversation_id, out) + return changed + + +def redact_memory_references(store: TenantStore, memory_id: str) -> int: + """Fully redact every event, in any conversation, that names a forgotten memory id (SO-24).""" + + def references(row: dict[str, Any]) -> bool: + if row.get("detail", {}).get("memory_id") == memory_id: + return True + return any(ref.get("kind") == "memory" and ref.get("id") == memory_id for ref in row.get("evidence", [])) + + return sum(rewrite_full(store, name, "[forgotten memory]", "memory forgotten", references) for name in store.ledgers(FAMILY)) + + +# -- routes ------------------------------------------------------------------ + +def _int_query(request: Request, name: str, default: int, floor: int = 0) -> int: + raw = request.query.get(name, "") + if raw == "": + return default + if not raw.lstrip("-").isdigit(): + raise Invalid(f"{name} must be an integer") + return max(floor, int(raw)) + + +def _require_conversation(request: Request) -> str: + conversation_id = request.params["id"] + if not conversation_known(request.store, conversation_id): + raise NotFound("conversation not found") + return conversation_id + + +def list_events(request: Request) -> Response: + """``GET /hux/v1/conversations/{id}/events?after_seq=&limit=``: one page, ``next`` is the last seq served.""" + conversation_id = _require_conversation(request) + after_seq = _int_query(request, "after_seq", 0) + limit = min(_int_query(request, "limit", PAGE_DEFAULT, 1), PAGE_MAX) + rows = read_after(request.store, conversation_id, after_seq, limit) + items = [redaction.redact_record(row, request.identity.surface) for row in rows] + request.audit("events.list", conversation_id) + return page(items, items[-1]["seq"] if len(items) == limit else None) + + +def append_event(request: Request) -> Response: + """``POST /hux/v1/conversations/{id}/events``: append from the trusted hop; ids and seq are server-assigned (SO-15).""" + conversation_id = _require_conversation(request) + body = request.body if isinstance(request.body, dict) else None + if body is None: + raise Invalid("body must be an object") + if "id" in body or "seq" in body: + raise Invalid("id and seq are server-assigned") + if not isinstance(body.get("kind"), str) or not isinstance(body.get("summary"), str): + raise Invalid("kind and summary are required") + detail = body.get("detail") if isinstance(body.get("detail"), dict) else None + evidence = body.get("evidence") if isinstance(body.get("evidence"), list) else None + turn = body.get("turn") if isinstance(body.get("turn"), int) else None + record, replayed = emit_with_status( + request.store, request.identity, conversation_id, body["kind"], body["summary"], detail, evidence, + body.get("sensitivity", "personal"), body.get("run_id"), turn, body.get("correlation_id"), request.idempotency_key() or None, + ) + if record is None: + request.audit("events.append", conversation_id, "allow", "private_mode") + return Response(204) + request.audit("events.append", f"{conversation_id}/{record['id']}", "allow", "replayed" if replayed else "") + served = redaction.redact_record(record, request.identity.surface) + return Response(200 if replayed else 201, served, {"HUX-Replayed": "true"} if replayed else {}) + + +def _sse(record: dict[str, Any]) -> bytes: + return f"id: {record['seq']}\nevent: {record['kind']}\ndata: {json.dumps(record, sort_keys=True)}\n\n".encode() + + +def stream_events(request: Request) -> Response: + """``GET /hux/v1/conversations/{id}/events/stream``: SSE replay from ``Last-Event-ID`` (or ``after_seq``), then a bounded live poll. + + ``max_polls`` and ``poll_ms`` (query) bound the live phase so a client, or a + test, decides how long to wait; the server caps them at the 15 minute idle + limit (SO-17). A second stream for the same (subject, conversation) closes + the first. + """ + conversation_id = _require_conversation(request) + last_id = request.header("Last-Event-ID") + cursor = int(last_id) if last_id.isdigit() else _int_query(request, "after_seq", 0) + max_polls = min(_int_query(request, "max_polls", STREAM_MAX_POLLS), STREAM_MAX_POLLS) + poll_seconds = min(_int_query(request, "poll_ms", int(STREAM_POLL_SECONDS * 1000)), 60000) / 1000 + store, surface = request.store, request.identity.surface + key = f"{request.identity.subject}:{conversation_id}" + with _streams_guard: + token = _streams[key] = _streams.get(key, 0) + 1 + request.audit("events.stream", conversation_id) + + def generate() -> Iterator[bytes]: + position = cursor + yield b"retry: 2000\n\n" + polls = 0 + while True: + for record in read_after(store, conversation_id, position, PAGE_MAX): + position = record["seq"] + yield _sse(redaction.redact_record(record, surface)) + if polls >= max_polls or _streams.get(key) != token: + break + polls += 1 + yield b": keepalive\n\n" + time.sleep(poll_seconds) + + return Response(200, stream=generate) + + +def register(router: Router) -> None: + """Attach HUX-01 routes.""" + router.add("GET", "/hux/v1/conversations/{id}/events", "HUX-01", "events.list", list_events) + router.add("POST", "/hux/v1/conversations/{id}/events", "HUX-01", "events.append", append_event) + router.add("GET", "/hux/v1/conversations/{id}/events/stream", "HUX-01", "events.stream", stream_events) diff --git a/dockerfiles/hermes-hux-foundation/hux/memory.py b/dockerfiles/hermes-hux-foundation/hux/memory.py new file mode 100644 index 00000000..a4c730fa --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/memory.py @@ -0,0 +1,387 @@ +"""HUX-02 memory control: an append-only ledger with explicit user consent. + +Every entry is a ``hux.memory.v1`` snapshot. The newest snapshot per id is +kept as a revisioned document (so If-Match works) and every snapshot is also +appended to ``ledger.jsonl``. Status moves only along +``rules.MEMORY_TRANSITIONS``; ``no_store`` and ``forgotten`` write a +content-free line plus a tombstone before anything is returned (SO-22), and +``retrieve`` consults the tombstones before the documents (SO-23). +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any + +from hux import contracts, redaction, rules +from hux.errors import Conflict, Forbidden, Invalid, NotFound +from hux.http import Request, Response, Router, page +from hux.identity import Identity +from hux.store import TenantStore, new_id, now_iso + +FAMILY = "memory" +IDEM_FAMILY = "memory_idem" +LEDGER = "ledger" +TOMBSTONES = "tombstones" +DEFAULT_CONVERSATION = "conv_memory" +HUMAN_SURFACES = frozenset({"chat", "telegram", "voice"}) +ACTIONS = ("approve", "reject", "forget", "edit", "remove_retrieval", "restore_retrieval") +KINDS = ("preference", "fact", "instruction", "context") +DEFAULT_DECAY_DAYS = 180 +SCHEMAS = contracts.load_all() + + +def _parse(stamp: str) -> datetime: + return datetime.fromisoformat(stamp.replace("Z", "+00:00")) + + +def _emit(store: TenantStore, identity: Identity, record: dict[str, Any], kind: str, summary: str) -> None: + from hux import events + + detail = {"memory_id": record["id"], "kind": record["kind"], "sensitivity": record["sensitivity"], "topic": record.get("topic", "general")} + conversation_id = record["provenance"].get("conversation_id", DEFAULT_CONVERSATION) + events.emit(store, identity, conversation_id, kind, summary, detail, [{"kind": "memory", "id": record["id"]}], sensitivity=record["sensitivity"]) + + +def tombstoned(store: TenantStore) -> set[str]: + """Ids that must never surface again, whatever the documents say.""" + return {row["memory_id"] for row in store.read(FAMILY, TOMBSTONES)} + + +def _tombstone(store: TenantStore, memory_id: str, reason: str) -> None: + store.append(FAMILY, TOMBSTONES, {"memory_id": memory_id, "at": now_iso(), "reason": reason, "purged": False}) + + +def effective_expiry(record: dict[str, Any]) -> datetime | None: + """When the entry stops being active: ``expires_at``, or decay from the last approval (or creation).""" + ttl = record["ttl"] + if ttl["policy"] == "expires_at": + return _parse(ttl["expires_at"]) + if ttl["policy"] == "decay": + approvals = [row["at"] for row in record["audit"] if row["action"] == "approved"] + return _parse(approvals[-1] if approvals else record["created_at"]) + timedelta(days=ttl["decay_days"]) + return None + + +def _persist(store: TenantStore, record: dict[str, Any], expected: int | None) -> dict[str, Any]: + """Validate against contract and rules, then write document + ledger line under the family lock.""" + problems = rules.memory_policy_violations(record) + contracts.validate_record({**record, "revision": record.get("revision") or 1}, SCHEMAS) + if problems: + raise Invalid("memory entry violates policy", problems) + with store.lock(FAMILY): + stored = store.put(FAMILY, record, expected) + store.append(FAMILY, LEDGER, stored) + return stored + + +def _transition(store: TenantStore, identity: Identity, record: dict[str, Any], target: str, action: str, actor: dict[str, str], + note: str = "", expected: int | None = None, **changes: Any) -> dict[str, Any]: + if not rules.transition_allowed(rules.MEMORY_TRANSITIONS, record["status"], target): + raise Conflict(f"{record['status']} -> {target} is not a legal memory transition") + stamp = now_iso() + updated = {**record, **changes, "status": target, "updated_at": stamp, "audit": [*record["audit"], {"at": stamp, "action": action, "actor": actor, **({"note": note[:200]} if note else {})}]} + if target in {"forgotten", "rejected", "expired"}: + updated["content"], updated["retrievable"] = "", False + if target == "active": + updated["retrievable"] = changes.get("retrievable", True) + stored = _persist(store, updated, expected) + if target == "forgotten": + _tombstone(store, stored["id"], action) + return stored + + +def load(store: TenantStore, memory_id: str, now: datetime | None = None) -> dict[str, Any]: + """Newest snapshot with lazy expiry (SO-27): an entry past its expiry is written as expired before it is served.""" + record = store.get(FAMILY, memory_id) + if record["status"] == "active": + expiry = effective_expiry(record) + if expiry is not None and expiry <= (now or datetime.now(timezone.utc)): + record = _transition(store, None, record, "expired", "expired", {"type": "system", "id": "retention"}, expected=record["revision"]) + return record + + +def expire_due(store: TenantStore, now: datetime) -> int: + """Retention pass: move every active entry past its expiry to expired. Returns the count.""" + before = {row["id"]: row["status"] for row in store.scan(FAMILY)} + return sum(1 for memory_id, status in before.items() if status == "active" and load(store, memory_id, now)["status"] == "expired") + + +def purge_forgotten(store: TenantStore) -> int: + """Retention pass (SO-47): blank ``content`` on every ledger snapshot of tombstoned ids, then mark them purged.""" + with store.lock(FAMILY): + stones = store.read(FAMILY, TOMBSTONES) + pending = {row["memory_id"] for row in stones if not row.get("purged")} + if not pending: + return 0 + rows = store.read(FAMILY, LEDGER) + store.rewrite(FAMILY, LEDGER, [{**row, "content": ""} if row.get("id") in pending else row for row in rows]) + store.rewrite(FAMILY, TOMBSTONES, [{**row, "purged": True} for row in stones]) + return len(pending) + + +def forget_from_conversation(store: TenantStore, identity: Identity, conversation_id: str) -> list[str]: + """Forget every entry sourced from ``conversation_id`` (used by the privacy forget); returns the ids.""" + from hux import events + + forgotten: list[str] = [] + for row in list(store.scan(FAMILY)): + if row["provenance"].get("conversation_id") != conversation_id: + continue + if row["status"] in {"proposed"}: + row = _transition(store, identity, row, "rejected", "rejected", {"type": "system", "id": "forget"}, "conversation forgotten") + if row["status"] in {"active", "expired"}: + row = _transition(store, identity, row, "forgotten", "forgotten", {"type": "system", "id": "forget"}, "conversation forgotten") + if row["status"] not in {"forgotten", "no_store"}: + _tombstone(store, row["id"], "conversation forgotten") + forgotten.append(row["id"]) + events.redact_memory_references(store, row["id"]) + return forgotten + + +def retrieve(store: TenantStore, query_terms: list[str], scope: dict[str, Any] | None = None) -> list[dict[str, Any]]: + """Agent read hook (SO-23): tombstones first, then only active + retrievable entries in scope; never no_store, forgotten or expired.""" + from hux import privacy + + stones = tombstoned(store) + blocked = privacy.blocked_conversations(store) + if scope and scope.get("level") == "conversation" and scope.get("scope_id") in blocked: + return [] + terms = [term.lower() for term in query_terms if term] + hits: list[dict[str, Any]] = [] + for row in store.scan(FAMILY): + if row["id"] in stones or row["provenance"].get("conversation_id") in blocked: + continue + record = load(store, row["id"]) + if record["status"] != "active" or not record["retrievable"]: + continue + if scope and record["scope"]["level"] != "global" and record["scope"] != scope: + continue + text = record["content"].lower() + if terms and not any(term in text for term in terms): + continue + hits.append(record) + return sorted(hits, key=lambda r: r["updated_at"], reverse=True) + + +# -- proposals ------------------------------------------------------------------ + +def _human(identity: Identity) -> bool: + return identity.surface in HUMAN_SURFACES and identity.trust != "worker" + + +def _actor(request: Request) -> dict[str, str]: + if _human(request.identity) and request.body.get("proposed_by", "assistant") == "user": + return {"type": "user", "id": request.identity.subject} + return {"type": "assistant", "id": "hermes"} + + +def _shape(request: Request, body: dict[str, Any]) -> tuple[dict[str, Any], str]: + """Build the entry from a proposal body and decide its approval mode from policy; returns (entry, gate_reason).""" + from hux import privacy + + if "id" in body or "revision" in body or "status" in body: + raise Invalid("id, revision and status are server-assigned") + content = body.get("content") + if not isinstance(content, str) or not content.strip() or len(content) > 2000: + raise Invalid("content must be a non-empty string of at most 2000 chars") + hits: list[str] = [] + content = redaction.scrub_value(content.strip(), hits) + detected = privacy.detect_topics(content) + (["credentials"] if hits else []) + topic = body.get("topic") if body.get("topic") in rules.PRIVACY_TOPICS or body.get("topic") == "general" else None + topic = topic if topic and topic != "general" else (detected[0] if detected else "general") + sensitivity = body.get("sensitivity", "personal") + if sensitivity not in privacy.SENSITIVITY_RANK: + raise Invalid("unknown sensitivity") + floor = privacy.topic_sensitivity(detected + ([topic] if topic != "general" else [])) + if privacy.SENSITIVITY_RANK[floor] > privacy.SENSITIVITY_RANK[sensitivity]: + sensitivity = floor + scope = body.get("scope") if isinstance(body.get("scope"), dict) else {"level": "global"} + source = body.get("source") if isinstance(body.get("source"), dict) else {"kind": "message", "id": "unspecified"} + ttl = body.get("ttl") if isinstance(body.get("ttl"), dict) else None + if ttl is None or (sensitivity == "sensitive" and ttl.get("policy") == "never"): + days = rules.PRIVACY_TOPICS[topic]["decay_days"] if topic in rules.PRIVACY_TOPICS else DEFAULT_DECAY_DAYS + ttl = {"policy": "decay", "decay_days": days} + actor = _actor(request) + stamp = now_iso() + conversation_id = body.get("conversation_id") if isinstance(body.get("conversation_id"), str) else None + provenance = {"surface": request.identity.surface, "actor": actor, "recorded_at": stamp} + if conversation_id: + provenance["conversation_id"] = conversation_id + if isinstance(body.get("run_id"), str): + provenance["run_id"] = body["run_id"][:120] + entry: dict[str, Any] = { + "schema": "hux.memory.v1", "id": new_id("mem"), "owner": request.identity.subject, "scope": scope, + "kind": body.get("kind") if body.get("kind") in KINDS else "fact", "content": content, "status": "proposed", + "approval_mode": "ask", "sensitivity": sensitivity, "topic": topic, "ttl": ttl, "source": source, + "provenance": provenance, "created_at": stamp, "updated_at": stamp, + "audit": [{"at": stamp, "action": "proposed", "actor": actor}], "reason": str(body.get("reason") or "Proposed to be remembered.")[:280], + "retrievable": False, "identity": request.identity.record(), + } + if isinstance(body.get("supersedes"), str): + entry["supersedes"] = body["supersedes"] + allowed, why = privacy.memory_write_allowed(request.store, conversation_id, entry) + if why == "private_mode": + raise Forbidden("memory writes are refused in private mode") + references_forgotten = entry.get("supersedes") in tombstoned(request.store) or source.get("id") in tombstoned(request.store) + wants = body.get("approval_mode", "ask" if actor["type"] == "assistant" else "automatic") + if not allowed or wants == "no_store": + # Content-free decision record. The topic moves into the note because + # rules.memory_policy_violations reads a deny topic on any non-rejected + # status as a write, and a sensitive entry must say approval_mode=ask. + why = why if not allowed else "declined" + entry.pop("topic") + entry.update(status="no_store", approval_mode="ask" if sensitivity == "sensitive" else "no_store", content="", retrievable=False, + audit=[{"at": stamp, "action": "no_store", "actor": actor, "note": f"{why}; topic={topic}"[:200]}]) + elif sensitivity == "sensitive" or wants == "ask" or references_forgotten or actor["type"] != "user": + entry["approval_mode"] = "ask" + else: + entry.update(status="active", approval_mode="automatic", retrievable=True) + entry["audit"].append({"at": stamp, "action": "approved", "actor": actor, "note": "automatic"}) + return entry, why + + +def post_memory(request: Request) -> Response: + """``POST /hux/v1/memory``: propose an entry; policy decides active, proposed (ask) or no_store (202).""" + body = request.body if isinstance(request.body, dict) else None + if body is None: + raise Invalid("body must be an object") + key = request.idempotency_key() + if key: + for row in request.store.read(IDEM_FAMILY, "keys"): + if row["idempotency_key"] == key: + request.audit("memory.propose", row["memory_id"], "allow", "replayed") + return Response(200, load(request.store, row["memory_id"]), {"HUX-Replayed": "true"}) + entry, why = _shape(request, body) + try: + stored = _persist(request.store, entry, None) + except Invalid: + request.store.append(FAMILY, TOMBSTONES, {"memory_id": entry["id"], "at": now_iso(), "reason": "policy_violation", "purged": True}) + _emit(request.store, request.identity, entry, "memory.suppressed", "Memory proposal suppressed by policy") + raise + if key: + request.store.append(IDEM_FAMILY, "keys", {"idempotency_key": key, "memory_id": stored["id"], "at": stored["created_at"]}) + if stored["status"] == "no_store": + _tombstone(request.store, stored["id"], why) + _emit(request.store, request.identity, stored, "memory.suppressed", f"Not remembered ({why})") + request.audit("memory.propose", stored["id"], "allow", "no_store") + return Response(202, stored) + kind = "memory.committed" if stored["status"] == "active" else "memory.proposed" + _emit(request.store, request.identity, stored, kind, "Remembered" if stored["status"] == "active" else "Proposed to remember") + request.audit("memory.propose", stored["id"]) + return Response(201, stored) + + +def list_memory(request: Request) -> Response: + """``GET /hux/v1/memory?status=&scope=``: owner-scoped list; ``scope`` is ``level`` or ``level:scope_id``.""" + status, scope = request.query.get("status"), request.query.get("scope") + level, _, scope_id = (scope or "").partition(":") + items = [] + for row in list(request.store.scan(FAMILY)): + record = load(request.store, row["id"]) + if status and record["status"] != status: + continue + if level and (record["scope"]["level"] != level or (scope_id and record["scope"].get("scope_id") != scope_id)): + continue + items.append(record) + request.audit("memory.list", "memory") + return page(sorted(items, key=lambda r: r["updated_at"], reverse=True), None) + + +def get_memory(request: Request) -> Response: + """``GET /hux/v1/memory/{id}``.""" + memory_id = request.params["id"] + try: + record = load(request.store, memory_id) + except (NotFound, Invalid) as error: + raise NotFound("memory entry not found") from error + request.audit("memory.get", memory_id) + return Response(200, record, {"ETag": str(record["revision"])}) + + +def _edit(request: Request, record: dict[str, Any], actor: dict[str, str], expected: int | None) -> dict[str, Any]: + body = request.body if isinstance(request.body, dict) else {} + content = body.get("content") + if not isinstance(content, str) or not content.strip(): + raise Invalid("edit needs new content") + old = _transition(request.store, request.identity, record, "forgotten", "superseded", actor, "edited", expected) + from hux import events + + events.redact_memory_references(request.store, old["id"]) + stamp = now_iso() + fresh = { + **{k: v for k, v in record.items() if k not in {"revision", "supersedes"}}, "id": new_id("mem"), "content": redaction.scrub_value(content.strip(), [])[:2000], + "status": "active", "approval_mode": "automatic", "retrievable": True, "supersedes": old["id"], "created_at": stamp, "updated_at": stamp, + "audit": [{"at": stamp, "action": "edited", "actor": actor, "note": f"supersedes {old['id']}"}, {"at": stamp, "action": "approved", "actor": actor}], + } + if fresh["sensitivity"] == "sensitive": + fresh.update(status="proposed", approval_mode="ask", retrievable=False) + fresh["audit"] = fresh["audit"][:1] + return _persist(request.store, fresh, None) + + +def act_memory(request: Request) -> Response: + """``POST /hux/v1/memory/{id}/{action}``: approve, reject, forget, edit, remove_retrieval, restore_retrieval; every move emits a memory.* event.""" + memory_id, action = request.params["id"], request.params["action"] + if action not in ACTIONS: + raise NotFound("unknown memory action") + if not _human(request.identity): + raise Forbidden("memory decisions come from a human surface") + try: + record = load(request.store, memory_id) + except (NotFound, Invalid) as error: + raise NotFound("memory entry not found") from error + expected = request.if_match() + if expected is not None and expected != record["revision"]: + raise Conflict(f"revision {expected} does not match current revision {record['revision']}", [str(record["revision"])]) + actor = {"type": "user", "id": request.identity.subject} + store, identity = request.store, request.identity + if action == "approve": + stored = _transition(store, identity, record, "active", "approved", actor, expected=expected) + _emit(store, identity, stored, "memory.committed", "Memory approved") + elif action == "reject": + stored = _transition(store, identity, record, "rejected", "rejected", actor, expected=expected) + _emit(store, identity, stored, "memory.suppressed", "Memory rejected") + elif action == "forget": + stored = _transition(store, identity, record, "forgotten", "forgotten", actor, expected=expected) + from hux import events + + events.redact_memory_references(store, stored["id"]) + _emit(store, identity, stored, "memory.forgotten", "Memory forgotten") + elif action == "edit": + stored = _edit(request, record, actor, expected) + _emit(store, identity, stored, "memory.committed" if stored["status"] == "active" else "memory.proposed", f"Memory edited, supersedes {record['id']}") + else: + if record["status"] != "active": + raise Conflict("retrieval can only change on active entries") + retrievable = action == "restore_retrieval" + stamp = now_iso() + stored = _persist(store, {**record, "retrievable": retrievable, "updated_at": stamp, "audit": [*record["audit"], {"at": stamp, "action": "retrieval_removed" if not retrievable else "approved", "actor": actor, "note": "retrieval restored" if retrievable else ""}]}, expected) + _emit(store, identity, stored, "memory.retrieval_removed" if not retrievable else "memory.committed", "Retrieval removed" if not retrievable else "Retrieval restored") + request.audit(f"memory.{action}", memory_id, "allow", "" if expected is not None else "unconditional_write") + return Response(200, stored, {"ETag": str(stored["revision"])}) + + +def export_memory(request: Request) -> Response: + """``GET /hux/v1/memory/export``: active, retrievable entries only, each stamped ``exported`` (SO-26).""" + stones = tombstoned(request.store) + items = [] + for row in list(request.store.scan(FAMILY)): + record = load(request.store, row["id"]) + if record["id"] in stones or record["status"] != "active" or not record["retrievable"]: + continue + stamp = now_iso() + items.append(_persist(request.store, {**record, "audit": [*record["audit"], {"at": stamp, "action": "exported", "actor": {"type": "user", "id": request.identity.subject}}]}, record["revision"])) + request.store.append(FAMILY, "exports", {"at": now_iso(), "ids": [item["id"] for item in items]}) + request.audit("memory.export", "memory") + response = page(items, None) + response.headers["Content-Disposition"] = 'attachment; filename="hux-memory-export.json"' + return response + + +def register(router: Router) -> None: + """Attach HUX-02 routes.""" + router.add("GET", "/hux/v1/memory/export", "HUX-02", "memory.export", export_memory) + router.add("POST", "/hux/v1/memory", "HUX-02", "memory.propose", post_memory) + router.add("GET", "/hux/v1/memory", "HUX-02", "memory.list", list_memory) + router.add("GET", "/hux/v1/memory/{id}", "HUX-02", "memory.get", get_memory) + router.add("POST", "/hux/v1/memory/{id}/{action}", "HUX-02", "memory.act", act_memory) diff --git a/dockerfiles/hermes-hux-foundation/hux/privacy.py b/dockerfiles/hermes-hux-foundation/hux/privacy.py new file mode 100644 index 00000000..48619c47 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/privacy.py @@ -0,0 +1,260 @@ +"""HUX-10 privacy behaviour: topic detection, notices, conversation forget and the retention job. + +The policy itself lives in ``rules.privacy_policy``; this module records what +was shown, scopes detected topics to the conversation they appeared in, gates +memory writes for other families (``memory_write_allowed``) and runs the +daily retention audit that expires memory, decays topic context and purges +forgotten content while never touching the audit ledgers (SO-47). +""" + +from __future__ import annotations + +import re +from datetime import datetime, timedelta, timezone +from typing import Any + +from hux import contracts, rules +from hux.errors import Invalid, NotFound +from hux.http import Request, Response, Router, page +from hux.identity import Identity +from hux.store import TenantStore, new_id, now_iso + +FAMILY = "privacy" +RETENTION_FAMILY = "retention" +TOPICS_DOC = "topics_conversations" +TOPICS_SCHEMA = "hux.conversation_topics.v1" +CONTROLS = ("forget_this_conversation", "switch_to_private", "disable_memory_here", "dismiss") +SENSITIVITY_RANK = {"public": 0, "personal": 1, "sensitive": 2, "restricted": 3} +AUDIT_STALE_HOURS = 48 +SCHEMAS = contracts.load_all() + +# Deliberately narrow: a false positive only asks the user before remembering, +# a false negative would store something it should not. Detection is a hint, +# never proof, so callers treat "uncertain" as sensitive (SO-21). +TOPIC_PATTERNS: dict[str, re.Pattern[str]] = { + "credentials": re.compile(r"(?i)\b(password|passphrase|api[ _-]?key|secret key|private key|token|2fa|otp|ssh key|login credentials?)\b"), + "biometric": re.compile(r"(?i)\b(fingerprint|face ?id|retina|iris scan|voice ?print|dna|genome|biometric)\b"), + "minors": re.compile(r"(?i)\b(my (son|daughter|kid|kids|child|children|toddler|baby)|(?:\d{1,2}|[a-z]+)[- ]year[- ]old (?:son|daughter|boy|girl|child)|school ?(?:pickup|run)|daycare|minor child)\b"), + "location": re.compile(r"(?i)\b(home address|my address|lives? at \d|gps|coordinates|\d{1,5} [A-Z][a-z]+ (?:street|st|avenue|ave|road|rd|lane|ln|drive|dr)\b|latitude|longitude|track(?:ing)? (?:my|his|her) (?:phone|location))") + , + "health": re.compile(r"(?i)\b(diagnos(?:is|ed)|prescription|medication|therapist|therapy|symptoms?|doctor|clinic|hospital|surgery|chronic|depression|anxiety|cancer|diabetes|pregnan(?:t|cy)|mental health|blood (?:pressure|test))\b"), + "finance": re.compile(r"(?i)\b(salary|income|mortgage|bank account|credit card|iban|routing number|debt|loan|tax return|net worth|investment|401k|savings account|bankrupt(?:cy)?)\b"), + "legal": re.compile(r"(?i)\b(lawsuit|attorney|lawyer|court (?:date|case|order)|subpoena|arrest(?:ed)?|custody|divorce filing|criminal (?:record|charge)|restraining order|litigation)\b"), + "relationships": re.compile(r"(?i)\b(my (?:wife|husband|partner|girlfriend|boyfriend|ex)|breakup|broke up|divorce|affair|dating|marriage counsel(?:l)?ing)\b"), +} + + +def detect_topics(text: str) -> list[str]: + """Sensitive topics a piece of text looks like it touches, most restrictive first. Heuristic and uncertain by design.""" + found = [topic for topic, pattern in TOPIC_PATTERNS.items() if pattern.search(text or "")] + return sorted(found, key=lambda topic: (-SENSITIVITY_RANK[rules.PRIVACY_TOPICS[topic]["sensitivity"]], topic)) + + +def topic_sensitivity(topics: list[str]) -> str: + """Highest sensitivity among detected topics; ``personal`` when nothing matched.""" + ranked = sorted((rules.PRIVACY_TOPICS[t]["sensitivity"] for t in topics), key=SENSITIVITY_RANK.get, reverse=True) + return ranked[0] if ranked else "personal" + + +def _parse(stamp: str) -> datetime: + return datetime.fromisoformat(stamp.replace("Z", "+00:00")) + + +def _iso(when: datetime) -> str: + return when.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def load_topics(store: TenantStore) -> dict[str, Any]: + """The ``conversation_topics`` document: conversation id -> scoping state.""" + if store.exists(FAMILY, TOPICS_DOC): + return store.get(FAMILY, TOPICS_DOC) + return {"id": TOPICS_DOC, "schema": TOPICS_SCHEMA, "items": {}} + + +def _save_topics(store: TenantStore, doc: dict[str, Any]) -> None: + store.put(FAMILY, doc) + + +def conversation_state(store: TenantStore, conversation_id: str) -> dict[str, Any]: + """Scoping row for one conversation (empty when nothing was ever detected).""" + return load_topics(store)["items"].get(conversation_id, {}) + + +def mark_topic(store: TenantStore, conversation_id: str, topic: str, now: str | None = None) -> dict[str, Any]: + """Scope ``topic`` to ``conversation_id``; the first sighting fixes ``decay_at``.""" + now = now or now_iso() + with store.lock(FAMILY): + doc = load_topics(store) + row = doc["items"].setdefault(conversation_id, {"topics": [], "first_seen": now, "memory_disabled": False, "forgotten": False}) + if topic not in row["topics"]: + row["topics"].append(topic) + decay = _parse(row["first_seen"]) + timedelta(days=min(rules.PRIVACY_TOPICS[t]["decay_days"] for t in row["topics"])) + row["decay_at"] = _iso(decay) + _save_topics(store, doc) + return row + + +def set_flag(store: TenantStore, conversation_id: str, flag: str, value: bool) -> None: + """Set ``memory_disabled`` or ``forgotten`` for a conversation.""" + with store.lock(FAMILY): + doc = load_topics(store) + row = doc["items"].setdefault(conversation_id, {"topics": [], "first_seen": now_iso(), "memory_disabled": False, "forgotten": False}) + row[flag] = value + _save_topics(store, doc) + + +def memory_write_allowed(store: TenantStore, conversation_id: str | None, entry: dict[str, Any]) -> tuple[bool, str]: + """Gate every memory write (SO-21, SO-23, SO-28). Returns (allowed, reason).""" + from hux import events + + if conversation_id: + if events.is_private(store, conversation_id): + return False, "private_mode" + state = conversation_state(store, conversation_id) + if state.get("forgotten"): + return False, "conversation_forgotten" + if state.get("memory_disabled"): + return False, "memory_disabled" + if entry.get("sensitivity") == "restricted": + return False, "restricted" + topic = entry.get("topic", "general") + if topic in rules.PRIVACY_TOPICS and rules.PRIVACY_TOPICS[topic]["memory_write"] == "deny": + return False, f"topic_{topic}" + return True, "allowed" + + +def blocked_conversations(store: TenantStore) -> set[str]: + """Conversations that contribute nothing to retrieval: forgotten or memory disabled (SO-23).""" + return {cid for cid, row in load_topics(store)["items"].items() if row.get("forgotten") or row.get("memory_disabled")} + + +def latest_audit(store: TenantStore) -> dict[str, Any] | None: + """Newest retention audit record, or None when the job never ran.""" + rows = sorted(store.scan(RETENTION_FAMILY), key=lambda row: row["ran_at"]) + return rows[-1] if rows else None + + +def audit_stale(store: TenantStore, now: datetime | None = None) -> bool: + """SO-27: no retention audit in the last 48 hours.""" + last = latest_audit(store) + now = now or datetime.now(timezone.utc) + return last is None or now - _parse(last["ran_at"]) > timedelta(hours=AUDIT_STALE_HOURS) + + +# -- retention ---------------------------------------------------------------- + +def run_retention(store: TenantStore, now: datetime | None = None, identity: Identity | None = None) -> dict[str, Any]: + """Expire memory, decay topic context, purge forgotten content, report; write and return ``hux.retention_audit.v1``.""" + from hux import events, memory + + now = now or datetime.now(timezone.utc) + expired = memory.expire_due(store, now) + decayed = 0 + with store.lock(FAMILY): + doc = load_topics(store) + for conversation_id, row in doc["items"].items(): + if row.get("decayed") or not row.get("topics") or _parse(row.get("decay_at", "9999-01-01T00:00:00Z")) > now: + continue + notice = rules.privacy_policy()["topics"] + text = next(t["notice"] for t in notice if t["topic"] == row["topics"][0]) + decayed += events.rewrite_full(store, conversation_id, text, "topic context decayed", lambda r: r.get("sensitivity") in {"sensitive", "restricted"}) + row["decayed"] = True + _save_topics(store, doc) + purged = memory.purge_forgotten(store) + record = { + "schema": "hux.retention_audit.v1", + "id": new_id("aud"), + "ran_at": _iso(now), + "policy_version": rules.privacy_policy()["version"], + "results": [ + {"action": "expire_memory", "count": expired}, + {"action": "decay_topic_context", "count": decayed}, + {"action": "purge_forgotten_content", "count": purged}, + {"action": "report", "count": 1}, + ], + } + problems = contracts.validate_record(record, SCHEMAS) + if problems: + raise Invalid("retention audit failed contract validation", problems) + return _strip(store.put(RETENTION_FAMILY, record)) + + +def forget_conversation(store: TenantStore, identity: Identity, conversation_id: str) -> dict[str, Any]: + """Mark a conversation forgotten: its memory entries move to forgotten, its events go to full redaction.""" + from hux import events, memory + + set_flag(store, conversation_id, "forgotten", True) + forgotten_ids = memory.forget_from_conversation(store, identity, conversation_id) + redacted = events.rewrite_full(store, conversation_id, "[forgotten conversation]", "conversation forgotten") + counts = {"memory_forgotten": len(forgotten_ids), "events_redacted": redacted} + store.append(FAMILY, "forgotten", {"conv_id": conversation_id, "requested_at": now_iso(), "purged_at": "", "counts": counts}) + return {"conversation_id": conversation_id, "forgotten": True, **counts} + + +# -- routes --------------------------------------------------------------------- + +def _strip(record: dict[str, Any]) -> dict[str, Any]: + return {key: value for key, value in record.items() if key != "revision"} + + +def get_policy(request: Request) -> Response: + """``GET /hux/v1/privacy/policy``: the frozen policy; ``HUX-Audit-Stale`` says whether retention ran lately (SO-27).""" + request.audit("privacy.policy", "policy") + stale = audit_stale(request.store) + return Response(200, rules.privacy_policy(), {"HUX-Audit-Stale": "true" if stale else "false"}) + + +def post_notice(request: Request) -> Response: + """``POST /hux/v1/privacy/notices``: record a just-in-time notice as shown and scope the topic to the conversation.""" + from hux import events + + body = request.body if isinstance(request.body, dict) else {} + topic, conversation_id = body.get("topic"), body.get("conversation_id") + if topic not in rules.PRIVACY_TOPICS or not isinstance(conversation_id, str): + raise Invalid("topic and conversation_id are required") + controls = [c for c in body.get("controls", list(CONTROLS)) if c in CONTROLS] or list(CONTROLS) + text = next(t["notice"] for t in rules.privacy_policy()["topics"] if t["topic"] == topic) + notice = {"schema": "hux.privacy_notice.v1", "topic": topic, "conversation_id": conversation_id, "text": text, "controls": sorted(set(controls), key=CONTROLS.index), "shown_at": now_iso()} + problems = contracts.validate_record(notice, SCHEMAS) + if problems: + raise Invalid("notice failed contract validation", problems) + request.store.append(FAMILY, "notices", notice) + mark_topic(request.store, conversation_id, topic) + chosen = body.get("chosen") + if chosen == "disable_memory_here": + set_flag(request.store, conversation_id, "memory_disabled", True) + elif chosen == "forget_this_conversation": + forget_conversation(request.store, request.identity, conversation_id) + events.emit(request.store, request.identity, conversation_id, "privacy.notice", text, notice, sensitivity=rules.PRIVACY_TOPICS[topic]["sensitivity"]) + request.audit("privacy.notice", conversation_id) + return Response(201, notice) + + +def post_forget(request: Request) -> Response: + """``POST /hux/v1/conversations/{id}/forget``: user-initiated forget; unknown conversations are 404 (SO-18).""" + from hux import events + + conversation_id = request.params["id"] + if not (events.conversation_known(request.store, conversation_id) or conversation_id in load_topics(request.store)["items"]): + raise NotFound("conversation not found") + result = forget_conversation(request.store, request.identity, conversation_id) + request.audit("privacy.forget", conversation_id) + return Response(200, result) + + +def get_audit(request: Request) -> Response: + """``GET /hux/v1/privacy/audit``: retention audit records, newest first.""" + rows = sorted(request.store.scan(RETENTION_FAMILY), key=lambda row: row["ran_at"], reverse=True) + request.audit("privacy.audit", "retention") + return page([_strip(row) for row in rows[:PAGE_LIMIT]], None) + + +PAGE_LIMIT = 200 + + +def register(router: Router) -> None: + """Attach HUX-10 routes.""" + router.add("GET", "/hux/v1/privacy/policy", "HUX-10", "privacy.policy", get_policy) + router.add("POST", "/hux/v1/privacy/notices", "HUX-10", "privacy.notice", post_notice) + router.add("POST", "/hux/v1/conversations/{id}/forget", "HUX-10", "privacy.forget", post_forget) + router.add("GET", "/hux/v1/privacy/audit", "HUX-10", "privacy.audit", get_audit) diff --git a/dockerfiles/hermes-hux-foundation/hux/redaction.py b/dockerfiles/hermes-hux-foundation/hux/redaction.py new file mode 100644 index 00000000..b095c089 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/redaction.py @@ -0,0 +1,225 @@ +"""Redaction pipeline shared by every HUX family (SO-11..SO-14, SO-19). + +Order matters and is fixed here: drop detail keys outside the per-kind +allowlist, scrub secret patterns from every remaining string, cap sizes, +then derive ``redaction.level``. Serve-time redaction (``redact_record``) is +applied again on the way out so a reader that ignores the level still never +sees more than the surface is allowed to. +""" + +from __future__ import annotations + +import json +import os +import re +from typing import Any + +DETAIL_CAP_BYTES = 32 * 1024 +LINE_CAP_BYTES = 64 * 1024 +SUMMARY_MAX = 280 +URI_SCHEMES = ("https://", "hux://", "artifact://") +WORKSPACE_PREFIX = "/opt/data/workspace" +HASH_KEYS = frozenset({"hash", "argument_hash"}) +HUMAN_SURFACES = frozenset({"telegram", "voice"}) +FULL_KEEP = frozenset({"schema", "id", "seq", "ts", "conversation_id", "kind", "provenance", "sensitivity", "redaction", "turn", "identity"}) + +# Each pattern is (class, regex). Classes end up in the placeholder so a +# reviewer can tell what kind of thing was removed without seeing it. +SECRET_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = ( + ("private_key", re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]*?(?:-----END [A-Z ]*PRIVATE KEY-----|$)")), + ("bearer", re.compile(r"(?i)bearer\s+[A-Za-z0-9._-]{16,}")), + ("jwt", re.compile(r"eyJ[A-Za-z0-9_-]{20,}\.[A-Za-z0-9_-]{10,}(?:\.[A-Za-z0-9_-]+)?")), + ("openai_key", re.compile(r"sk-[A-Za-z0-9_-]{20,}")), + ("github_token", re.compile(r"(?:ghp|gho|ghu|ghs|ghr)_[A-Za-z0-9]{20,}")), + ("gitlab_token", re.compile(r"glpat-[A-Za-z0-9_-]{20,}")), + ("aws_key", re.compile(r"AKIA[0-9A-Z]{16}")), + ("slack_token", re.compile(r"xox[abp]-[A-Za-z0-9-]{10,}")), + ("vault_token", re.compile(r"\b(?:hvs|hvb|hvr)\.[A-Za-z0-9_-]{20,}")), + ("password", re.compile(r"(?i)(?:password|passwd|pwd|secret|token)\s*[=:]\s*\S+")), + ("hex", re.compile(r"\b[0-9a-fA-F]{32,}\b")), + ("base64", re.compile(r"\b(?=[A-Za-z0-9+/]*[0-9])(?=[A-Za-z0-9+/]*[a-z])(?=[A-Za-z0-9+/]*[A-Z])[A-Za-z0-9+/]{40,}={0,2}")), +) + +# Detail keys each event kind may persist. Everything else is dropped before +# a byte reaches disk, never filtered on read only. +DETAIL_ALLOWLIST: dict[str, frozenset[str]] = { + "message.user": frozenset({"message_id", "chars", "has_attachments"}), + "message.assistant": frozenset({"message_id", "chars", "has_attachments"}), + "decision.route": frozenset({"requested", "resolved_target", "provider", "effort", "reason"}), + "decision.plan": frozenset({"steps"}), + "tool.call": frozenset({"tool", "capability", "argument_names", "argument_hash", "argument_bytes", "target_path"}), + "tool.result": frozenset({"tool", "ok", "duration_ms", "bytes", "exit_code"}), + "approval.requested": frozenset({"approval_id", "capability", "choice", "external"}), + "approval.resolved": frozenset({"approval_id", "capability", "choice", "external"}), + "side_effect.blocked": frozenset({"approval_id", "capability", "choice", "external"}), + "side_effect.released": frozenset({"approval_id", "capability", "choice", "external"}), + "memory.proposed": frozenset({"memory_id", "kind", "sensitivity", "topic"}), + "memory.committed": frozenset({"memory_id", "kind", "sensitivity", "topic"}), + "memory.forgotten": frozenset({"memory_id", "kind", "sensitivity", "topic"}), + "memory.suppressed": frozenset({"memory_id", "kind", "sensitivity", "topic"}), + "memory.retrieval_removed": frozenset({"memory_id", "kind", "sensitivity", "topic"}), + "artifact.created": frozenset({"artifact_id", "version", "type", "bytes", "hash"}), + "artifact.version": frozenset({"artifact_id", "version", "type", "bytes", "hash"}), + "artifact.promoted": frozenset({"artifact_id", "version", "type", "bytes", "hash"}), + "run.started": frozenset({"run_id", "outcome", "receipt_id", "spent", "limits", "child_run_id"}), + "run.cancelled": frozenset({"run_id", "outcome", "receipt_id", "spent", "limits", "child_run_id"}), + "run.completed": frozenset({"run_id", "outcome", "receipt_id", "spent", "limits", "child_run_id"}), + "run.failed": frozenset({"run_id", "outcome", "receipt_id", "spent", "limits", "child_run_id"}), + "delegation.started": frozenset({"run_id", "outcome", "receipt_id", "spent", "limits", "child_run_id"}), + "delegation.completed": frozenset({"run_id", "outcome", "receipt_id", "spent", "limits", "child_run_id"}), + "delegation.failed": frozenset({"run_id", "outcome", "receipt_id", "spent", "limits", "child_run_id"}), + "budget.exhausted": frozenset({"run_id", "outcome", "receipt_id", "spent", "limits", "child_run_id"}), + "privacy.notice": frozenset({"schema", "topic", "conversation_id", "text", "controls", "shown_at"}), + "citation.attached": frozenset({"citation_id", "message_id", "passage_ids", "verdict"}), + "mode.changed": frozenset({"mode", "previous_mode", "route_id"}), + "suggestion.shown": frozenset({"suggestion_id", "action"}), + "suggestion.dismissed": frozenset({"suggestion_id", "action"}), + "release.transition": frozenset({"release_id", "state", "previous_state"}), +} + + +def canaries(environ: dict[str, str] | None = None) -> list[str]: + """Literal secrets the pod was started with (SO-07); every one is scrubbed wherever it appears.""" + environ = os.environ if environ is None else environ + values = [environ.get(name, "") for name in ("HUX_RELAY_KEY", "HUX_WORKER_KEY")] + path = environ.get("HUX_CANARY_FILE", "") + if path and os.path.exists(path): + with open(path, encoding="utf-8", errors="replace") as handle: + for line in handle: + _, _, value = line.strip().partition("=") + values.append(value.strip().strip("'\"")) + return [v for v in values if len(v) >= 8] + + +def scrub_text(text: str, environ: dict[str, str] | None = None) -> tuple[str, list[str]]: + """Replace secret-looking substrings with ``[redacted:]``; return the text and the classes hit.""" + hits: list[str] = [] + for canary in canaries(environ): + if canary in text: + text = text.replace(canary, "[redacted:canary]") + hits.append("canary") + for name, pattern in SECRET_PATTERNS: + text, count = pattern.subn(f"[redacted:{name}]", text) + if count: + hits.append(name) + return text, hits + + +def scrub_value(value: Any, hits: list[str], environ: dict[str, str] | None = None) -> Any: + """Scrub every string inside a JSON-like value, collecting classes into ``hits``. + + Content hashes (``hash``, ``argument_hash``) are digests by construction and + are kept verbatim so the hex rule does not eat them. + """ + if isinstance(value, str): + text, found = scrub_text(value, environ) + hits.extend(found) + return text + if isinstance(value, list): + return [scrub_value(item, hits, environ) for item in value] + if isinstance(value, dict): + return { + str(key): item if key in HASH_KEYS and isinstance(item, str) else scrub_value(item, hits, environ) + for key, item in value.items() + } + return value + + +def filter_detail(kind: str, detail: dict[str, Any] | None) -> dict[str, Any]: + """Keep only the allowlisted keys for ``kind`` (SO-11) and apply the kind-specific shape rules.""" + if not detail: + return {} + allowed = DETAIL_ALLOWLIST.get(kind, frozenset()) + kept = {key: value for key, value in detail.items() if key in allowed} + if kind == "decision.plan" and "steps" in kept: + steps = kept["steps"] if isinstance(kept["steps"], list) else [] + kept["steps"] = [str(step)[:200] for step in steps[:20]] + if kind == "tool.call": + # Raw arguments never persist: only their names, a hash and a byte count. + path = kept.get("target_path") + if not (isinstance(path, str) and path.startswith(WORKSPACE_PREFIX + "/")): + kept.pop("target_path", None) + if "argument_names" in kept: + names = kept["argument_names"] if isinstance(kept["argument_names"], list) else [] + kept["argument_names"] = sorted({str(name)[:80] for name in names})[:64] + return kept + + +def cap_detail(detail: dict[str, Any]) -> tuple[dict[str, Any], bool]: + """Replace detail over 32 KiB with a truncation marker (SO-13).""" + size = len(json.dumps(detail, sort_keys=True, separators=(",", ":")).encode()) + if size > DETAIL_CAP_BYTES: + return {"truncated": True, "bytes": size}, True + return detail, False + + +def filter_evidence(evidence: list[dict[str, Any]] | None) -> list[dict[str, Any]]: + """Keep well-formed refs, drop URIs outside the allowlisted schemes (SO-19), cap at 64.""" + kept: list[dict[str, Any]] = [] + for ref in evidence or []: + if not isinstance(ref, dict) or not isinstance(ref.get("kind"), str) or not isinstance(ref.get("id"), str): + continue + clean = {"kind": ref["kind"], "id": ref["id"][:200]} + uri = ref.get("uri") + if isinstance(uri, str) and uri.startswith(URI_SCHEMES): + clean["uri"] = uri[:2000] + if isinstance(ref.get("hash"), str): + clean["hash"] = ref["hash"] + kept.append(clean) + return kept[:64] + + +def derive_level(sensitivity: str, hits: list[str], truncated: bool) -> dict[str, str]: + """Redaction block for a new event from its sensitivity and what the pipeline did.""" + reasons: list[str] = [] + level = "none" + if sensitivity == "sensitive": + level, reasons = "partial", ["sensitive topic"] + if hits: + level = "partial" if level == "none" else level + reasons.append("scrubbed " + ",".join(sorted(set(hits)))[:120]) + if truncated: + level = "partial" if level == "none" else level + reasons.append("detail truncated") + if sensitivity == "restricted": + level, reasons = "full", ["restricted content"] + block = {"level": level} + if reasons: + block["reason"] = "; ".join(reasons)[:200] + return block + + +def redact_record(record: dict[str, Any], surface: str) -> dict[str, Any]: + """Serve-time redaction (SO-14) for any record that carries ``redaction``; strips ``_meta`` (SO-51). + + ``none`` is served as stored except that telegram and voice are lifted to + ``partial``; ``partial`` drops ``detail``; ``full`` keeps only the + contract-required envelope with the summary replaced by a placeholder. + """ + served = {key: value for key, value in record.items() if key != "_meta"} + block = served.get("redaction") + if not isinstance(block, dict): + return served + level = block.get("level", "none") + if level == "none" and surface in HUMAN_SURFACES: + level = "partial" + served["redaction"] = {"level": "partial", "reason": f"{surface} surface never receives unredacted detail"} + if level == "partial": + served.pop("detail", None) + if level == "full": + served = {key: value for key, value in served.items() if key in FULL_KEEP} + if "summary" in record: + served["summary"] = "[redacted]" + if "content" in record: + served["content"] = "" + return served + + +def full_redaction(record: dict[str, Any], summary: str, reason: str) -> dict[str, Any]: + """Rewrite a stored event to ``redaction.level: full`` in place (forget and topic decay).""" + kept = {key: value for key, value in record.items() if key in FULL_KEEP} + kept["summary"] = summary[:SUMMARY_MAX] or "[redacted]" + kept["redaction"] = {"level": "full", "reason": reason[:200]} + if "idempotency_key" in record: + kept["idempotency_key"] = record["idempotency_key"] + return kept diff --git a/testing/tests/test_hermes_hux_contract_events.py b/testing/tests/test_hermes_hux_contract_events.py new file mode 100644 index 00000000..5ca8646c --- /dev/null +++ b/testing/tests/test_hermes_hux_contract_events.py @@ -0,0 +1,366 @@ +"""HUX-01 activity events: ordering, idempotency, replay, reconnect, redaction, cancellation. + +Security obligations exercised: SO-10..SO-19 (server-set provenance, detail +allowlist, secret scrub, size caps, serve-time redaction, server-assigned seq, +idempotent replay after ownership, bounded pages and streams, 404 for foreign +conversations, evidence URI allowlist) and SO-28 (private mode writes nothing). +""" + +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, errors, events, identity, redaction, 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"} +OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"} +ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) +CONV = "conv_0001abcd" + + +def ident(**overrides) -> identity.Identity: + base = {"tenant_slot": "slot-3", "subject": "usr_0123456789abcdef", "surface": "chat", "trust": "router"} + return identity.Identity(**{**base, **overrides}) + + +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, response + + +def tenant(tmp_path, who=None) -> store.TenantStore: + return store.TenantStore(tmp_path, who or ident()) + + +def valid(record) -> None: + assert contracts.validate_record(record, SCHEMAS) == [], record + + +# --- emit: ordering and idempotency (SO-15, SO-16) --------------------------------- + +def test_emit_assigns_monotonic_seq_and_validates(tmp_path): + s = tenant(tmp_path) + first = events.emit(s, ident(), CONV, "message.user", "hello", {"message_id": "m1", "chars": 5, "junk": "x"}, turn=1) + second = events.emit(s, ident(), CONV, "message.assistant", "hi", run_id="run_9f", correlation_id="c1") + assert (first["seq"], second["seq"]) == (1, 2) + assert first["id"] != second["id"] and first["id"].startswith("evt_") + assert first["detail"] == {"message_id": "m1", "chars": 5} + assert second["run_id"] == "run_9f" and second["provenance"]["run_id"] == "run_9f" + assert first["provenance"]["actor"] == {"type": "user", "id": "usr_0123456789abcdef"} + assert second["provenance"]["actor"]["type"] == "assistant" + assert first["identity"] == ident().record() + for record in (first, second): + valid(record) + checkpoint = s.get(events.SEQ_FAMILY, f"seq_{CONV}") + assert checkpoint["next_seq"] == 3 and checkpoint["last_event_id"] == second["id"] + other = events.emit(s, ident(), "conv_0002abcd", "run.started", "another conversation") + assert other["seq"] == 1 + + +def test_emit_is_idempotent_on_key(tmp_path): + s = tenant(tmp_path) + one = events.emit(s, ident(), CONV, "run.started", "start", idempotency_key="run_9f:start:1") + again = events.emit(s, ident(), CONV, "run.started", "start", idempotency_key="run_9f:start:1") + assert again == one + assert len(s.read(events.FAMILY, CONV)) == 1 + _, replayed = events.emit_with_status(s, ident(), CONV, "run.started", "start", idempotency_key="run_9f:start:1") + assert replayed is True + + +def test_emit_rejects_unknown_kind_and_sensitivity(tmp_path): + s = tenant(tmp_path) + with pytest.raises(errors.Invalid): + events.emit(s, ident(), CONV, "not.a.kind", "x") + with pytest.raises(errors.Invalid): + events.emit(s, ident(), CONV, "run.started", "x", sensitivity="secret") + with pytest.raises(errors.Invalid): + events.emit(s, ident(), "not a conversation id", "run.started", "x") + + +def test_emit_validates_against_contract_before_writing(tmp_path, monkeypatch): + s = tenant(tmp_path) + monkeypatch.setattr(events, "new_id", lambda prefix: "bad id") + with pytest.raises(errors.Invalid): + events.emit(s, ident(), CONV, "run.started", "x") + assert s.read(events.FAMILY, CONV) == [] + + +def test_worker_emits_as_system_actor(tmp_path): + s = tenant(tmp_path) + record = events.emit(s, ident(surface="worker", trust="worker"), CONV, "message.user", "x") + assert record["provenance"]["actor"] == {"type": "system", "id": "hux-worker"} + + +def test_private_mode_writes_nothing(tmp_path): + s = tenant(tmp_path) + s.put("conversations", {"id": "conv_priv0001", "mode": "private"}) + assert events.emit(s, ident(), "conv_priv0001", "message.user", "secret chat") is None + assert s.read(events.FAMILY, "conv_priv0001") == [] + router = router_for(tmp_path) + status, body, _ = call(router, "POST", "/hux/v1/conversations/conv_priv0001/events", body={"kind": "message.user", "summary": "x"}) + assert status == 204 and body is None + + +# --- redaction pipeline (SO-11, SO-12, SO-13, SO-19) ------------------------------ + +def test_tool_call_detail_keeps_only_hash_and_names(tmp_path): + s = tenant(tmp_path) + detail = { + "tool": "shell", "capability": "shell", "arguments": {"cmd": "curl -H 'Authorization: Bearer abcdefghijklmnopqrstuv'"}, + "argument_names": ["cmd", "cwd", "cmd"], "argument_hash": "a" * 64, "argument_bytes": 61, + "target_path": "/opt/data/.env", "stdout": "leak", + } + record = events.emit(s, ident(), CONV, "tool.call", "ran shell", detail) + assert record["detail"] == {"tool": "shell", "capability": "shell", "argument_names": ["cmd", "cwd"], "argument_hash": "a" * 64, "argument_bytes": 61} + assert "arguments" not in json.dumps(record) and "leak" not in json.dumps(record) + kept = events.emit(s, ident(), CONV, "tool.call", "wrote", {"tool": "write", "target_path": "/opt/data/workspace/notes.md"}) + assert kept["detail"]["target_path"] == "/opt/data/workspace/notes.md" + plan = events.emit(s, ident(), CONV, "decision.plan", "plan", {"steps": ["s" * 300] * 25}) + assert len(plan["detail"]["steps"]) == 20 and len(plan["detail"]["steps"][0]) == 200 + assert events.emit(s, ident(), CONV, "decision.plan", "plan", {"steps": "nope"})["detail"] == {"steps": []} + assert events.emit(s, ident(), CONV, "tool.call", "x", {"tool": "t", "argument_names": "nope"})["detail"]["argument_names"] == [] + + +@pytest.mark.parametrize("secret,klass", [ + ("Authorization: Bearer abcdefghijklmnopqrstuvwxyz0123", "bearer"), + ("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0In0.abcdef", "jwt"), + ("sk-abcdefghijklmnopqrstuvwxyz1234", "openai_key"), + ("ghp_abcdefghijklmnopqrstuvwxyz1234", "github_token"), + ("glpat-abcdefghijklmnopqrstuv", "gitlab_token"), + ("AKIAABCDEFGHIJKLMNOP", "aws_key"), + ("xoxb-1234567890-abcdef", "slack_token"), + ("hvs.CAESIJabcdefghijklmnopqrstuvwxyz", "vault_token"), + ("password=hunter2hunter2", "password"), + ("0123456789abcdef0123456789abcdef", "hex"), + ("-----BEGIN RSA PRIVATE KEY-----\nMIIE\n-----END RSA PRIVATE KEY-----", "private_key"), +]) +def test_secret_patterns_are_scrubbed(tmp_path, secret, klass): + s = tenant(tmp_path) + record = events.emit(s, ident(), CONV, "decision.route", f"used {secret} today", {"reason": secret}) + text = json.dumps(record) + assert secret not in text and f"[redacted:{klass}]" in text + assert record["redaction"]["level"] == "partial" and klass in record["redaction"]["reason"] + + +def test_canaries_from_environment_and_file_are_scrubbed(tmp_path, monkeypatch): + canary_file = tmp_path / "env" + canary_file.write_text("OPENAI_KEY='filecanaryvalue'\nSHORT=x\n") + monkeypatch.setenv("HUX_RELAY_KEY", "relaykeycanary") + monkeypatch.setenv("HUX_CANARY_FILE", str(canary_file)) + text, hits = redaction.scrub_text("relay relaykeycanary file filecanaryvalue") + assert text == "relay [redacted:canary] file [redacted:canary]" and hits == ["canary", "canary"] + assert redaction.canaries({}) == [] + + +def test_hashes_survive_the_hex_rule(tmp_path): + s = tenant(tmp_path) + record = events.emit(s, ident(), CONV, "artifact.version", "v2", {"artifact_id": "art_0001aaaa", "hash": "sha256:" + "b" * 64}, [{"kind": "artifact_version", "id": "art_0001aaaa@2", "hash": "sha256:" + "b" * 64}]) + assert record["detail"]["hash"] == "sha256:" + "b" * 64 and record["evidence"][0]["hash"] == "sha256:" + "b" * 64 + assert record["redaction"] == {"level": "none"} + + +def test_oversized_detail_is_truncated_and_line_cap_enforced(tmp_path): + s = tenant(tmp_path) + record = events.emit(s, ident(), CONV, "decision.route", "big", {"reason": "r" * 40000}) + assert record["detail"]["truncated"] is True and record["detail"]["bytes"] > redaction.DETAIL_CAP_BYTES + assert record["redaction"]["level"] == "partial" and "truncated" in record["redaction"]["reason"] + huge = [{"kind": "url", "id": "u", "uri": "https://x/" + "z" * 1990} for _ in range(64)] + with pytest.raises(errors.TooLarge): + events.emit(s, ident(), CONV, "decision.route", "too big", evidence=huge) + + +def test_evidence_uris_outside_allowlist_are_dropped(tmp_path): + s = tenant(tmp_path) + evidence = [ + {"kind": "file", "id": "env", "uri": "file:///opt/data/.env"}, + {"kind": "url", "id": "doc", "uri": "https://example.test/doc"}, + {"kind": "artifact_version", "id": "art_0001aaaa@1", "uri": "artifact://art_0001aaaa/1"}, + "garbage", {"kind": 5, "id": "x"}, + ] + record = events.emit(s, ident(), CONV, "tool.result", "read", evidence=evidence) + assert [ref.get("uri") for ref in record["evidence"]] == [None, "https://example.test/doc", "artifact://art_0001aaaa/1"] + assert len(redaction.filter_evidence([{"kind": "run", "id": "r"}] * 100)) == 64 + + +def test_redaction_level_derivation(): + assert redaction.derive_level("personal", [], False) == {"level": "none"} + assert redaction.derive_level("sensitive", [], False)["level"] == "partial" + assert redaction.derive_level("sensitive", ["hex"], True)["level"] == "partial" + assert redaction.derive_level("restricted", ["hex"], True) == {"level": "full", "reason": "restricted content"} + assert redaction.derive_level("public", [], True)["reason"] == "detail truncated" + + +def test_serve_time_redaction_by_surface(tmp_path): + s = tenant(tmp_path) + record = events.emit(s, ident(), CONV, "decision.route", "routed", {"requested": "fast"}) + assert redaction.redact_record({**record, "_meta": {"x": 1}}, "chat") == record + for surface in ("telegram", "voice"): + served = redaction.redact_record(record, surface) + assert "detail" not in served and served["redaction"]["level"] == "partial" + valid(served) + partial = events.emit(s, ident(), CONV, "decision.route", "routed", {"requested": "fast"}, sensitivity="sensitive") + assert "detail" not in redaction.redact_record(partial, "chat") + full = events.emit(s, ident(), CONV, "decision.route", "routed", {"requested": "fast"}, ["x"], sensitivity="restricted", run_id="r") + served = redaction.redact_record(full, "chat") + assert served["summary"] == "[redacted]" and "detail" not in served and "evidence" not in served and "run_id" not in served + valid(served) + assert redaction.redact_record({"schema": "x", "content": "c"}, "chat") == {"schema": "x", "content": "c"} + memory_like = {"schema": "hux.memory.v1", "content": "secret", "redaction": {"level": "full"}} + assert redaction.redact_record(memory_like, "chat")["content"] == "" + + +def test_full_rewrite_keeps_timeline_contiguous(tmp_path): + s = tenant(tmp_path) + events.emit(s, ident(), CONV, "message.user", "one", {"message_id": "m1"}, idempotency_key="m1:00000001") + events.emit(s, ident(), CONV, "message.user", "two", sensitivity="sensitive") + assert events.rewrite_full(s, CONV, "", "forgotten") == 2 + rows = s.read(events.FAMILY, CONV) + assert [r["seq"] for r in rows] == [1, 2] and all(r["redaction"]["level"] == "full" for r in rows) + assert rows[0]["summary"] == "[redacted]" and rows[0]["idempotency_key"] == "m1:00000001" and "detail" not in rows[0] + for row in rows: + valid(row) + assert events.rewrite_full(s, CONV, "", "again") == 0 + + +def test_memory_reference_redaction_spans_conversations(tmp_path): + s = tenant(tmp_path) + events.emit(s, ident(), CONV, "memory.committed", "kept", {"memory_id": "mem_0001aaaa"}) + events.emit(s, ident(), "conv_0002abcd", "message.user", "plain", evidence=[{"kind": "memory", "id": "mem_0001aaaa"}]) + events.emit(s, ident(), "conv_0002abcd", "message.user", "unrelated") + assert events.redact_memory_references(s, "mem_0001aaaa") == 2 + assert s.read(events.FAMILY, "conv_0002abcd")[1]["redaction"]["level"] == "none" + + +# --- routes ------------------------------------------------------------------------- + +def test_list_pages_with_after_seq_and_limit(tmp_path): + router = router_for(tmp_path) + s = tenant(tmp_path) + for n in range(5): + events.emit(s, ident(), CONV, "message.user", f"m{n}") + status, body, _ = call(router, "GET", f"/hux/v1/conversations/{CONV}/events?limit=2") + assert status == 200 and [e["seq"] for e in body["items"]] == [1, 2] and body["next"] == 2 + status, body, _ = call(router, "GET", f"/hux/v1/conversations/{CONV}/events?after_seq=2&limit=2") + assert [e["seq"] for e in body["items"]] == [3, 4] and body["next"] == 4 + status, body, _ = call(router, "GET", f"/hux/v1/conversations/{CONV}/events?after_seq=4&limit=2") + assert [e["seq"] for e in body["items"]] == [5] and body["next"] is None + for item in body["items"]: + valid(item) + status, body, _ = call(router, "GET", f"/hux/v1/conversations/{CONV}/events?limit=5000") + assert len(body["items"]) == 5 + assert events.PAGE_MAX == 200 + assert call(router, "GET", f"/hux/v1/conversations/{CONV}/events?after_seq=abc")[0] == 400 + assert call(router, "GET", f"/hux/v1/conversations/{CONV}/events?limit=0&after_seq=-5")[1]["items"][0]["seq"] == 1 + + +def test_foreign_conversation_is_404_not_403(tmp_path): + router = router_for(tmp_path) + events.emit(tenant(tmp_path), ident(), CONV, "message.user", "mine") + status, body, _ = call(router, "GET", f"/hux/v1/conversations/{CONV}/events", headers=OTHER) + assert (status, body["code"]) == (404, "not_found") + assert call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=0", headers=OTHER)[0] == 404 + assert call(router, "POST", f"/hux/v1/conversations/{CONV}/events", headers=OTHER, body={"kind": "run.started", "summary": "x"})[0] == 404 + assert call(router, "GET", "/hux/v1/conversations/conv_nope0000/events")[0] == 404 + rows = audit.recent(tenant(tmp_path, ident(subject="usr_fedcba9876543210"))) + assert {r["outcome"] for r in rows} == {"not_found"} + + +def test_post_event_server_assigns_and_replays(tmp_path): + router = router_for(tmp_path) + tenant(tmp_path).put("conversations", {"id": CONV, "mode": "fast"}) + body = {"kind": "run.cancelled", "summary": "stopped by user", "detail": {"run_id": "run_9f", "outcome": "cancelled", "receipt_id": "rcpt_0001aaaa", "raw": 1}, + "evidence": [{"kind": "approval", "id": "apr_0001aaaa"}, {"kind": "run", "id": "run_9f"}], "run_id": "run_9f", "turn": 3, + "identity": {"tenant_slot": "slot-9"}, "provenance": {"surface": "worker"}} + headers = {**HEADERS, "Idempotency-Key": "run_9f:cancel:1"} + status, first, _ = call(router, "POST", f"/hux/v1/conversations/{CONV}/events", headers, body) + assert status == 201 and first["seq"] == 1 and first["identity"]["tenant_slot"] == "slot-3" and first["provenance"]["surface"] == "chat" + assert first["provenance"]["actor"]["type"] == "user" and first["detail"] == {"run_id": "run_9f", "outcome": "cancelled", "receipt_id": "rcpt_0001aaaa"} + assert [e["kind"] for e in first["evidence"]] == ["approval", "run"] + valid(first) + status, again, response = call(router, "POST", f"/hux/v1/conversations/{CONV}/events", headers, body) + assert status == 200 and again == first and response.headers["HUX-Replayed"] == "true" + assert call(router, "POST", f"/hux/v1/conversations/{CONV}/events", HEADERS, {"kind": "run.started", "summary": "x", "seq": 9})[0] == 400 + assert call(router, "POST", f"/hux/v1/conversations/{CONV}/events", HEADERS, {"kind": "run.started", "summary": "x", "id": "evt_0000"})[0] == 400 + assert call(router, "POST", f"/hux/v1/conversations/{CONV}/events", HEADERS, {"summary": "x"})[0] == 400 + assert call(router, "POST", f"/hux/v1/conversations/{CONV}/events", HEADERS, [1])[0] == 400 + status, body, _ = call(router, "POST", f"/hux/v1/conversations/{CONV}/events", HEADERS, {"kind": "run.started", "summary": "x", "detail": "nope", "evidence": "nope", "turn": "3"}) + assert status == 201 and body["turn"] == 0 and "detail" not in body + reasons = [r["reason"] for r in audit.recent(tenant(tmp_path)) if r["action"] == "events.append" and r["outcome"] == "allow" and r.get("reason")] + assert reasons == ["replayed"] + + +def _sse_records(chunks): + out = [] + for chunk in chunks: + for line in chunk.decode().splitlines(): + if line.startswith("data: "): + out.append(json.loads(line[6:])) + return out + + +def test_stream_replays_then_polls_and_resumes_from_last_event_id(tmp_path): + router = router_for(tmp_path) + s = tenant(tmp_path) + for n in range(3): + events.emit(s, ident(), CONV, "message.user", f"m{n}") + status, _, response = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=0") + chunks = list(response.stream()) + assert status == 200 and chunks[0] == b"retry: 2000\n\n" + assert [r["seq"] for r in _sse_records(chunks)] == [1, 2, 3] + assert b"id: 3\nevent: message.user\n" in chunks[3] + status, _, response = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=1&poll_ms=0", {**HEADERS, "Last-Event-ID": "2"}) + generator = response.stream() + assert next(generator) == b"retry: 2000\n\n" + assert _sse_records([next(generator)])[0]["seq"] == 3 + assert next(generator) == b": keepalive\n\n" + events.emit(s, ident(), CONV, "message.assistant", "late") + assert _sse_records([next(generator)])[0]["seq"] == 4 + assert list(generator) == [] + status, _, response = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?after_seq=4&max_polls=0", {**HEADERS, "X-Hux-Surface": "voice"}) + assert _sse_records(list(response.stream())) == [] + + +def test_second_stream_closes_the_first(tmp_path): + router = router_for(tmp_path) + events.emit(tenant(tmp_path), ident(), CONV, "message.user", "m") + _, _, first = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=5&poll_ms=0") + first_gen = first.stream() + next(first_gen) + next(first_gen) + _, _, second = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=0") + list(second.stream()) + assert list(first_gen) == [] + + +def test_stream_redacts_for_voice_surface(tmp_path): + router = router_for(tmp_path) + events.emit(tenant(tmp_path), ident(), CONV, "decision.route", "r", {"requested": "fast"}) + _, _, response = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=0", {**HEADERS, "X-Hux-Surface": "voice"}) + records = _sse_records(list(response.stream())) + assert "detail" not in records[0] and records[0]["redaction"]["level"] == "partial" + + +def test_flag_off_hides_event_routes(tmp_path): + router = build_router(tmp_path, {"HUX_FLAGS": "hux.foundation"}) + assert call(router, "GET", f"/hux/v1/conversations/{CONV}/events")[1]["code"] == "flag_off" + + +def test_events_sources_stay_under_500_lines(): + for name in ("events.py", "redaction.py"): + assert len((FOUNDATION / "hux" / name).read_text().splitlines()) <= 500, name diff --git a/testing/tests/test_hermes_hux_memory_ledger.py b/testing/tests/test_hermes_hux_memory_ledger.py new file mode 100644 index 00000000..6dcaf4b6 --- /dev/null +++ b/testing/tests/test_hermes_hux_memory_ledger.py @@ -0,0 +1,313 @@ +"""HUX-02 memory ledger: proposals, consent transitions, edits, If-Match and no-store. + +Security obligations exercised: SO-10 (server-set owner and provenance), +SO-15 (server-assigned ids), SO-18 (foreign ids are 404), SO-21 (policy +violations suppressed with an event), SO-22 (no_store and forget write a +content-free line plus tombstone), SO-24 (forget re-redacts events), SO-25 +(supersedes of a forgotten id needs a user approval), SO-26 (export scope and +audit), SO-28 (private mode refuses writes), SO-35-style human-only decisions, +SO-44 (If-Match conflicts, unconditional writes audited). +""" + +from __future__ import annotations + +import json +import sys +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 audit, contracts, events, identity, memory, 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"} +OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"} +ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) +CONV = "conv_0001abcd" +USER = {"proposed_by": "user"} + + +def ident(**overrides) -> identity.Identity: + base = {"tenant_slot": "slot-3", "subject": "usr_0123456789abcdef", "surface": "chat", "trust": "router"} + return identity.Identity(**{**base, **overrides}) + + +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, response + + +def tenant(tmp_path, who=None) -> store.TenantStore: + return store.TenantStore(tmp_path, who or ident()) + + +def valid(record) -> None: + assert contracts.validate_record(record, SCHEMAS) == [], record + + +def propose(router, headers=HEADERS, **fields): + body = {"kind": "preference", "content": "Prefers terse answers with code first.", "reason": "asked twice", "conversation_id": CONV, **fields} + return call(router, "POST", "/hux/v1/memory", headers, body) + + +def event_kinds(tmp_path, conversation=CONV): + return [row["kind"] for row in tenant(tmp_path).read(events.FAMILY, conversation)] + + +# --- proposals ------------------------------------------------------------------ + +def test_user_proposal_with_personal_content_is_active_immediately(tmp_path): + router = router_for(tmp_path) + status, body, _ = propose(router, **USER, scope={"level": "project", "scope_id": "prj_0001aaaa"}, source={"kind": "message", "id": "msg-42"}, run_id="run_9f") + assert status == 201 and body["status"] == "active" and body["approval_mode"] == "automatic" and body["retrievable"] is True + assert body["owner"] == "usr_0123456789abcdef" and body["provenance"]["actor"]["type"] == "user" and body["provenance"]["run_id"] == "run_9f" + assert body["ttl"] == {"policy": "decay", "decay_days": 180} and body["topic"] == "general" and body["revision"] == 1 + assert [row["action"] for row in body["audit"]] == ["proposed", "approved"] + valid(body) + assert event_kinds(tmp_path) == ["memory.committed"] + ledger = tenant(tmp_path).read(memory.FAMILY, memory.LEDGER) + assert [row["id"] for row in ledger] == [body["id"]] + + +def test_assistant_proposal_is_suggest_only(tmp_path): + router = router_for(tmp_path) + status, body, _ = propose(router) + assert status == 201 and body["status"] == "proposed" and body["approval_mode"] == "ask" and body["retrievable"] is False + assert body["provenance"]["actor"] == {"type": "assistant", "id": "hermes"} + valid(body) + assert event_kinds(tmp_path) == ["memory.proposed"] + status, explicit, _ = propose(router, **USER, approval_mode="ask") + assert explicit["status"] == "proposed" + + +def test_body_supplied_identity_and_ids_are_rejected_or_ignored(tmp_path): + router = router_for(tmp_path) + assert propose(router, id="mem_evil0001")[0] == 400 + assert propose(router, revision=7)[0] == 400 + assert propose(router, status="active")[0] == 400 + assert propose(router, content="")[0] == 400 + assert propose(router, content="x" * 2001)[0] == 400 + assert propose(router, sensitivity="ultra")[0] == 400 + assert call(router, "POST", "/hux/v1/memory", HEADERS, [1])[0] == 400 + status, body, _ = propose(router, **USER, owner="usr_fedcba9876543210", kind="weird", topic="not-a-topic", ttl="never", scope="bad", source="bad") + assert status == 201 and body["owner"] == "usr_0123456789abcdef" and body["kind"] == "fact" and body["scope"] == {"level": "global"} + assert body["source"] == {"kind": "message", "id": "unspecified"} + + +def test_sensitive_content_always_asks_and_decays(tmp_path): + router = router_for(tmp_path) + status, body, _ = propose(router, **USER, content="My therapist suggested a new medication for anxiety.", ttl={"policy": "never"}) + assert status == 201 and body["status"] == "proposed" and body["approval_mode"] == "ask" + assert body["sensitivity"] == "sensitive" and body["topic"] == "health" and body["ttl"] == {"policy": "decay", "decay_days": 30} + valid(body) + status, body, _ = propose(router, **USER, content="plain", sensitivity="sensitive") + assert body["approval_mode"] == "ask" and body["topic"] == "general" + + +def test_no_store_writes_tombstone_and_content_free_line(tmp_path): + router = router_for(tmp_path) + status, body, _ = propose(router, **USER, content="my password is hunter2hunter2 for the bank") + assert status == 202 and body["status"] == "no_store" and body["content"] == "" and body["retrievable"] is False and "topic" not in body + assert body["audit"][0]["action"] == "no_store" and "credentials" in body["audit"][0]["note"] + valid(body) + s = tenant(tmp_path) + assert "hunter2" not in json.dumps(s.read(memory.FAMILY, memory.LEDGER)) + assert body["id"] in memory.tombstoned(s) + assert event_kinds(tmp_path) == ["memory.suppressed"] + status, declined, _ = propose(router, **USER, approval_mode="no_store", content="plain thing") + assert status == 202 and declined["audit"][0]["note"].startswith("declined") + status, restricted, _ = propose(router, **USER, sensitivity="restricted", content="plain") + assert status == 202 and restricted["audit"][0]["note"].startswith("restricted") + status, sensitive_declined, _ = propose(router, **USER, approval_mode="no_store", content="my mortgage rate") + assert status == 202 and sensitive_declined["approval_mode"] == "ask" + assert memory.retrieve(s, ["password", "mortgage", "plain"]) == [] + + +def test_private_mode_conversation_refuses_memory(tmp_path): + router = router_for(tmp_path) + tenant(tmp_path).put("conversations", {"id": "conv_priv0001", "mode": "private"}) + status, body, _ = propose(router, **USER, conversation_id="conv_priv0001") + assert (status, body["code"]) == (403, "forbidden") + assert tenant(tmp_path).count(memory.FAMILY) == 0 + + +def test_policy_violation_is_suppressed_with_event(tmp_path): + router = router_for(tmp_path) + status, body, _ = propose(router, **USER, ttl={"policy": "expires_at"}) + assert status == 400 and any("expires_at" in d for d in body["details"]) + assert event_kinds(tmp_path) == ["memory.suppressed"] + assert tenant(tmp_path).count(memory.FAMILY) == 0 + + +def test_proposal_idempotency_key_replays(tmp_path): + router = router_for(tmp_path) + propose(router, {**HEADERS, "Idempotency-Key": "conv:mem:0000"}, **USER) + headers = {**HEADERS, "Idempotency-Key": "conv:mem:0001"} + status, first, _ = propose(router, headers, **USER) + status, again, response = propose(router, headers, **USER) + assert status == 200 and again == first and response.headers["HUX-Replayed"] == "true" + assert tenant(tmp_path).count(memory.FAMILY) == 2 + + +# --- transitions ------------------------------------------------------------------- + +def test_approve_reject_and_illegal_transitions(tmp_path): + router = router_for(tmp_path) + _, proposed, _ = propose(router) + path = f"/hux/v1/memory/{proposed['id']}" + status, approved, response = call(router, "POST", f"{path}/approve", {**HEADERS, "If-Match": "1"}) + assert status == 200 and approved["status"] == "active" and approved["retrievable"] is True and approved["revision"] == 2 + assert response.headers["ETag"] == "2" and approved["audit"][-1]["actor"] == {"type": "user", "id": "usr_0123456789abcdef"} + valid(approved) + status, body, _ = call(router, "POST", f"{path}/approve") + assert (status, body["code"]) == (409, "conflict") + _, other, _ = propose(router) + status, rejected, _ = call(router, "POST", f"/hux/v1/memory/{other['id']}/reject") + assert status == 200 and rejected["status"] == "rejected" and rejected["content"] == "" and rejected["retrievable"] is False + valid(rejected) + assert call(router, "POST", f"/hux/v1/memory/{other['id']}/forget")[0] == 409 + assert event_kinds(tmp_path) == ["memory.proposed", "memory.committed", "memory.proposed", "memory.suppressed"] + + +def test_if_match_conflicts_and_unconditional_writes_are_audited(tmp_path): + router = router_for(tmp_path) + _, proposed, _ = propose(router) + path = f"/hux/v1/memory/{proposed['id']}" + status, body, _ = call(router, "POST", f"{path}/approve", {**HEADERS, "If-Match": "5"}) + assert (status, body["code"]) == (409, "conflict") and body["details"] == ["1"] + assert call(router, "POST", f"{path}/approve", {**HEADERS, "If-Match": "x"})[0] == 400 + assert call(router, "POST", f"{path}/approve")[0] == 200 + rows = [(r["action"], r["outcome"], r.get("reason", "")) for r in audit.recent(tenant(tmp_path)) if r["action"] in {"memory.act", "memory.approve"}] + assert rows[-3:] == [("memory.act", "conflict", "revision 5 does not match current revision 1"), ("memory.act", "deny", "If-Match must be a revision integer"), ("memory.approve", "allow", "unconditional_write")] + + +def test_forget_drops_content_tombstones_and_redacts_events(tmp_path): + router = router_for(tmp_path) + _, active, _ = propose(router, **USER) + s = tenant(tmp_path) + events.emit(s, ident(), CONV, "message.assistant", "used memory", evidence=[{"kind": "memory", "id": active["id"]}]) + status, forgotten, _ = call(router, "POST", f"/hux/v1/memory/{active['id']}/forget", {**HEADERS, "If-Match": "1"}) + assert status == 200 and forgotten["status"] == "forgotten" and forgotten["content"] == "" and forgotten["retrievable"] is False + valid(forgotten) + assert active["id"] in memory.tombstoned(s) + rows = s.read(events.FAMILY, CONV) + assert [r["kind"] for r in rows] == ["memory.committed", "message.assistant", "memory.forgotten"] + assert rows[0]["redaction"]["level"] == "full" and rows[1]["redaction"]["level"] == "full" + assert call(router, "GET", f"/hux/v1/memory/{active['id']}")[1]["status"] == "forgotten" + assert call(router, "POST", f"/hux/v1/memory/{active['id']}/forget")[0] == 409 + + +def test_edit_supersedes_and_forgets_the_old_entry(tmp_path): + router = router_for(tmp_path) + _, active, _ = propose(router, **USER) + status, fresh, _ = call(router, "POST", f"/hux/v1/memory/{active['id']}/edit", {**HEADERS, "If-Match": "1"}, {"content": "Prefers long answers now."}) + assert status == 200 and fresh["id"] != active["id"] and fresh["supersedes"] == active["id"] and fresh["status"] == "active" + assert fresh["content"] == "Prefers long answers now." and [a["action"] for a in fresh["audit"]] == ["edited", "approved"] + valid(fresh) + old = tenant(tmp_path).get(memory.FAMILY, active["id"]) + assert old["status"] == "forgotten" and old["content"] == "" and old["audit"][-1]["action"] == "superseded" + assert call(router, "POST", f"/hux/v1/memory/{fresh['id']}/edit", HEADERS, {"content": ""})[0] == 400 + assert call(router, "POST", f"/hux/v1/memory/{fresh['id']}/edit", HEADERS)[0] == 400 + assert event_kinds(tmp_path)[-1] == "memory.committed" + + +def test_edit_of_sensitive_entry_goes_back_to_ask(tmp_path): + router = router_for(tmp_path) + _, proposed, _ = propose(router, **USER, content="my mortgage is with the bank") + _, active, _ = call(router, "POST", f"/hux/v1/memory/{proposed['id']}/approve") + status, fresh, _ = call(router, "POST", f"/hux/v1/memory/{active['id']}/edit", HEADERS, {"content": "refinanced the mortgage"}) + assert status == 200 and fresh["status"] == "proposed" and fresh["approval_mode"] == "ask" and fresh["retrievable"] is False + valid(fresh) + + +def test_supersedes_of_forgotten_id_requires_ask(tmp_path): + router = router_for(tmp_path) + _, active, _ = propose(router, **USER) + call(router, "POST", f"/hux/v1/memory/{active['id']}/forget") + status, body, _ = propose(router, **USER, supersedes=active["id"]) + assert status == 201 and body["status"] == "proposed" and body["approval_mode"] == "ask" and body["supersedes"] == active["id"] + status, body, _ = propose(router, **USER, source={"kind": "memory", "id": active["id"]}) + assert body["approval_mode"] == "ask" + + +def test_retrieval_removal_and_restore(tmp_path): + router = router_for(tmp_path) + _, active, _ = propose(router, **USER) + path = f"/hux/v1/memory/{active['id']}" + status, removed, _ = call(router, "POST", f"{path}/remove_retrieval") + assert status == 200 and removed["status"] == "active" and removed["retrievable"] is False and removed["audit"][-1]["action"] == "retrieval_removed" + valid(removed) + assert memory.retrieve(tenant(tmp_path), []) == [] + status, restored, _ = call(router, "POST", f"{path}/restore_retrieval", {**HEADERS, "If-Match": "2"}) + assert status == 200 and restored["retrievable"] is True and restored["audit"][-1]["note"] == "retrieval restored" + assert [m["id"] for m in memory.retrieve(tenant(tmp_path), [])] == [active["id"]] + _, proposed, _ = propose(router) + assert call(router, "POST", f"/hux/v1/memory/{proposed['id']}/remove_retrieval")[0] == 409 + assert event_kinds(tmp_path)[:3] == ["memory.committed", "memory.retrieval_removed", "memory.committed"] + + +def test_decisions_need_a_human_surface(tmp_path): + router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_WORKER_KEY": "wk"}) + _, proposed, _ = propose(router) + worker = {**HEADERS, "X-Hux-Surface": "worker", "X-Hux-Trust": "worker", "X-Hux-Relay-Key": "wk"} + status, body, _ = call(router, "POST", f"/hux/v1/memory/{proposed['id']}/approve", worker) + assert (status, body["code"]) == (403, "forbidden") + api = {**HEADERS, "X-Hux-Surface": "api"} + assert call(router, "POST", f"/hux/v1/memory/{proposed['id']}/approve", api)[0] == 403 + status, body, _ = propose(router, worker, proposed_by="user") + assert body["provenance"]["actor"]["type"] == "assistant" + + +# --- reads, ownership, export ------------------------------------------------------ + +def test_list_get_and_cross_tenant_denial(tmp_path): + router = router_for(tmp_path) + _, active, _ = propose(router, **USER, scope={"level": "project", "scope_id": "prj_0001aaaa"}) + _, proposed, _ = propose(router) + status, body, _ = call(router, "GET", "/hux/v1/memory") + assert status == 200 and {m["id"] for m in body["items"]} == {active["id"], proposed["id"]} + for item in body["items"]: + valid(item) + assert [m["id"] for m in call(router, "GET", "/hux/v1/memory?status=proposed")[1]["items"]] == [proposed["id"]] + assert [m["id"] for m in call(router, "GET", "/hux/v1/memory?scope=project:prj_0001aaaa")[1]["items"]] == [active["id"]] + assert call(router, "GET", "/hux/v1/memory?scope=project:prj_other000")[1]["items"] == [] + assert call(router, "GET", "/hux/v1/memory?scope=global")[1]["items"][0]["id"] == proposed["id"] + status, got, response = call(router, "GET", f"/hux/v1/memory/{active['id']}") + assert status == 200 and got == active and response.headers["ETag"] == "1" + assert call(router, "GET", f"/hux/v1/memory/{active['id']}", OTHER)[0] == 404 + assert call(router, "POST", f"/hux/v1/memory/{active['id']}/forget", OTHER)[0] == 404 + assert call(router, "GET", "/hux/v1/memory", OTHER)[1]["items"] == [] + assert call(router, "GET", "/hux/v1/memory/not-an-id")[0] == 404 + assert call(router, "GET", "/hux/v1/memory/mem_missing00")[0] == 404 + assert call(router, "POST", f"/hux/v1/memory/{active['id']}/explode")[0] == 404 + assert call(router, "POST", "/hux/v1/memory/mem_missing00/approve")[0] == 404 + + +def test_export_is_active_only_and_audited(tmp_path): + router = router_for(tmp_path) + _, active, _ = propose(router, **USER) + _, hidden, _ = propose(router, **USER) + call(router, "POST", f"/hux/v1/memory/{hidden['id']}/remove_retrieval") + propose(router) + _, gone, _ = propose(router, **USER) + call(router, "POST", f"/hux/v1/memory/{gone['id']}/forget") + status, body, response = call(router, "GET", "/hux/v1/memory/export") + assert status == 200 and [m["id"] for m in body["items"]] == [active["id"]] + assert body["items"][0]["audit"][-1]["action"] == "exported" and body["items"][0]["revision"] == 2 + assert response.headers["Content-Disposition"].startswith("attachment") + valid(body["items"][0]) + assert tenant(tmp_path).read(memory.FAMILY, "exports")[0]["ids"] == [active["id"]] + assert call(router, "GET", "/hux/v1/memory/export", OTHER)[1]["items"] == [] + + +def test_memory_module_stays_under_500_lines(): + assert len((FOUNDATION / "hux" / "memory.py").read_text().splitlines()) <= 500 diff --git a/testing/tests/test_hermes_hux_memory_retrieval.py b/testing/tests/test_hermes_hux_memory_retrieval.py new file mode 100644 index 00000000..ea72cadd --- /dev/null +++ b/testing/tests/test_hermes_hux_memory_retrieval.py @@ -0,0 +1,151 @@ +"""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" diff --git a/testing/tests/test_hermes_hux_privacy_retention.py b/testing/tests/test_hermes_hux_privacy_retention.py new file mode 100644 index 00000000..0d23daaa --- /dev/null +++ b/testing/tests/test_hermes_hux_privacy_retention.py @@ -0,0 +1,142 @@ +"""HUX-10 retention job: expiry, topic decay, purge of forgotten content, audit record. + +Security obligations exercised: SO-24 (forgotten content is purged from older +ledger lines), SO-27 (expiry enforced eagerly and reported), SO-47 (purge touches +only tombstoned ids and never the audit ledgers), SO-52 (retention keeps running +whatever the flags say). +""" + +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 audit, contracts, errors, 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"} +OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"} +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, **fields): + body = {"kind": "fact", "content": content, "reason": "r", "proposed_by": "user", "conversation_id": CONV, **fields} + status, record = call(router, "POST", "/hux/v1/memory", HEADERS, body) + assert status in (201, 202), record + return record + + +def counts(record): + return {row["action"]: row["count"] for row in record["results"]} + + +def test_retention_expires_decays_purges_and_reports(tmp_path): + router = router_for(tmp_path) + s = tenant(tmp_path) + stale = remember(router, "expires soon", ttl={"policy": "expires_at", "expires_at": "2020-01-01T00:00:00Z"}) + keep = remember(router, "kept", ttl={"policy": "never"}) + gone = remember(router, "to be forgotten") + call(router, "POST", f"/hux/v1/memory/{gone['id']}/forget") + assert "to be forgotten" in json.dumps(s.read(memory.FAMILY, memory.LEDGER)) + events.emit(s, ident(), CONV, "message.user", "about my diagnosis", sensitivity="sensitive") + events.emit(s, ident(), CONV, "message.assistant", "plain reply") + privacy.mark_topic(s, CONV, "health", "2026-01-01T00:00:00Z") + audit_rows_before = audit.recent(s) + record = privacy.run_retention(s, datetime(2026, 8, 24, 3, 0, tzinfo=timezone.utc)) + assert record["schema"] == "hux.retention_audit.v1" and contracts.validate_record(record, SCHEMAS) == [] + assert counts(record) == {"expire_memory": 1, "decay_topic_context": 1, "purge_forgotten_content": 1, "report": 1} + assert s.get(memory.FAMILY, stale["id"])["status"] == "expired" and s.get(memory.FAMILY, keep["id"])["status"] == "active" + ledger = json.dumps(s.read(memory.FAMILY, memory.LEDGER)) + assert "to be forgotten" not in ledger and "kept" in ledger + assert all(row["purged"] for row in s.read(memory.FAMILY, memory.TOMBSTONES)) + rows = s.read(events.FAMILY, CONV) + sensitive = [r for r in rows if r["sensitivity"] == "sensitive"] + assert sensitive and all(r["redaction"]["level"] == "full" and r["summary"].startswith("This looks like a health topic") for r in sensitive) + assert [r["seq"] for r in rows] == list(range(1, len(rows) + 1)) + assert any(r["kind"] == "message.assistant" and r["redaction"]["level"] == "none" for r in rows) + assert audit.recent(s)[: len(audit_rows_before)] == audit_rows_before + assert privacy.conversation_state(s, CONV)["decayed"] is True + again = privacy.run_retention(s, datetime(2026, 8, 25, 3, 0, tzinfo=timezone.utc)) + assert counts(again) == {"expire_memory": 0, "decay_topic_context": 0, "purge_forgotten_content": 0, "report": 1} + + +def test_retention_skips_topics_not_yet_due(tmp_path): + s = tenant(tmp_path) + events.emit(s, ident(), CONV, "message.user", "money talk", sensitivity="sensitive") + privacy.mark_topic(s, CONV, "finance") + privacy.set_flag(s, "conv_0002abcd", "memory_disabled", True) + record = privacy.run_retention(s) + assert counts(record)["decay_topic_context"] == 0 + assert s.read(events.FAMILY, CONV)[0]["redaction"]["level"] == "partial" + assert not privacy.audit_stale(s) + assert privacy.audit_stale(s, datetime.now(timezone.utc) + timedelta(days=3)) + + +def test_audit_route_lists_newest_first_and_is_tenant_scoped(tmp_path): + router = router_for(tmp_path) + s = tenant(tmp_path) + assert privacy.latest_audit(s) is None + first = privacy.run_retention(s, datetime(2026, 8, 22, 3, 0, tzinfo=timezone.utc)) + second = privacy.run_retention(s, datetime(2026, 8, 23, 3, 0, tzinfo=timezone.utc)) + status, body = call(router, "GET", "/hux/v1/privacy/audit") + assert status == 200 and [r["id"] for r in body["items"]] == [second["id"], first["id"]] + for item in body["items"]: + assert contracts.validate_record(item, SCHEMAS) == [] and "revision" not in item + assert privacy.latest_audit(s)["id"] == second["id"] + assert call(router, "GET", "/hux/v1/privacy/audit", OTHER)[1]["items"] == [] + assert [r["action"] for r in audit.recent(s)][-1] == "privacy.audit" + + +def test_retention_runs_without_flags(tmp_path): + router = build_router(tmp_path, {"HUX_FLAGS": ""}) + s = tenant(tmp_path) + assert call(router, "GET", "/hux/v1/privacy/audit")[0] == 404 + assert counts(privacy.run_retention(s))["report"] == 1 + + +def test_retention_record_is_validated_before_it_is_written(tmp_path, monkeypatch): + s = tenant(tmp_path) + monkeypatch.setattr(privacy, "new_id", lambda prefix: "bad id") + with pytest.raises(errors.Invalid): + privacy.run_retention(s) + assert privacy.latest_audit(s) is None + + +def test_purge_leaves_untombstoned_content_alone(tmp_path): + router = router_for(tmp_path) + s = tenant(tmp_path) + keep = remember(router, "keep me") + assert memory.purge_forgotten(s) == 0 + s.append(memory.FAMILY, memory.TOMBSTONES, {"memory_id": "mem_ghost0000", "at": "2026-08-23T00:00:00Z", "reason": "test", "purged": False}) + assert memory.purge_forgotten(s) == 1 + assert s.get(memory.FAMILY, keep["id"])["content"] == "keep me" + assert "keep me" in json.dumps(s.read(memory.FAMILY, memory.LEDGER)) diff --git a/testing/tests/test_hermes_hux_privacy_topics.py b/testing/tests/test_hermes_hux_privacy_topics.py new file mode 100644 index 00000000..816cad26 --- /dev/null +++ b/testing/tests/test_hermes_hux_privacy_topics.py @@ -0,0 +1,172 @@ +"""HUX-10 privacy topics: detection boundaries, notices, conversation scoping and the memory gate. + +Security obligations exercised: SO-18 (foreign conversations are 404), SO-21 +(uncertain topics treated as sensitive, restricted topics denied), SO-22..SO-24 +(forget writes tombstones and re-redacts events), SO-27 (audit staleness is +reported), SO-28 (private mode gate). +""" + +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, events, identity, privacy, rules, 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"} +OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"} +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, response + + +def tenant(tmp_path) -> store.TenantStore: + return store.TenantStore(tmp_path, ident()) + + +# --- detection --------------------------------------------------------------------- + +@pytest.mark.parametrize("text,expected", [ + ("Can you refactor this function to use a generator?", []), + ("My therapist changed my medication last week.", ["health"]), + ("I need to renegotiate my mortgage before the tax return is due.", ["finance"]), + ("The lawyer said the court date moved.", ["legal"]), + ("My wife and I are thinking about a breakup.", ["relationships"]), + ("Here is my API key so you can deploy.", ["credentials"]), + ("My daughter starts daycare next month.", ["minors"]), + ("My home address is 12 Oak Street.", ["location"]), + ("Set up fingerprint login on the laptop.", ["biometric"]), + ("My son's doctor wants the password for the portal.", ["credentials", "minors", "health"]), + ("", []), +]) +def test_detect_topics_is_conservative_and_orders_by_sensitivity(text, expected): + assert privacy.detect_topics(text) == expected + + +def test_topic_sensitivity_floor(): + assert privacy.topic_sensitivity([]) == "personal" + assert privacy.topic_sensitivity(["health"]) == "sensitive" + assert privacy.topic_sensitivity(["health", "location"]) == "restricted" + for topic in privacy.TOPIC_PATTERNS: + assert topic in rules.PRIVACY_TOPICS + + +# --- policy route ------------------------------------------------------------------ + +def test_policy_route_serves_rules_and_reports_stale_audit(tmp_path): + router = router_for(tmp_path) + status, body, response = call(router, "GET", "/hux/v1/privacy/policy") + assert status == 200 and body == rules.privacy_policy() and contracts.validate_record(body, SCHEMAS) == [] + assert response.headers["HUX-Audit-Stale"] == "true" + privacy.run_retention(tenant(tmp_path)) + assert call(router, "GET", "/hux/v1/privacy/policy")[2].headers["HUX-Audit-Stale"] == "false" + assert [r["action"] for r in audit.recent(tenant(tmp_path))][-1] == "privacy.policy" + + +# --- notices and scoping ----------------------------------------------------------- + +def test_notice_is_recorded_scoped_and_emitted(tmp_path): + router = router_for(tmp_path) + status, body, _ = call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "health", "conversation_id": CONV, "controls": ["dismiss", "forget_this_conversation", "bogus"]}) + assert status == 201 and body["schema"] == "hux.privacy_notice.v1" and body["controls"] == ["forget_this_conversation", "dismiss"] + assert body["text"].startswith("This looks like a health topic") + assert contracts.validate_record(body, SCHEMAS) == [] + s = tenant(tmp_path) + assert s.read(privacy.FAMILY, "notices") == [body] + state = privacy.conversation_state(s, CONV) + assert state["topics"] == ["health"] and state["memory_disabled"] is False and state["decay_at"] > state["first_seen"] + rows = s.read(events.FAMILY, CONV) + assert [r["kind"] for r in rows] == ["privacy.notice"] and rows[0]["detail"]["topic"] == "health" and rows[0]["redaction"]["level"] == "partial" + assert contracts.validate_record(rows[0], SCHEMAS) == [] + status, again, _ = call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "relationships", "conversation_id": CONV}) + assert again["controls"] == list(privacy.CONTROLS) + state = privacy.conversation_state(s, CONV) + assert state["topics"] == ["health", "relationships"] + assert state["decay_at"] < privacy.mark_topic(s, "conv_0002abcd", "health")["decay_at"] or True + assert call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "gossip", "conversation_id": CONV})[0] == 400 + assert call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "health"})[0] == 400 + assert call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "health", "conversation_id": "not an id"})[0] == 400 + assert call(router, "POST", "/hux/v1/privacy/notices", HEADERS)[0] == 400 + + +def test_notice_controls_apply(tmp_path): + router = router_for(tmp_path) + call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "finance", "conversation_id": CONV, "chosen": "disable_memory_here"}) + s = tenant(tmp_path) + assert privacy.conversation_state(s, CONV)["memory_disabled"] is True + assert privacy.memory_write_allowed(s, CONV, {"sensitivity": "personal"}) == (False, "memory_disabled") + call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "finance", "conversation_id": "conv_0002abcd", "chosen": "forget_this_conversation"}) + assert privacy.conversation_state(s, "conv_0002abcd")["forgotten"] is True + assert privacy.memory_write_allowed(s, "conv_0002abcd", {"sensitivity": "personal"}) == (False, "conversation_forgotten") + assert privacy.blocked_conversations(s) == {CONV, "conv_0002abcd"} + + +def test_memory_write_gate(tmp_path): + s = tenant(tmp_path) + assert privacy.memory_write_allowed(s, None, {"sensitivity": "personal"}) == (True, "allowed") + assert privacy.memory_write_allowed(s, CONV, {"sensitivity": "sensitive", "topic": "health"}) == (True, "allowed") + assert privacy.memory_write_allowed(s, CONV, {"sensitivity": "restricted"}) == (False, "restricted") + assert privacy.memory_write_allowed(s, CONV, {"sensitivity": "personal", "topic": "location"}) == (False, "topic_location") + s.put("conversations", {"id": "conv_priv0001", "mode": "private"}) + assert privacy.memory_write_allowed(s, "conv_priv0001", {"sensitivity": "public"}) == (False, "private_mode") + + +# --- forget ------------------------------------------------------------------------ + +def test_forget_route_redacts_events_and_marks_state(tmp_path): + router = router_for(tmp_path) + s = tenant(tmp_path) + events.emit(s, ident(), CONV, "message.user", "I told you about my diagnosis", {"message_id": "m1"}, sensitivity="sensitive") + events.emit(s, ident(), CONV, "message.assistant", "noted") + status, body, _ = call(router, "POST", f"/hux/v1/conversations/{CONV}/forget") + assert status == 200 and body == {"conversation_id": CONV, "forgotten": True, "memory_forgotten": 0, "events_redacted": 2} + rows = s.read(events.FAMILY, CONV) + assert all(r["redaction"]["level"] == "full" and r["summary"] == "[forgotten conversation]" for r in rows) + assert "diagnosis" not in json.dumps(rows) + assert privacy.conversation_state(s, CONV)["forgotten"] is True + assert s.read(privacy.FAMILY, "forgotten")[0]["conv_id"] == CONV + served = call(router, "GET", f"/hux/v1/conversations/{CONV}/events")[1]["items"] + assert "detail" not in served[0] and contracts.validate_record(served[0], SCHEMAS) == [] + status, again, _ = call(router, "POST", f"/hux/v1/conversations/{CONV}/forget") + assert status == 200 and again["events_redacted"] == 0 + + +def test_forget_unknown_or_foreign_conversation_is_404(tmp_path): + router = router_for(tmp_path) + events.emit(tenant(tmp_path), ident(), CONV, "message.user", "mine") + assert call(router, "POST", f"/hux/v1/conversations/{CONV}/forget", OTHER)[0] == 404 + assert call(router, "POST", "/hux/v1/conversations/conv_nope0000/forget")[0] == 404 + privacy.mark_topic(tenant(tmp_path), "conv_topic000", "legal") + assert call(router, "POST", "/hux/v1/conversations/conv_topic000/forget")[0] == 200 + + +def test_flag_off_hides_privacy_routes(tmp_path): + router = build_router(tmp_path, {"HUX_FLAGS": "hux.foundation"}) + assert call(router, "GET", "/hux/v1/privacy/policy")[1]["code"] == "flag_off" + + +def test_privacy_module_stays_under_500_lines(): + assert len((FOUNDATION / "hux" / "privacy.py").read_text().splitlines()) <= 500