From 3ebee2cc155023151e56f11a7d0997f05919a849 Mon Sep 17 00:00:00 2001 From: jenkins Date: Mon, 24 Aug 2026 00:21:51 -0300 Subject: [PATCH] hermes(hux): HUX-05 autonomy engine and minimal HUX-03 organization API Scoped policies resolved through the single capability matrix, grants with server-set expiry, approval queue with once/session/always/deny, pre-side-effect gate that releases a once approval exactly once against the argument hash, run budgets with exhaustion events, honest cancellation receipts; projects, conversations, branch lineage and search over the indexed fields. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RNPhwu2bsaRNg3DETSAZoM --- .../hermes-hux-foundation/hux/budgets.py | 203 ++++++++++ .../hermes-hux-foundation/hux/organization.py | 297 +++++++++++++++ .../hermes-hux-foundation/hux/policy.py | 354 ++++++++++++++++++ .../test_hermes_hux_contract_organization.py | 236 ++++++++++++ .../tests/test_hermes_hux_policy_approvals.py | 309 +++++++++++++++ .../tests/test_hermes_hux_policy_matrix.py | 191 ++++++++++ .../tests/test_hermes_hux_policy_receipts.py | 138 +++++++ 7 files changed, 1728 insertions(+) create mode 100644 dockerfiles/hermes-hux-foundation/hux/budgets.py create mode 100644 dockerfiles/hermes-hux-foundation/hux/organization.py create mode 100644 dockerfiles/hermes-hux-foundation/hux/policy.py create mode 100644 testing/tests/test_hermes_hux_contract_organization.py create mode 100644 testing/tests/test_hermes_hux_policy_approvals.py create mode 100644 testing/tests/test_hermes_hux_policy_matrix.py create mode 100644 testing/tests/test_hermes_hux_policy_receipts.py diff --git a/dockerfiles/hermes-hux-foundation/hux/budgets.py b/dockerfiles/hermes-hux-foundation/hux/budgets.py new file mode 100644 index 00000000..7e16e35b --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/budgets.py @@ -0,0 +1,203 @@ +"""HUX-05 run controls: budgets, the pre-side-effect gate and cancellation receipts. + +The agent hook reports spend per run and asks the gate before every side +effect. The gate releases only against an approved, unexpired approval for the +same run, capability and argument hash; a ``once`` approval is consumed by its +first release (SO-36, SO-37). A stop is not done until a receipt says what +actually happened to the run and its side effects (SO-41). +""" + +from __future__ import annotations + +from typing import Any + +from hux import policy, rules +from hux.errors import Invalid +from hux.http import Request, Response, Router +from hux.identity import Identity +from hux.store import TenantStore, check_id + +BUDGETS = "budgets" +RECEIPTS = "receipts" +SPEND_KEYS = ("tokens", "tool_calls", "wall_clock_seconds", "delegations", "spend_units", "subagents") +LIMIT_OF = { + "tokens": "tokens_per_run", "tool_calls": "tool_calls_per_run", "wall_clock_seconds": "wall_clock_seconds", + "delegations": "delegations_per_run", "spend_units": "spend_units", "subagents": "subagents_per_run", +} + + +def run_id_from(request: Request) -> str: + """The run id in the path; the router already bounded its alphabet.""" + run_id = request.params["id"] + if len(run_id) > 120: + raise Invalid("run id too long") + return run_id + + +# -- budgets ------------------------------------------------------------------- + +def exhausted(spent: dict[str, int], limits: dict[str, Any]) -> list[str]: + """Limit names whose spend has reached them; a zero limit is exhausted immediately.""" + return [LIMIT_OF[k] for k in SPEND_KEYS if LIMIT_OF[k] in limits and spent.get(k, 0) >= limits[LIMIT_OF[k]]] + + +def budget_state(store: TenantStore, identity: Identity, run_id: str, conversation_id: str | None = None) -> dict[str, Any]: + """Current ``hux.budget_state.v1`` for a run measured against the effective policy.""" + doc_id = f"bud_{policy.run_key(run_id)}" + stored = store.get(BUDGETS, doc_id) if store.exists(BUDGETS, doc_id) else {"spent": {}, "_conversation_id": None} + conversation_id = conversation_id or stored.get("_conversation_id") + level, scope_id = ("conversation", conversation_id) if conversation_id else ("global", None) + limits = {k: v for k, v in policy.effective_policy(store, identity, level, scope_id)["budgets"].items() if k != "scope"} + spent = {k: int(stored["spent"].get(k, 0)) for k in SPEND_KEYS} + return policy.checked({ + "schema": "hux.budget_state.v1", "run_id": run_id[:120], "spent": spent, "limits": limits, + "exhausted": exhausted(spent, limits), "_conversation_id": conversation_id, "_id": doc_id, + }) + + +def get_budget(request: Request) -> Response: + """``GET /hux/v1/runs/{id}/budget``: spend so far and what is exhausted.""" + state = budget_state(request.store, request.identity, run_id_from(request)) + request.audit("budgets.read", state["_id"]) + return Response(200, policy.public(state)) + + +def post_budget(request: Request) -> Response: + """``POST /hux/v1/runs/{id}/budget``: add spend increments; emits budget.exhausted on the crossing.""" + body = policy.body_dict(request) + run_id = run_id_from(request) + conversation_id = body.get("conversation_id") + if conversation_id is not None: + conversation_id = check_id(conversation_id) + increments = {} + for key in SPEND_KEYS: + value = body.get(key, 0) + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise Invalid(f"{key} must be a non-negative integer") + increments[key] = value + with request.store.lock(BUDGETS): + before = budget_state(request.store, request.identity, run_id, conversation_id) + known = before["exhausted"] if request.store.exists(BUDGETS, before["_id"]) else [] + spent = {k: before["spent"][k] + increments[k] for k in SPEND_KEYS} + doc = {"id": before["_id"], "run_id": run_id, "spent": spent, "_conversation_id": before["_conversation_id"]} + request.store.put(BUDGETS, doc) + after = budget_state(request.store, request.identity, run_id, conversation_id) + request.audit("budgets.write", after["_id"]) + newly = [k for k in after["exhausted"] if k not in known] + if newly: + policy.emit(request.store, request.identity, after["_conversation_id"], "budget.exhausted", + f"Budget exhausted: {', '.join(newly)}", run_id=run_id) + return Response(200, policy.public(after)) + + +# -- gate ---------------------------------------------------------------------- + +def hashes_of(approval: dict[str, Any]) -> set[str]: + """Argument hashes the approval was requested for (tool_call evidence with a hash).""" + return {e["hash"] for e in approval["request"].get("evidence", []) if e.get("kind") == "tool_call" and e.get("hash")} + + +def matching_approval(store: TenantStore, run_id: str, capability: str, argument_hash: str, external: bool, conversation_id: str | None) -> tuple[dict[str, Any] | None, str, str | None]: + """The approval that releases this side effect, or why none does, plus the run's conversation when an approval names it.""" + reason = "no approval for this run and capability" + for record in store.scan(policy.APPROVALS): + record = policy.refresh(store, record) + same_run = record["run_id"] == run_id + if same_run: + conversation_id = conversation_id or record["conversation_id"] + same_conv = bool(conversation_id) and record["conversation_id"] == conversation_id + choice = record.get("decision", {}).get("choice") + if record["capability"] != capability or not (same_run or (choice == "session" and same_conv)): + continue + if record["status"] != "approved": + reason = f"approval {record['id']} is {record['status']}" + continue + if policy.parse(record["expires_at"]) <= policy.now(): + reason = f"approval {record['id']} has expired" + continue + if external and not record["request"]["external"]: + reason = f"approval {record['id']} was not requested as external" + continue + if choice == "once" and argument_hash not in hashes_of(record): + reason = f"approval {record['id']} was for different arguments" + continue + if choice == "once" and record.get("_consumed_at"): + reason = f"approval {record['id']} was already consumed" + continue + return record, "released", conversation_id + return None, reason, conversation_id + + +def gate(request: Request) -> Response: + """``POST /hux/v1/runs/{id}/gate``: may this side effect proceed right now?""" + body = policy.body_dict(request) + run_id = run_id_from(request) + capability = body.get("capability") + argument_hash = body.get("argument_hash") + if capability not in rules.CAPABILITIES: + raise Invalid("unknown capability") + if not isinstance(argument_hash, str) or not argument_hash.startswith("sha256:"): + raise Invalid("argument_hash must be sha256:") + external = bool(body.get("external", False)) + conversation_id = body.get("conversation_id") + if conversation_id is not None: + conversation_id = check_id(conversation_id) + with request.store.lock(policy.APPROVALS): + record, reason, conversation_id = matching_approval(request.store, run_id, capability, argument_hash, external, conversation_id) + if record is not None and record["decision"]["choice"] == "once": + request.store.put(policy.APPROVALS, {**record, "_consumed_at": policy.iso(policy.now())}) + if record is None: + request.audit("gate.check", f"{run_id}:{capability}", outcome="deny", reason="policy_violation") + policy.emit(request.store, request.identity, conversation_id, "side_effect.blocked", f"{capability} blocked: {reason}"[:280], run_id=run_id) + return Response(200, {"proceed": False, "reason": reason}) + request.audit("gate.check", record["id"]) + policy.emit(request.store, request.identity, conversation_id, "side_effect.released", f"{capability} released by approval {record['id']}", + run_id=run_id, evidence=[{"kind": "approval", "id": record["id"]}]) + return Response(200, {"proceed": True, "approval_id": record["id"], "reason": reason}) + + +# -- stop ---------------------------------------------------------------------- + +def stop(request: Request) -> Response: + """``POST /hux/v1/runs/{id}/stop``: write the cancellation receipt; a repeat returns it.""" + body = policy.body_dict(request) + run_id = run_id_from(request) + receipt_id = f"rcpt_{policy.run_key(run_id)}" + with request.store.lock(RECEIPTS): + if request.store.exists(RECEIPTS, receipt_id): + existing = request.store.get(RECEIPTS, receipt_id) + request.audit("runs.stop", receipt_id, reason="replayed") + return Response(200, policy.public(existing)) + side_effects = body.get("side_effects", []) + if not isinstance(side_effects, list): + raise Invalid("side_effects must be a list") + if body.get("already_complete"): + outcome = "already_complete" + elif body.get("process_registry_empty") is True: + outcome = "cancelled" + else: + outcome = "failed_to_cancel" + stamp = policy.iso(policy.now()) + record: dict[str, Any] = { + "schema": "hux.cancel_receipt.v1", "id": receipt_id, "run_id": run_id, "requested_by": policy.actor_for(request.identity), + "requested_at": stamp, "acknowledged_at": stamp, "outcome": outcome, "side_effects": side_effects, + } + if outcome != "failed_to_cancel": + record["completed_at"] = stamp + conversation_id = body.get("conversation_id") + if conversation_id is not None: + record["conversation_id"] = check_id(conversation_id) + stored = request.store.put(RECEIPTS, policy.checked(record)) + request.audit("runs.stop", receipt_id, reason=outcome) + policy.emit(request.store, request.identity, record.get("conversation_id"), "run.cancelled", f"Run stopped: {outcome}", + run_id=run_id, evidence=[{"kind": "run", "id": receipt_id}]) + return Response(201, policy.public(stored)) + + +def register_routes(router: Router) -> None: + """Attach the run-scoped HUX-05 routes; called by ``hux.policy.register``.""" + router.add("GET", "/hux/v1/runs/{id}/budget", policy.CARD, "budgets.read", get_budget) + router.add("POST", "/hux/v1/runs/{id}/budget", policy.CARD, "budgets.write", post_budget) + router.add("POST", "/hux/v1/runs/{id}/gate", policy.CARD, "gate.check", gate) + router.add("POST", "/hux/v1/runs/{id}/stop", policy.CARD, "runs.stop", stop) + diff --git a/dockerfiles/hermes-hux-foundation/hux/organization.py b/dockerfiles/hermes-hux-foundation/hux/organization.py new file mode 100644 index 00000000..1e3d70c1 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/organization.py @@ -0,0 +1,297 @@ +"""HUX-03 organisation: projects, conversations, branch lineage and search. + +Projects own conversations; conversations carry tags, pins, a mode and an +optional branch pointer to the conversation they forked from. Search covers +only the ``project.schema.json#/$defs/search_index`` fields this increment +indexes (title, tags, project_name, and artifact titles when the artifacts +lane is present); message text is not indexed here and the response says so. +""" + +from __future__ import annotations + +import re +from typing import Any + +from hux import contracts +from hux.errors import Conflict, Invalid, NotFound +from hux.http import Request, Response, Router, page +from hux.store import TenantStore, check_id, new_id, now_iso + +CARD = "HUX-03" +PROJECTS = "projects" +CONVERSATIONS = "conversations" +MAX_PROJECTS = 200 +MAX_CONVERSATIONS = 2000 +PROJECT_FIELDS = ("name", "description", "tags", "pinned", "archived", "default_mode") +CONVERSATION_FIELDS = ("title", "tags", "pinned", "archived", "mode", "project_id") +INDEXED = ("title", "tags", "project_name", "artifact_titles") +NOT_INDEXED = ("message_text",) +SCHEMAS = contracts.load_all() +TOKEN_RE = re.compile(r"[a-z0-9]+") + + +def project_exists(store: TenantStore, project_id: str) -> bool: + """True when this tenant owns a project with that id (helper for other lanes).""" + return isinstance(project_id, str) and bool(re.match(r"^[a-z]{2,6}_[A-Za-z0-9._-]{4,80}$", project_id)) and store.exists(PROJECTS, project_id) + + +def conversation_exists(store: TenantStore, conversation_id: str) -> bool: + """True when this tenant owns a conversation with that id (helper for other lanes).""" + return isinstance(conversation_id, str) and bool(re.match(r"^[a-z]{2,6}_[A-Za-z0-9._-]{4,80}$", conversation_id)) and store.exists(CONVERSATIONS, conversation_id) + + +def project_of(store: TenantStore, conversation_id: str | None) -> str | None: + """The project a conversation belongs to, or None when unknown or unfiled.""" + if not conversation_id or not conversation_exists(store, conversation_id): + return None + return store.get(CONVERSATIONS, conversation_id).get("project_id") + + +def checked(record: dict[str, Any]) -> dict[str, Any]: + """Raise Invalid unless ``record`` satisfies its contract.""" + problems = contracts.validate_record(record, SCHEMAS) + if problems: + raise Invalid("record fails contract", problems) + return record + + +def body_dict(request: Request) -> dict[str, Any]: + """The JSON object body or Invalid.""" + if not isinstance(request.body, dict): + raise Invalid("body must be a JSON object") + return request.body + + +def pick(body: dict[str, Any], fields: tuple[str, ...]) -> dict[str, Any]: + """Only the client-settable fields; ids, owner, timestamps and revision are server-set.""" + return {k: body[k] for k in fields if k in body} + + +def replay(store: TenantStore, family: str, key: str) -> dict[str, Any] | None: + """The record an Idempotency-Key already created in this family, if any.""" + for row in store.read(family, "idempotency"): + if row["key"] == key: + return store.get(family, row["id"]) + return None + + +def create(request: Request, family: str, record: dict[str, Any], key: str, cap: int) -> tuple[dict[str, Any], int]: + """Create under the family lock honouring Idempotency-Key and the family count cap.""" + with request.store.lock(family): + if key: + existing = replay(request.store, family, key) + if existing is not None: + return existing, 200 + if request.store.count(family) >= cap: + raise Conflict(f"{family} cap of {cap} reached") + stored = request.store.put(family, checked({**record, "revision": 1})) + if key: + request.store.append(family, "idempotency", {"key": key, "id": stored["id"]}) + return stored, 201 + + +def update(request: Request, family: str, fields: tuple[str, ...]) -> dict[str, Any]: + """PATCH under If-Match; a missing If-Match is accepted but audited as unconditional (SO-44).""" + changes = pick(body_dict(request), fields) + expected = request.if_match() + with request.store.lock(family): + current = request.store.get(family, check_id(request.params["id"])) + if "project_id" in changes and changes["project_id"] is not None and not project_exists(request.store, changes["project_id"]): + raise NotFound("project not found") + record = {**current, **{k: v for k, v in changes.items() if v is not None}, "updated_at": now_iso()} + if changes.get("project_id", "") is None: + record.pop("project_id", None) + stored = request.store.put(family, checked(record), expected_revision=expected) + request.audit(f"{family}.update", stored["id"], reason="" if expected is not None else "unconditional_write") + return stored + + +def etag(record: dict[str, Any]) -> dict[str, str]: + """Revision as ETag so clients can send it back in If-Match.""" + return {"ETag": str(record["revision"])} + + +# -- projects ------------------------------------------------------------------ + +def create_project(request: Request) -> Response: + """``POST /hux/v1/projects``.""" + body = pick(body_dict(request), PROJECT_FIELDS) + stamp = now_iso() + record = { + "schema": "hux.project.v1", "id": new_id("prj"), "owner": request.identity.subject, + "tags": [], "pinned": False, "archived": False, **body, "created_at": stamp, "updated_at": stamp, + } + stored, status = create(request, PROJECTS, record, request.idempotency_key(), MAX_PROJECTS) + request.audit("projects.create", stored["id"], reason="replayed" if status == 200 else "") + return Response(status, stored, etag(stored)) + + +def list_projects(request: Request) -> Response: + """``GET /hux/v1/projects?archived=``: pinned first, then most recently updated.""" + archived = request.query.get("archived") + items = [p for p in request.store.scan(PROJECTS) if archived is None or p["archived"] == (archived == "true")] + items.sort(key=lambda p: p["updated_at"], reverse=True) + items.sort(key=lambda p: not p["pinned"]) + request.audit("projects.list", "projects") + return page(items) + + +def get_project(request: Request) -> Response: + """``GET /hux/v1/projects/{id}``.""" + record = request.store.get(PROJECTS, check_id(request.params["id"])) + request.audit("projects.read", record["id"]) + return Response(200, record, etag(record)) + + +def patch_project(request: Request) -> Response: + """``PATCH /hux/v1/projects/{id}`` with If-Match.""" + stored = update(request, PROJECTS, PROJECT_FIELDS) + return Response(200, stored, etag(stored)) + + +# -- conversations ------------------------------------------------------------- + +def new_conversation(request: Request, body: dict[str, Any], extra: dict[str, Any] | None = None) -> Response: + """Build, validate and store a conversation from client fields plus server-set extras.""" + if body.get("project_id") is not None and not project_exists(request.store, body["project_id"]): + raise NotFound("project not found") + stamp = now_iso() + record = { + "schema": "hux.conversation.v1", "id": new_id("conv"), "owner": request.identity.subject, + "tags": [], "pinned": False, "archived": False, **{k: v for k, v in body.items() if v is not None}, + **(extra or {}), "artifact_ids": [], "created_at": stamp, "updated_at": stamp, + } + stored, status = create(request, CONVERSATIONS, record, request.idempotency_key(), MAX_CONVERSATIONS) + request.audit("conversations.create", stored["id"], reason="replayed" if status == 200 else "") + return Response(status, stored, etag(stored)) + + +def create_conversation(request: Request) -> Response: + """``POST /hux/v1/conversations``.""" + return new_conversation(request, pick(body_dict(request), CONVERSATION_FIELDS)) + + +def list_conversations(request: Request) -> Response: + """``GET /hux/v1/conversations?project_id=&tag=&pinned=&archived=``: newest activity first.""" + q = request.query + flags = {k: q[k] == "true" for k in ("pinned", "archived") if k in q} + items = [] + for index, record in enumerate(request.store.scan(CONVERSATIONS)): + if "project_id" in q and record.get("project_id") != q["project_id"]: + continue + if "tag" in q and q["tag"] not in record["tags"]: + continue + if any(record[k] != v for k, v in flags.items()): + continue + items.append((record.get("last_message_at", record["updated_at"]), index, record)) + items.sort(key=lambda item: item[:2], reverse=True) + request.audit("conversations.list", "conversations") + return page([record for _, _, record in items]) + + +def get_conversation(request: Request) -> Response: + """``GET /hux/v1/conversations/{id}``.""" + record = request.store.get(CONVERSATIONS, check_id(request.params["id"])) + request.audit("conversations.read", record["id"]) + return Response(200, record, etag(record)) + + +def patch_conversation(request: Request) -> Response: + """``PATCH /hux/v1/conversations/{id}`` with If-Match.""" + stored = update(request, CONVERSATIONS, CONVERSATION_FIELDS) + return Response(200, stored, etag(stored)) + + +def branch_conversation(request: Request) -> Response: + """``POST /hux/v1/conversations/{id}/branch``: fork at a message, keeping project, tags and mode.""" + body = body_dict(request) + point = body.get("branch_point_message_id") + if not isinstance(point, str) or not 1 <= len(point) <= 120: + raise Invalid("branch_point_message_id required") + parent = request.store.get(CONVERSATIONS, check_id(request.params["id"])) + fields = {"title": body.get("title") or f"{parent['title']} (branch)"[:200], "tags": list(parent["tags"]), + "project_id": parent.get("project_id"), "mode": parent.get("mode")} + return new_conversation(request, fields, {"branch": {"parent_conversation_id": parent["id"], "branch_point_message_id": point}}) + + +def lineage(request: Request) -> Response: + """``GET /hux/v1/conversations/{id}/lineage``: ancestors root-first plus direct children.""" + record = request.store.get(CONVERSATIONS, check_id(request.params["id"])) + ancestors: list[dict[str, Any]] = [] + cursor, seen = record, {record["id"]} + while "branch" in cursor and len(ancestors) < 64: + parent_id = cursor["branch"]["parent_conversation_id"] + if parent_id in seen or not request.store.exists(CONVERSATIONS, parent_id): + break + cursor = request.store.get(CONVERSATIONS, parent_id) + seen.add(parent_id) + ancestors.insert(0, cursor) + children = [c for c in request.store.scan(CONVERSATIONS) if c.get("branch", {}).get("parent_conversation_id") == record["id"]] + request.audit("conversations.lineage", record["id"]) + return Response(200, {"conversation": record, "ancestors": ancestors, "children": children}) + + +# -- search -------------------------------------------------------------------- + +def artifact_titles(store: TenantStore, conversation: dict[str, Any]) -> list[str]: + """Artifact titles for a conversation via the artifacts lane, or nothing when it is absent.""" + try: + from hux import artifacts + except ModuleNotFoundError: + return [] + helper = getattr(artifacts, "artifact_titles", None) + return list(helper(store, conversation["id"])) if helper else [] + + +def tokens(text: str) -> list[str]: + """Lowercased alphanumeric terms.""" + return TOKEN_RE.findall(text.lower()) + + +def score(record: dict[str, Any], terms: list[str], project_name: str, titles: list[str]) -> int: + """Title hit beats tag hit beats project/artifact hit; every term must match somewhere.""" + fields = {"title": tokens(record["title"]), "tags": [t for tag in record["tags"] for t in tokens(tag)], + "project_name": tokens(project_name), "artifact_titles": [t for title in titles for t in tokens(title)]} + weight = {"title": 4, "tags": 3, "project_name": 2, "artifact_titles": 1} + total = 0 + for term in terms: + hit = sum(weight[f] * words.count(term) for f, words in fields.items()) + if not hit: + return 0 + total += hit + return total + + +def search(request: Request) -> Response: + """``GET /hux/v1/search?q=&project_id=``: token match over the indexed fields, best first.""" + terms = tokens(request.query.get("q", "")) + if not terms: + raise Invalid("q is required") + project_filter = request.query.get("project_id") + names = {p["id"]: p["name"] for p in request.store.scan(PROJECTS)} + ranked = [] + for record in request.store.scan(CONVERSATIONS): + if project_filter and record.get("project_id") != project_filter: + continue + points = score(record, terms, names.get(record.get("project_id", ""), ""), artifact_titles(request.store, record)) + if points: + ranked.append((points, record)) + ranked.sort(key=lambda pair: (-pair[0], pair[1]["updated_at"])) + request.audit("search.query", "conversations") + return Response(200, {"items": [r for _, r in ranked], "next": None, "scores": {r["id"]: s for s, r in ranked}, + "indexed": list(INDEXED), "not_indexed": list(NOT_INDEXED)}) + + +def register(router: Router) -> None: + """Attach HUX-03 routes.""" + router.add("POST", "/hux/v1/projects", CARD, "projects.create", create_project) + router.add("GET", "/hux/v1/projects", CARD, "projects.list", list_projects) + router.add("GET", "/hux/v1/projects/{id}", CARD, "projects.read", get_project) + router.add("PATCH", "/hux/v1/projects/{id}", CARD, "projects.update", patch_project) + router.add("POST", "/hux/v1/conversations", CARD, "conversations.create", create_conversation) + router.add("GET", "/hux/v1/conversations", CARD, "conversations.list", list_conversations) + router.add("GET", "/hux/v1/conversations/{id}", CARD, "conversations.read", get_conversation) + router.add("PATCH", "/hux/v1/conversations/{id}", CARD, "conversations.update", patch_conversation) + router.add("POST", "/hux/v1/conversations/{id}/branch", CARD, "conversations.branch", branch_conversation) + router.add("GET", "/hux/v1/conversations/{id}/lineage", CARD, "conversations.lineage", lineage) + router.add("GET", "/hux/v1/search", CARD, "search.query", search) diff --git a/dockerfiles/hermes-hux-foundation/hux/policy.py b/dockerfiles/hermes-hux-foundation/hux/policy.py new file mode 100644 index 00000000..048b7d81 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/policy.py @@ -0,0 +1,354 @@ +"""HUX-05 autonomy: policy documents per scope and the approval queue. + +A policy fixes the autonomy level, explicit grants and budgets for a scope +(global, project or conversation). ``rules.effective_decision`` is the only +resolver: an approval request resolves to allow, ask or deny against the most +specific policy, and every external side effect asks regardless. Budgets, +the pre-side-effect gate and cancellation receipts live in ``hux.budgets``; +this module registers their routes so the family stays one card. +""" + +from __future__ import annotations + +import hashlib +from datetime import datetime, timedelta, timezone +from typing import Any + +from hux import contracts, rules +from hux.errors import BudgetExhausted, Conflict, Forbidden, Invalid +from hux.http import Request, Response, Router, page +from hux.identity import Identity +from hux.store import TenantStore, check_id, new_id + +CARD = "HUX-05" +FAMILY = "policy" +APPROVALS = "approvals" +SCOPES = ("global", "project", "conversation") +APPROVAL_TTL = timedelta(hours=24) +ALWAYS_TTL = timedelta(days=30) +SESSION_TTL = timedelta(hours=24) +HUMAN_SURFACES = frozenset({"chat", "telegram", "voice"}) +HUMAN_TRUSTS = frozenset({"router", "relay"}) +DEFAULT_BUDGETS = { + "tokens_per_run": 200000, "tool_calls_per_run": 40, "wall_clock_seconds": 900, + "delegations_per_run": 4, "spend_units": 50, "subagents_per_run": 2, +} +SCHEMAS = contracts.load_all() + + +def clock() -> datetime: + """Current UTC time; tests replace this to move approvals past expiry.""" + return datetime.now(timezone.utc) + + +def now() -> datetime: + """Indirection so monkeypatching ``clock`` reaches every module.""" + return clock() + + +def iso(when: datetime) -> str: + """RFC 3339 second-precision UTC string.""" + return when.strftime("%Y-%m-%dT%H:%M:%SZ") + + +def parse(stamp: str) -> datetime: + """Inverse of ``iso``.""" + return datetime.fromisoformat(stamp.replace("Z", "+00:00")) + + +def run_key(run_id: str) -> str: + """Stable hex handle for a free-form run id so it can name a document.""" + return hashlib.sha256(run_id.encode()).hexdigest()[:32] + + +def actor_for(identity: Identity) -> dict[str, str]: + """The actor a record attributes to this caller: humans are users, hops are system.""" + if identity.trust in HUMAN_TRUSTS and identity.surface in HUMAN_SURFACES: + return {"type": "user", "id": identity.subject} + return {"type": "system", "id": identity.surface} + + +def provenance(identity: Identity) -> dict[str, Any]: + """Server-set provenance; nothing here comes from the body.""" + return {"surface": identity.surface, "actor": actor_for(identity), "recorded_at": iso(now())} + + +def public(record: dict[str, Any], revisioned: bool = False) -> dict[str, Any]: + """Strip store-internal fields before a record leaves the service.""" + return {k: v for k, v in record.items() if not k.startswith("_") and (revisioned or k != "revision")} + + +def emit(store: TenantStore, identity: Identity, conversation_id: str | None, kind: str, summary: str, **extra: Any) -> None: + """Record an activity event through the events lane when it is present.""" + if not conversation_id: + return + try: + from hux import events + except ModuleNotFoundError: + return + events.emit(store, identity, conversation_id, kind, summary, **extra) + + +def checked(record: dict[str, Any]) -> dict[str, Any]: + """Raise Invalid unless ``record`` satisfies its contract; internal fields are ignored.""" + if record["schema"] == "hux.policy.v1": + candidate = public({"revision": 1, **record}, revisioned=True) + else: + candidate = public(record) + problems = contracts.validate_record(candidate, SCHEMAS) + if problems: + raise Invalid("record fails contract", problems) + return record + + +def body_dict(request: Request) -> dict[str, Any]: + """The JSON object body or Invalid.""" + if not isinstance(request.body, dict): + raise Invalid("body must be a JSON object") + return request.body + + +# -- policy documents -------------------------------------------------------- + +def policy_id(level: str, scope_id: str | None) -> str: + """Document id for a scope; global has exactly one.""" + if level not in SCOPES: + raise Invalid("scope must be global, project or conversation") + if level == "global": + return "pol_global" + if scope_id is None or len(scope_id) > 60: + raise Invalid("scope_id required and at most 60 characters for project and conversation scopes") + return f"pol_{level}.{check_id(scope_id)}" + + +def default_policy(identity: Identity) -> dict[str, Any]: + """The ``safe`` policy every tenant starts with.""" + return { + "schema": "hux.policy.v1", "id": "pol_global", "owner": identity.subject, "scope": {"level": "global"}, + "autonomy": "safe", "grants": [], "budgets": dict(DEFAULT_BUDGETS), + "provenance": provenance(identity), "updated_at": iso(now()), + } + + +def global_policy(store: TenantStore, identity: Identity) -> dict[str, Any]: + """Read the global policy, creating the default on first touch.""" + with store.lock(FAMILY): + if store.exists(FAMILY, "pol_global"): + return store.get(FAMILY, "pol_global") + return store.put(FAMILY, checked({**default_policy(identity), "revision": 1})) + + +def effective_policy(store: TenantStore, identity: Identity, level: str, scope_id: str | None) -> dict[str, Any]: + """Most specific policy along conversation -> project -> global.""" + chain: list[tuple[str, str | None]] = [(level, scope_id)] + if level == "conversation": + from hux import organization # lazy: organization never imports policy, but keep import order free + parent = organization.project_of(store, scope_id) + if parent: + chain.append(("project", parent)) + for lvl, sid in chain: + if lvl != "global" and store.exists(FAMILY, policy_id(lvl, sid)): + return store.get(FAMILY, policy_id(lvl, sid)) + return global_policy(store, identity) + + +def normalise_grants(grants: Any, identity: Identity) -> list[dict[str, Any]]: + """Validate grants and pin every one to a server-set expiry of at most 30 days (SO-38).""" + if not isinstance(grants, list): + raise Invalid("grants must be a list") + ceiling = now() + ALWAYS_TTL + out = [] + for grant in grants: + if not isinstance(grant, dict) or grant.get("capability") not in rules.CAPABILITIES: + raise Invalid("grant needs a known capability") + if grant.get("decision") not in ("allow", "ask", "deny"): + raise Invalid("grant decision must be allow, ask or deny") + expires = grant.get("expires_at") + if not isinstance(expires, str) or parse(expires) > ceiling: + expires = iso(ceiling) + out.append({"capability": grant["capability"], "decision": grant["decision"], "expires_at": expires, "granted_by": actor_for(identity)}) + return out + + +def get_policy(request: Request) -> Response: + """``GET /hux/v1/policy?scope=&scope_id=``: the effective policy for a scope.""" + level = request.query.get("scope", "global") + scope_id = request.query.get("scope_id") + policy_id(level, scope_id) + record = effective_policy(request.store, request.identity, level, scope_id) + request.audit("policy.read", record["id"]) + return Response(200, public(record, revisioned=True), {"ETag": str(record["revision"])}) + + +def put_policy(request: Request) -> Response: + """``PUT /hux/v1/policy``: replace the policy for the scope named in the body.""" + body = body_dict(request) + scope = body.get("scope") or {} + if not isinstance(scope, dict): + raise Invalid("scope must be an object") + record_id = policy_id(scope.get("level", ""), scope.get("scope_id")) + if body.get("autonomy") not in ("ask_first", "safe", "autonomous"): + raise Invalid("autonomy must be ask_first, safe or autonomous") + budgets = body.get("budgets", dict(DEFAULT_BUDGETS)) + record = { + "schema": "hux.policy.v1", "id": record_id, "owner": request.identity.subject, "scope": scope, + "autonomy": body["autonomy"], "grants": normalise_grants(body.get("grants", []), request.identity), + "budgets": budgets, "provenance": provenance(request.identity), "updated_at": iso(now()), + } + expected = request.if_match() + with request.store.lock(FAMILY): + exists = request.store.exists(FAMILY, record_id) + stored = request.store.put(FAMILY, checked(record), expected_revision=expected) + reason = "unconditional_write" if exists and expected is None else "" + request.audit("policy.write", record_id, reason=reason) + return Response(200, public(stored, revisioned=True), {"ETag": str(stored["revision"])}) + + +def add_grant(store: TenantStore, identity: Identity, level: str, scope_id: str | None, capability: str, ttl: timedelta) -> None: + """Append an allow grant to a scope's policy after a session/always decision.""" + record_id = policy_id(level, scope_id) + with store.lock(FAMILY): + if store.exists(FAMILY, record_id): + record = store.get(FAMILY, record_id) + else: + scope = {"level": level, **({"scope_id": scope_id} if scope_id else {})} + record = {**global_policy(store, identity), "id": record_id, "scope": scope} + record.pop("revision") + expected = record.pop("revision", None) + grants = [g for g in record["grants"] if g["capability"] != capability] + grants.append({"capability": capability, "decision": "allow", "expires_at": iso(now() + min(ttl, ALWAYS_TTL)), "granted_by": actor_for(identity)}) + record = {**record, "grants": grants[-64:], "provenance": provenance(identity), "updated_at": iso(now())} + store.put(FAMILY, checked(record), expected_revision=expected) + + +# -- approvals ----------------------------------------------------------------- + +def refresh(store: TenantStore, record: dict[str, Any]) -> dict[str, Any]: + """Expire a pending approval whose window has closed (SO-42); terminal records never change.""" + if record["status"] == "pending" and parse(record["expires_at"]) <= now(): + with store.lock(APPROVALS): + record = store.put(APPROVALS, {**record, "status": "expired"}) + return record + + +def load_approval(store: TenantStore, approval_id: str) -> dict[str, Any]: + """Read one approval for this tenant; unknown ids are 404 whoever owns them.""" + return refresh(store, store.get(APPROVALS, check_id(approval_id))) + + +def replay(store: TenantStore, key: str) -> dict[str, Any] | None: + """The approval an Idempotency-Key already created, if any.""" + for row in store.read(APPROVALS, "idempotency"): + if row["key"] == key: + return load_approval(store, row["id"]) + return None + + +def resolve_request(policy: dict[str, Any], capability: str, external: bool) -> str: + """Effective decision for a request; external side effects never auto-allow (SO-39).""" + decision = rules.effective_decision(policy, capability, now()) + if external and decision == "allow": + return "ask" + return decision + + +def create_approval(request: Request) -> Response: + """``POST /hux/v1/approvals``: the agent hook asks before a gated action.""" + body = body_dict(request) + key = request.idempotency_key() + if key: + existing = replay(request.store, key) + if existing is not None: + request.audit("approvals.create", existing["id"], reason="replayed") + return Response(200, public(existing)) + conversation_id = check_id(body.get("conversation_id")) + capability = body.get("capability") + if capability not in rules.CAPABILITIES: + raise Invalid("unknown capability") + from hux import budgets # lazy: budgets imports this module + state = budgets.budget_state(request.store, request.identity, str(body.get("run_id", "")), conversation_id) + if state["exhausted"]: + emit(request.store, request.identity, conversation_id, "budget.exhausted", f"Budget exhausted: {', '.join(state['exhausted'])}", run_id=state["run_id"]) + raise BudgetExhausted("run budget exhausted", state["exhausted"]) + policy = effective_policy(request.store, request.identity, "conversation", conversation_id) + req = body.get("request") if isinstance(body.get("request"), dict) else {} + external = bool(req.get("external", False)) + decision = resolve_request(policy, capability, external) + stamp = now() + record: dict[str, Any] = { + "schema": "hux.approval.v1", "id": new_id("apr"), "run_id": body.get("run_id"), "conversation_id": conversation_id, + "capability": capability, "request": {**req, "external": external}, + "status": {"allow": "approved", "deny": "denied", "ask": "pending"}[decision], + "requested_at": iso(stamp), "expires_at": iso(stamp + APPROVAL_TTL), + } + if decision != "ask": + record["decision"] = {"choice": "once" if decision == "allow" else "deny", "by": {"type": "system", "id": "policy"}, "at": iso(stamp)} + if key: + record["idempotency_key"] = key + with request.store.lock(APPROVALS): + stored = request.store.put(APPROVALS, checked(record)) + if key: + request.store.append(APPROVALS, "idempotency", {"key": key, "id": stored["id"]}) + request.audit("approvals.create", stored["id"], reason=f"policy_{decision}") + kind = "approval.requested" if decision == "ask" else "approval.resolved" + emit(request.store, request.identity, conversation_id, kind, f"{capability}: {record['request'].get('summary', '')}"[:280], + run_id=record["run_id"], evidence=[{"kind": "approval", "id": stored["id"]}]) + return Response(201, public(stored)) + + +def list_approvals(request: Request) -> Response: + """``GET /hux/v1/approvals?status=``: the queue, oldest request first.""" + wanted = request.query.get("status") + if wanted and wanted not in rules.APPROVAL_TRANSITIONS: + raise Invalid("unknown status") + items = [refresh(request.store, r) for r in request.store.scan(APPROVALS)] + items = [public(r) for r in items if not wanted or r["status"] == wanted] + items.sort(key=lambda r: (r["requested_at"], r["id"])) + request.audit("approvals.list", wanted or "all") + return page(items) + + +def decide_approval(request: Request) -> Response: + """``POST /hux/v1/approvals/{id}``: a human answers once, session, always or deny (SO-35).""" + identity = request.identity + if identity.trust not in HUMAN_TRUSTS or identity.surface not in HUMAN_SURFACES: + raise Forbidden("approvals are decided only from a human surface") + choice = body_dict(request).get("choice") + if choice not in ("once", "session", "always", "deny"): + raise Invalid("choice must be once, session, always or deny") + target = "denied" if choice == "deny" else "approved" + with request.store.lock(APPROVALS): + record = load_approval(request.store, request.params["id"]) + if not rules.transition_allowed(rules.APPROVAL_TRANSITIONS, record["status"], target): + raise Conflict(f"approval is {record['status']}; only pending approvals can be decided") + record = {**record, "status": target, "decision": {"choice": choice, "by": actor_for(identity), "at": iso(now())}} + stored = request.store.put(APPROVALS, checked(record)) + if choice == "session": + add_grant(request.store, identity, "conversation", record["conversation_id"], record["capability"], SESSION_TTL) + elif choice == "always": + add_grant(request.store, identity, "global", None, record["capability"], ALWAYS_TTL) + request.audit("approvals.decide", stored["id"], reason=choice) + emit(request.store, identity, record["conversation_id"], "approval.resolved", f"{record['capability']} {target} ({choice})", + run_id=record["run_id"], evidence=[{"kind": "approval", "id": stored["id"]}]) + return Response(200, public(stored)) + + +def get_approval(request: Request) -> Response: + """``GET /hux/v1/approvals/{id}``: one record.""" + record = load_approval(request.store, request.params["id"]) + request.audit("approvals.read", record["id"]) + return Response(200, public(record)) + + +def register(router: Router) -> None: + """Attach HUX-05 routes, including the budget, gate and stop routes from ``hux.budgets``.""" + from hux import budgets + + router.add("GET", "/hux/v1/policy", CARD, "policy.read", get_policy) + router.add("PUT", "/hux/v1/policy", CARD, "policy.write", put_policy) + router.add("POST", "/hux/v1/approvals", CARD, "approvals.create", create_approval) + router.add("GET", "/hux/v1/approvals", CARD, "approvals.list", list_approvals) + router.add("GET", "/hux/v1/approvals/{id}", CARD, "approvals.read", get_approval) + router.add("POST", "/hux/v1/approvals/{id}", CARD, "approvals.decide", decide_approval) + budgets.register_routes(router) + diff --git a/testing/tests/test_hermes_hux_contract_organization.py b/testing/tests/test_hermes_hux_contract_organization.py new file mode 100644 index 00000000..8b57b160 --- /dev/null +++ b/testing/tests/test_hermes_hux_contract_organization.py @@ -0,0 +1,236 @@ +"""HUX-03 projects, conversations, branch lineage and search. + +Security obligations exercised: SO-44 (If-Match on PATCH, 409 on mismatch, +unconditional writes audited) plus the storage rules every family shares: +ids and provenance are server-set, records the caller does not own are 404, +and every served record validates against ``project.schema.json``. +""" + +from __future__ import annotations + +import json +import sys +import types +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)) + +import hux # noqa: E402 +from hux import audit, contracts, identity, organization, 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"]) + + +@pytest.fixture +def router(tmp_path): + return build_router(tmp_path, {"HUX_FLAGS": ALL_ON}) + + +def call(router, method, path, body=None, headers=HEADERS): + raw = b"" if body is None else json.dumps(body).encode() + response = router.dispatch(method, path, headers, raw) + return response.status, response.body, response.headers + + +def valid(record): + problems = contracts.validate_record(record, SCHEMAS) + assert problems == [], problems + return record + + +def make_project(router, **fields): + status, body, _ = call(router, "POST", "/hux/v1/projects", {"name": "Kitchen", **fields}) + assert status == 201, body + return valid(body) + + +def make_conversation(router, **fields): + status, body, _ = call(router, "POST", "/hux/v1/conversations", {"title": "Cabinets", **fields}) + assert status == 201, body + return valid(body) + + +def test_module_stays_under_the_line_budget(): + assert len((FOUNDATION / "hux" / "organization.py").read_text().splitlines()) <= 500 + + +# --- projects ------------------------------------------------------------------- + +def test_project_create_read_list_and_server_set_fields(router): + body = make_project(router, tags=["home"], pinned=True, default_mode="research", id="prj_evil", owner="usr_ffffffffffffffff", revision=9) + assert body["id"].startswith("prj_") and body["id"] != "prj_evil" and body["owner"] == HEADERS["X-Hux-Subject"] and body["revision"] == 1 + assert body["created_at"] == body["updated_at"] and body["archived"] is False + status, got, headers = call(router, "GET", f"/hux/v1/projects/{body['id']}") + assert (status, got, headers["ETag"]) == (200, body, "1") + second = make_project(router, name="Garden") + status, listing, _ = call(router, "GET", "/hux/v1/projects") + assert [p["id"] for p in listing["items"]] == [body["id"], second["id"]], "pinned first" + assert call(router, "GET", "/hux/v1/projects?archived=true")[1]["items"] == [] + assert call(router, "GET", "/hux/v1/projects?archived=false")[1]["items"] == listing["items"] + + +def test_project_patch_needs_if_match_and_audits_unconditional_writes(router, tmp_path): + project = make_project(router) + path = f"/hux/v1/projects/{project['id']}" + status, body, headers = call(router, "PATCH", path, {"name": "Kitchen v2", "archived": True, "owner": "usr_ffffffffffffffff"}, {**HEADERS, "If-Match": "1"}) + assert status == 200 and valid(body)["name"] == "Kitchen v2" and body["archived"] is True and body["revision"] == 2 and headers["ETag"] == "2" + assert body["owner"] == project["owner"] and body["updated_at"] >= project["updated_at"] + status, error, _ = call(router, "PATCH", path, {"name": "stale"}, {**HEADERS, "If-Match": "1"}) + assert (status, error["code"]) == (409, "conflict") and valid(error) + status, body, _ = call(router, "PATCH", path, {"description": "no If-Match"}) + assert status == 200 and body["revision"] == 3 + assert call(router, "PATCH", path, {"name": ""}, {**HEADERS, "If-Match": "3"})[0] == 400 + assert call(router, "PATCH", path, {"default_mode": "turbo"}, {**HEADERS, "If-Match": "3"})[0] == 400 + assert call(router, "PATCH", path, [], {**HEADERS, "If-Match": "3"})[0] == 400 + rows = [(r["action"], r["outcome"], r.get("reason", "")) for r in audit.recent(store.TenantStore(tmp_path, identity.resolve(HEADERS, {}))) if r["action"] == "projects.update"] + assert rows[0] == ("projects.update", "allow", "") and rows[1][1] == "conflict" and rows[2] == ("projects.update", "allow", "unconditional_write") + + +def test_project_idempotency_and_cap(router, monkeypatch): + key = {**HEADERS, "Idempotency-Key": "chat:project:0001"} + status, first, _ = call(router, "POST", "/hux/v1/projects", {"name": "Once"}, key) + status, again, _ = call(router, "POST", "/hux/v1/projects", {"name": "Twice"}, key) + assert (status, again) == (200, first) + status, other, _ = call(router, "POST", "/hux/v1/projects", {"name": "Other key"}, {**HEADERS, "Idempotency-Key": "chat:project:0002"}) + assert status == 201 and other["id"] != first["id"] + assert call(router, "POST", "/hux/v1/projects", {"name": "x" * 121})[0] == 400 + assert call(router, "POST", "/hux/v1/projects", None)[0] == 400 + monkeypatch.setattr(organization, "MAX_PROJECTS", 1) + status, error, _ = call(router, "POST", "/hux/v1/projects", {"name": "Too many"}) + assert (status, error["code"]) == (409, "conflict") + assert call(router, "POST", "/hux/v1/projects", {"name": "Replays still work"}, key)[0] == 200 + + +def test_projects_are_invisible_to_a_second_subject(router): + project = make_project(router) + assert call(router, "GET", f"/hux/v1/projects/{project['id']}", headers=OTHER)[0] == 404 + assert call(router, "PATCH", f"/hux/v1/projects/{project['id']}", {"name": "mine now"}, {**OTHER, "If-Match": "1"})[0] == 404 + assert call(router, "GET", "/hux/v1/projects", headers=OTHER)[1]["items"] == [] + assert call(router, "GET", "/hux/v1/projects/prj_missing0001")[0] == 404 + assert call(router, "GET", "/hux/v1/projects/bad")[0] == 400 + + +# --- conversations ------------------------------------------------------------- + +def test_conversation_create_list_filters_and_patch(router): + project = make_project(router) + first = make_conversation(router, project_id=project["id"], tags=["suppliers"], mode="research") + second = make_conversation(router, title="Loose", pinned=True) + assert first["project_id"] == project["id"] and first["artifact_ids"] == [] and second.get("project_id") is None + assert call(router, "POST", "/hux/v1/conversations", {"title": "Orphan", "project_id": "prj_missing0001"})[0] == 404 + assert call(router, "POST", "/hux/v1/conversations", {"title": "", "tags": ["Bad Tag"]})[0] == 400 + items = lambda query: [c["id"] for c in call(router, "GET", f"/hux/v1/conversations{query}")[1]["items"]] # noqa: E731 + assert items("") == [second["id"], first["id"]] + assert items(f"?project_id={project['id']}") == [first["id"]] + assert items("?tag=suppliers") == [first["id"]] and items("?tag=nope") == [] + assert items("?pinned=true") == [second["id"]] and items("?archived=true") == [] + path = f"/hux/v1/conversations/{first['id']}" + status, body, _ = call(router, "PATCH", path, {"title": "Cabinet suppliers", "archived": True, "project_id": None}, {**HEADERS, "If-Match": "1"}) + assert status == 200 and valid(body)["title"] == "Cabinet suppliers" and "project_id" not in body and body["revision"] == 2 + assert items("?archived=true") == [first["id"]] + assert call(router, "PATCH", path, {"project_id": "prj_missing0001"}, {**HEADERS, "If-Match": "2"})[0] == 404 + assert call(router, "PATCH", path, {"title": "stale"}, {**HEADERS, "If-Match": "1"})[0] == 409 + status, body, headers = call(router, "GET", path) + assert status == 200 and headers["ETag"] == "2" and valid(body) + assert call(router, "GET", path, headers=OTHER)[0] == 404 + assert call(router, "PATCH", path, {"title": "x"}, {**OTHER, "If-Match": "2"})[0] == 404 + + +def test_helpers_for_other_lanes(router, tmp_path): + project = make_project(router) + conversation = make_conversation(router, project_id=project["id"]) + mine = store.TenantStore(tmp_path, identity.resolve(HEADERS, {})) + theirs = store.TenantStore(tmp_path, identity.resolve(OTHER, {})) + assert organization.project_exists(mine, project["id"]) and not organization.project_exists(theirs, project["id"]) + assert organization.conversation_exists(mine, conversation["id"]) and not organization.conversation_exists(theirs, conversation["id"]) + assert not organization.project_exists(mine, "../escape") and not organization.conversation_exists(mine, None) + assert organization.project_of(mine, conversation["id"]) == project["id"] + assert organization.project_of(mine, None) is None and organization.project_of(mine, "conv_missing0001") is None + + +# --- branches and lineage --------------------------------------------------------- + +def test_branch_copies_project_tags_and_mode_and_lineage_walks_both_ways(router): + project = make_project(router) + root = make_conversation(router, project_id=project["id"], tags=["a", "b"], mode="create") + status, child, _ = call(router, "POST", f"/hux/v1/conversations/{root['id']}/branch", {"branch_point_message_id": "msg-12"}) + assert status == 201 and valid(child)["branch"] == {"parent_conversation_id": root["id"], "branch_point_message_id": "msg-12"} + assert child["project_id"] == project["id"] and child["tags"] == ["a", "b"] and child["mode"] == "create" and child["title"] == "Cabinets (branch)" + status, grandchild, _ = call(router, "POST", f"/hux/v1/conversations/{child['id']}/branch", {"branch_point_message_id": "msg-3", "title": "Deeper"}) + assert grandchild["title"] == "Deeper" + sibling = call(router, "POST", f"/hux/v1/conversations/{root['id']}/branch", {"branch_point_message_id": "msg-1"})[1] + status, body, _ = call(router, "GET", f"/hux/v1/conversations/{grandchild['id']}/lineage") + assert status == 200 and body["conversation"] == grandchild + assert [a["id"] for a in body["ancestors"]] == [root["id"], child["id"]] and body["children"] == [] + body = call(router, "GET", f"/hux/v1/conversations/{root['id']}/lineage")[1] + assert body["ancestors"] == [] and {c["id"] for c in body["children"]} == {child["id"], sibling["id"]} + for record in body["children"]: + valid(record) + assert call(router, "POST", f"/hux/v1/conversations/{root['id']}/branch", {})[0] == 400 + assert call(router, "POST", f"/hux/v1/conversations/{root['id']}/branch", {"branch_point_message_id": "m"}, OTHER)[0] == 404 + assert call(router, "GET", f"/hux/v1/conversations/{root['id']}/lineage", headers=OTHER)[0] == 404 + + +def test_lineage_survives_a_missing_or_cyclic_parent(router, tmp_path): + orphan = make_conversation(router) + mine = store.TenantStore(tmp_path, identity.resolve(HEADERS, {})) + mine.put("conversations", {**orphan, "branch": {"parent_conversation_id": "conv_gone00000001", "branch_point_message_id": "m"}}) + assert call(router, "GET", f"/hux/v1/conversations/{orphan['id']}/lineage")[1]["ancestors"] == [] + mine.put("conversations", {**mine.get("conversations", orphan["id"]), "branch": {"parent_conversation_id": orphan["id"], "branch_point_message_id": "m"}}) + body = call(router, "GET", f"/hux/v1/conversations/{orphan['id']}/lineage")[1] + assert body["ancestors"] == [] and [c["id"] for c in body["children"]] == [orphan["id"]] + + +# --- search ----------------------------------------------------------------------- + +def test_search_ranks_title_over_tag_over_project_and_says_what_is_indexed(router): + kitchen = make_project(router, name="Kitchen renovation") + by_title = make_conversation(router, title="Cabinet suppliers compared", project_id=kitchen["id"]) + by_tag = make_conversation(router, title="Budget", tags=["cabinet"]) + by_project = make_conversation(router, title="Flooring", project_id=kitchen["id"]) + make_conversation(router, title="Unrelated") + status, body, _ = call(router, "GET", "/hux/v1/search?q=cabinet") + assert status == 200 and [c["id"] for c in body["items"]] == [by_title["id"], by_tag["id"]] + assert body["indexed"] == ["title", "tags", "project_name", "artifact_titles"] and body["not_indexed"] == ["message_text"] + assert body["scores"][by_title["id"]] > body["scores"][by_tag["id"]] and body["next"] is None + for record in body["items"]: + valid(record) + assert [c["id"] for c in call(router, "GET", "/hux/v1/search?q=kitchen")[1]["items"]] == [by_title["id"], by_project["id"]] + assert [c["id"] for c in call(router, "GET", "/hux/v1/search?q=cabinet+kitchen")[1]["items"]] == [by_title["id"]], "terms are ANDed" + assert call(router, "GET", f"/hux/v1/search?q=kitchen&project_id={kitchen['id']}")[1]["items"] == call(router, "GET", "/hux/v1/search?q=kitchen")[1]["items"] + assert call(router, "GET", "/hux/v1/search?q=kitchen&project_id=prj_other0000001")[1]["items"] == [] + assert call(router, "GET", "/hux/v1/search?q=%21%21")[0] == 400 + assert call(router, "GET", "/hux/v1/search")[0] == 400 + assert call(router, "GET", "/hux/v1/search?q=cabinet", headers=OTHER)[1]["items"] == [] + + +def test_search_uses_artifact_titles_when_the_artifacts_lane_offers_them(router, monkeypatch): + conversation = make_conversation(router, title="Plain") + fake = types.ModuleType("hux.artifacts") + fake.artifact_titles = lambda store, conversation_id: ["Supplier comparison sheet"] if conversation_id == conversation["id"] else [] + monkeypatch.setitem(sys.modules, "hux.artifacts", fake) + monkeypatch.setattr(hux, "artifacts", fake, raising=False) + assert [c["id"] for c in call(router, "GET", "/hux/v1/search?q=supplier")[1]["items"]] == [conversation["id"]] + bare = types.ModuleType("hux.artifacts") + monkeypatch.setitem(sys.modules, "hux.artifacts", bare) + monkeypatch.setattr(hux, "artifacts", bare, raising=False) + assert call(router, "GET", "/hux/v1/search?q=supplier")[1]["items"] == [] + monkeypatch.setitem(sys.modules, "hux.artifacts", None) + monkeypatch.delattr(hux, "artifacts", raising=False) + assert call(router, "GET", "/hux/v1/search?q=supplier")[1]["items"] == [] + assert call(router, "GET", "/hux/v1/search?q=plain")[1]["items"] == [conversation] + + +def test_flag_off_hides_the_card(tmp_path): + off = build_router(tmp_path, {"HUX_FLAGS": ""}) + status, body, _ = call(off, "GET", "/hux/v1/projects") + assert (status, body["code"]) == (404, "flag_off") diff --git a/testing/tests/test_hermes_hux_policy_approvals.py b/testing/tests/test_hermes_hux_policy_approvals.py new file mode 100644 index 00000000..a371f74e --- /dev/null +++ b/testing/tests/test_hermes_hux_policy_approvals.py @@ -0,0 +1,309 @@ +"""HUX-05 approval queue and the pre-side-effect gate. + +Security obligations exercised: SO-35 (only a human surface with router or +relay trust decides; ``decision.by`` is the asserted user), SO-36 (``once`` +is consumed exactly once), SO-37 (the gate must present the argument hash +recorded at request time), SO-39 (external side effects always need an +approval record), SO-40 (an exhausted budget blocks new approvals), SO-42 +(pending approvals expire 24 h after ``requested_at``). +""" + +from __future__ import annotations + +import importlib +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)) + +import hux # noqa: E402 +from hux import audit, budgets, contracts, identity, policy, store # noqa: E402 +from hux import events as hux_events # 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"} +WORKER = {**HEADERS, "X-Hux-Surface": "worker", "X-Hux-Trust": "worker", "X-Hux-Relay-Key": "wk"} +RELAY = {**HEADERS, "X-Hux-Surface": "telegram", "X-Hux-Trust": "relay", "X-Hux-Relay-Key": "rk"} +ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) +T0 = datetime(2026, 8, 24, 12, 0, tzinfo=timezone.utc) +CONV = "conv_0001abcd" +HASH = "sha256:" + "ab" * 32 +OTHER_HASH = "sha256:" + "cd" * 32 + + +@pytest.fixture +def router(tmp_path): + return build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_WORKER_KEY": "wk", "HUX_RELAY_KEY": "rk"}) + + +@pytest.fixture +def frozen(monkeypatch): + state = {"now": T0} + monkeypatch.setattr(policy, "clock", lambda: state["now"]) + return state + + +@pytest.fixture +def events(monkeypatch): + """Stand-in for the events lane; records every emit call.""" + seen: list[dict] = [] + monkeypatch.setattr(hux_events, "emit", lambda store, identity, conversation_id, kind, summary, **extra: seen.append({"conversation_id": conversation_id, "kind": kind, "summary": summary, **extra})) + return seen + + +def call(router, method, path, body=None, headers=HEADERS): + raw = b"" if body is None else json.dumps(body).encode() + response = router.dispatch(method, path, headers, raw) + return response.status, response.body + + +def valid(record): + problems = contracts.validate_record(record, SCHEMAS) + assert problems == [], problems + return record + + +def request_body(capability="write_files", external=False, run_id="run_9f", **extra): + evidence = [{"kind": "tool_call", "id": "call-7", "hash": HASH}] + return {"run_id": run_id, "conversation_id": CONV, "capability": capability, + "request": {"summary": f"do {capability}", "risk": "low", "external": external, "evidence": evidence, **extra}} + + +def set_autonomy(router, level): + status, body = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": level}) + assert status == 200, body + + +# --- creation resolves against the matrix ------------------------------------------ + +def test_safe_policy_auto_allows_reads_and_queues_mutations(router, frozen, events): + status, body = call(router, "POST", "/hux/v1/approvals", request_body("read_files"), WORKER) + assert status == 201 and valid(body)["status"] == "approved" + assert body["decision"] == {"choice": "once", "by": {"type": "system", "id": "policy"}, "at": policy.iso(T0)} + status, body = call(router, "POST", "/hux/v1/approvals", request_body("write_files"), WORKER) + assert status == 201 and valid(body)["status"] == "pending" and "decision" not in body + assert body["expires_at"] == policy.iso(T0 + timedelta(hours=24)) + status, body = call(router, "POST", "/hux/v1/approvals", request_body("network"), WORKER) + assert status == 201 and valid(body)["status"] == "denied" and body["decision"]["choice"] == "deny" + assert [e["kind"] for e in events] == ["approval.resolved", "approval.requested", "approval.resolved"] + assert events[0]["evidence"] == [{"kind": "approval", "id": events[0]["evidence"][0]["id"]}] + + +def test_external_side_effects_always_need_a_human_regardless_of_autonomy(router, events): + set_autonomy(router, "autonomous") + status, body = call(router, "POST", "/hux/v1/approvals", request_body("send_message"), WORKER) + assert status == 201 and body["status"] == "approved" + status, body = call(router, "POST", "/hux/v1/approvals", request_body("send_message", external=True), WORKER) + assert status == 201 and body["status"] == "pending" + for always_ask in ("deploy", "external_side_effect"): + status, body = call(router, "POST", "/hux/v1/approvals", request_body(always_ask), WORKER) + assert body["status"] == "pending", always_ask + call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous", "grants": [{"capability": "network", "decision": "deny"}]}) + status, body = call(router, "POST", "/hux/v1/approvals", request_body("network", external=True), WORKER) + assert body["status"] == "denied", "deny beats external ask" + + +def test_idempotency_key_returns_the_original(router): + key = {"Idempotency-Key": "run_9f:approval:call-7"} + status, first = call(router, "POST", "/hux/v1/approvals", request_body(), {**WORKER, **key}) + assert status == 201 and first["idempotency_key"] == "run_9f:approval:call-7" + status, again = call(router, "POST", "/hux/v1/approvals", request_body("shell"), {**WORKER, **key}) + assert (status, again) == (200, first) + status, other = call(router, "POST", "/hux/v1/approvals", request_body(), {**WORKER, "Idempotency-Key": "run_9f:approval:call-8"}) + assert status == 201 and other["id"] != first["id"] + assert call(router, "GET", f"/hux/v1/approvals/{first['id']}")[1] == first + + +@pytest.mark.parametrize("body,status", [ + (None, 400), ([], 400), ({"conversation_id": "nope", "capability": "shell"}, 400), + ({"conversation_id": CONV, "capability": "teleport"}, 400), + ({"conversation_id": CONV, "capability": "shell", "run_id": "r"}, 400), + ({"conversation_id": CONV, "capability": "shell", "run_id": "r", "request": {"summary": "", "risk": "low"}}, 400), + ({"conversation_id": CONV, "capability": "shell", "run_id": "r", "request": {"summary": "x", "risk": "silly"}}, 400), +]) +def test_create_rejects_bad_bodies(router, body, status): + got, error = call(router, "POST", "/hux/v1/approvals", body, WORKER) + assert got == status and valid(error)["code"] == "invalid" + + +# --- deciding ----------------------------------------------------------------------- + +def pending(router, capability="write_files", **kw): + status, body = call(router, "POST", "/hux/v1/approvals", request_body(capability, **kw), WORKER) + assert status == 201 and body["status"] == "pending", body + return body + + +def test_only_humans_on_router_or_relay_decide(router, tmp_path): + record = pending(router) + status, error = call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}, WORKER) + assert (status, error["code"]) == (403, "forbidden") + api = {**HEADERS, "X-Hux-Surface": "api"} + assert call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}, api)[0] == 403 + status, body = call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}, RELAY) + assert status == 200 and valid(body)["status"] == "approved" + assert body["decision"]["by"] == {"type": "user", "id": HEADERS["X-Hux-Subject"]} + rows = audit.recent(store.TenantStore(tmp_path, identity.resolve(HEADERS, {}))) + assert [(r["action"], r["outcome"]) for r in rows if r["action"] == "approvals.decide"] == [("approvals.decide", "deny")] * 2 + [("approvals.decide", "allow")] + + +def test_terminal_states_never_change_again(router, events): + record = pending(router) + path = f"/hux/v1/approvals/{record['id']}" + assert call(router, "POST", path, {"choice": "maybe"})[0] == 400 + assert call(router, "POST", path, [])[0] == 400 + status, body = call(router, "POST", path, {"choice": "deny"}) + assert status == 200 and body["status"] == "denied" and body["decision"]["choice"] == "deny" + status, error = call(router, "POST", path, {"choice": "once"}) + assert (status, error["code"]) == (409, "conflict") + assert call(router, "GET", path)[1] == body + assert events[-1]["kind"] == "approval.resolved" and "denied" in events[-1]["summary"] + + +def test_pending_approvals_expire_after_24h(router, frozen): + record = pending(router) + frozen["now"] = T0 + timedelta(hours=24) + status, body = call(router, "GET", f"/hux/v1/approvals/{record['id']}") + assert status == 200 and valid(body)["status"] == "expired" + assert call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"})[0] == 409 + assert call(router, "GET", "/hux/v1/approvals?status=expired")[1]["items"] == [body] + assert call(router, "GET", "/hux/v1/approvals?status=pending")[1]["items"] == [] + + +def test_session_and_always_create_grants_at_their_scope(router, frozen): + record = pending(router, "shell") + status, body = call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "session"}) + assert status == 200 and body["decision"]["choice"] == "session" + conv_policy = call(router, "GET", f"/hux/v1/policy?scope=conversation&scope_id={CONV}")[1] + assert valid(conv_policy)["scope"] == {"level": "conversation", "scope_id": CONV} + assert conv_policy["grants"] == [{"capability": "shell", "decision": "allow", "expires_at": policy.iso(T0 + policy.SESSION_TTL), "granted_by": {"type": "user", "id": HEADERS["X-Hux-Subject"]}}] + status, body = call(router, "POST", "/hux/v1/approvals", request_body("shell"), WORKER) + assert body["status"] == "approved", "later shell requests in this conversation auto-approve" + assert call(router, "POST", "/hux/v1/approvals", request_body("write_files"), WORKER)[1]["status"] == "pending" + record = pending(router, "write_files") + call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "always"}) + global_policy = call(router, "GET", "/hux/v1/policy")[1] + assert [g["capability"] for g in global_policy["grants"]] == ["write_files"] + assert global_policy["grants"][0]["expires_at"] == policy.iso(T0 + policy.ALWAYS_TTL) + frozen["now"] = T0 + timedelta(hours=25) + record = pending(router, "shell", run_id="run_other") + call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "session"}) + conv_policy = call(router, "GET", f"/hux/v1/policy?scope=conversation&scope_id={CONV}")[1] + assert len(conv_policy["grants"]) == 1 and conv_policy["revision"] == 2, "a repeat grant replaces, never duplicates" + + +def test_list_filters_and_orders_by_request_time(router, frozen): + first = pending(router, "shell") + frozen["now"] = T0 + timedelta(seconds=5) + second = pending(router, "write_files") + status, body = call(router, "GET", "/hux/v1/approvals") + assert status == 200 and [i["id"] for i in body["items"]] == [first["id"], second["id"]] and body["next"] is None + assert call(router, "GET", "/hux/v1/approvals?status=pending")[1]["items"] == body["items"] + assert call(router, "GET", "/hux/v1/approvals?status=approved")[1]["items"] == [] + assert call(router, "GET", "/hux/v1/approvals?status=bogus")[0] == 400 + for item in body["items"]: + valid(item) + + +def test_second_subject_cannot_see_or_decide(router): + record = pending(router) + assert call(router, "GET", f"/hux/v1/approvals/{record['id']}", headers=OTHER)[0] == 404 + assert call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}, OTHER)[0] == 404 + assert call(router, "GET", "/hux/v1/approvals", headers=OTHER)[1]["items"] == [] + assert call(router, "GET", "/hux/v1/approvals/apr_doesnotexist")[0] == 404 + assert call(router, "GET", "/hux/v1/approvals/..")[0] == 400 + + +# --- budgets block new approvals ------------------------------------------------------- + +def test_exhausted_budget_blocks_new_approvals(router, events): + call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "safe", "budgets": {"tool_calls_per_run": 2}}) + status, body = call(router, "POST", "/hux/v1/runs/run_9f/budget", {"conversation_id": CONV, "tool_calls": 2}, WORKER) + assert status == 200 and valid(body)["exhausted"] == ["tool_calls_per_run"] + status, error = call(router, "POST", "/hux/v1/approvals", request_body("read_files"), WORKER) + assert (status, error["code"]) == (429, "budget_exhausted") and error["details"] == ["tool_calls_per_run"] + assert [e["kind"] for e in events] == ["budget.exhausted", "budget.exhausted"] + status, body = call(router, "POST", "/hux/v1/approvals", request_body("read_files", run_id="run_fresh"), WORKER) + assert status == 201, "another run keeps its own budget" + + +# --- gate -------------------------------------------------------------------------- + +def gate(router, run_id="run_9f", capability="write_files", argument_hash=HASH, external=False, **extra): + return call(router, "POST", f"/hux/v1/runs/{run_id}/gate", {"capability": capability, "argument_hash": argument_hash, "external": external, **extra}, WORKER) + + +def test_gate_blocks_before_approval_and_releases_once_exactly_once(router, events, tmp_path): + status, body = gate(router) + assert status == 200 and body == {"proceed": False, "reason": "no approval for this run and capability"} + record = pending(router) + assert gate(router)[1]["proceed"] is False and "pending" in gate(router)[1]["reason"] + call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}) + status, wrong = gate(router, argument_hash=OTHER_HASH) + assert wrong["proceed"] is False and "different arguments" in wrong["reason"], "TOCTOU (SO-37)" + status, body = gate(router) + assert body == {"proceed": True, "approval_id": record["id"], "reason": "released"} + status, again = gate(router) + assert again["proceed"] is False and "already consumed" in again["reason"], "once is once (SO-36)" + kinds = [e["kind"] for e in events] + assert kinds.count("side_effect.released") == 1 and kinds.count("side_effect.blocked") == 4, "the first block predates any approval, so no conversation is known" + released = next(e for e in events if e["kind"] == "side_effect.released") + assert released["evidence"] == [{"kind": "approval", "id": record["id"]}] and released["run_id"] == "run_9f" + served = call(router, "GET", f"/hux/v1/approvals/{record['id']}")[1] + assert valid(served) and "_consumed_at" not in served + rows = audit.recent(store.TenantStore(tmp_path, identity.resolve(HEADERS, {}))) + assert [r["outcome"] for r in rows if r["action"] == "gate.check"] == ["deny", "deny", "deny", "deny", "allow", "deny"] + + +def test_gate_session_is_reusable_within_the_conversation(router): + record = pending(router, "shell") + call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "session"}) + assert gate(router, capability="shell")[1]["proceed"] is True + assert gate(router, capability="shell", argument_hash=OTHER_HASH)[1]["proceed"] is True + assert gate(router, run_id="run_later", capability="shell", conversation_id=CONV)[1]["proceed"] is True + assert gate(router, run_id="run_later", capability="shell")[1]["proceed"] is False + assert gate(router, run_id="run_later", capability="shell", conversation_id="conv_elsewhere01")[1]["proceed"] is False + assert gate(router, capability="write_files")[1]["proceed"] is False + + +def test_gate_requires_external_approvals_for_external_effects_and_respects_expiry(router, frozen): + record = pending(router, "send_message") + call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}) + status, body = gate(router, capability="send_message", external=True) + assert body["proceed"] is False and "not requested as external" in body["reason"] + record = pending(router, "send_message", external=True) + call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}) + frozen["now"] = T0 + timedelta(hours=25) + status, body = gate(router, capability="send_message", external=True) + assert body["proceed"] is False and "expired" in body["reason"] + + +def test_gate_rejects_bad_bodies_and_other_tenants(router): + assert gate(router, capability="teleport")[0] == 400 + assert gate(router, argument_hash="md5:zz")[0] == 400 + assert gate(router, conversation_id="nope")[0] == 400 + assert call(router, "POST", "/hux/v1/runs/run_9f/gate", [], WORKER)[0] == 400 + assert call(router, "POST", f"/hux/v1/runs/{'r' * 121}/gate", {"capability": "shell", "argument_hash": HASH}, WORKER)[0] == 400 + record = pending(router) + call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}) + status, body = call(router, "POST", "/hux/v1/runs/run_9f/gate", {"capability": "write_files", "argument_hash": HASH}, OTHER) + assert status == 200 and body["proceed"] is False + + +def test_events_module_absence_is_tolerated(router, monkeypatch): + monkeypatch.setitem(sys.modules, "hux.events", None) + monkeypatch.delattr(hux, "events") + with pytest.raises(ImportError): + importlib.import_module("hux.events") + assert call(router, "POST", "/hux/v1/approvals", request_body("read_files"), WORKER)[0] == 201 + policy.emit(None, None, None, "x", "no conversation means no event") + assert budgets.hashes_of({"request": {}}) == set() diff --git a/testing/tests/test_hermes_hux_policy_matrix.py b/testing/tests/test_hermes_hux_policy_matrix.py new file mode 100644 index 00000000..7d6a66a8 --- /dev/null +++ b/testing/tests/test_hermes_hux_policy_matrix.py @@ -0,0 +1,191 @@ +"""HUX-05 policy documents: the capability matrix, grants and their expiry. + +Security obligations exercised: SO-38 (``always`` grants carry a server-set +expiry of at most 30 days), SO-39 (``effective_decision`` is the sole +resolver and deploy/external always ask), SO-44 (If-Match on revisioned +writes; unconditional writes are audited). +""" + +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, identity, policy, 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"]) +T0 = datetime(2026, 8, 24, 12, 0, tzinfo=timezone.utc) + + +@pytest.fixture +def router(tmp_path): + return build_router(tmp_path, {"HUX_FLAGS": ALL_ON}) + + +@pytest.fixture +def frozen(monkeypatch): + state = {"now": T0} + monkeypatch.setattr(policy, "clock", lambda: state["now"]) + return state + + +def call(router, method, path, body=None, headers=HEADERS): + raw = b"" if body is None else json.dumps(body).encode() + response = router.dispatch(method, path, headers, raw) + return response.status, response.body, response.headers + + +def valid(record): + assert contracts.validate_record(record, SCHEMAS) == [], contracts.validate_record(record, SCHEMAS) + return record + + +def test_modules_stay_under_the_line_budget(): + for name in ("policy.py", "budgets.py"): + assert len((FOUNDATION / "hux" / name).read_text().splitlines()) <= 500 + + +def test_first_read_creates_the_safe_global_default(router, tmp_path): + status, body, headers = call(router, "GET", "/hux/v1/policy") + assert status == 200 and headers["ETag"] == "1" + valid(body) + assert body["scope"] == {"level": "global"} and body["autonomy"] == "safe" and body["grants"] == [] + assert body["owner"] == HEADERS["X-Hux-Subject"] and body["provenance"]["actor"] == {"type": "user", "id": body["owner"]} + assert call(router, "GET", "/hux/v1/policy?scope=global")[1] == body + rows = audit.recent(store.TenantStore(tmp_path, identity.resolve(HEADERS, {}))) + assert [r["action"] for r in rows] == ["policy.read", "policy.read"] + + +def test_matrix_is_the_single_resolver_and_external_always_asks(router): + matrix = rules.default_capability_matrix() + for level in ("ask_first", "safe", "autonomous"): + put = {"scope": {"level": "global"}, "autonomy": level, "grants": [{"capability": "deploy", "decision": "allow"}]} + status, body, _ = call(router, "PUT", "/hux/v1/policy", put) + assert status == 200, body + for capability in rules.CAPABILITIES: + expected = matrix[level][capability] + if capability == "deploy": + expected = "ask" # an explicit allow on an always-ask capability still asks (SO-39) + assert policy.resolve_request(body, capability, external=False) == expected + assert policy.resolve_request(body, capability, external=True) in {"ask", "deny"} + assert policy.resolve_request(body, "network", external=True) == "ask" + + +def test_grants_get_server_set_expiry_and_actor(router, frozen): + far = policy.iso(T0 + timedelta(days=90)) + soon = policy.iso(T0 + timedelta(days=2)) + put = {"scope": {"level": "global"}, "autonomy": "safe", "grants": [ + {"capability": "network", "decision": "allow", "expires_at": far, "granted_by": {"type": "operator", "id": "evil"}}, + {"capability": "shell", "decision": "allow", "expires_at": soon}, + {"capability": "write_files", "decision": "deny"}, + ]} + status, body, _ = call(router, "PUT", "/hux/v1/policy", put) + assert status == 200 + valid(body) + by_cap = {g["capability"]: g for g in body["grants"]} + assert by_cap["network"]["expires_at"] == policy.iso(T0 + policy.ALWAYS_TTL) + assert by_cap["shell"]["expires_at"] == soon + assert by_cap["write_files"]["expires_at"] == policy.iso(T0 + policy.ALWAYS_TTL) + assert all(g["granted_by"] == {"type": "user", "id": HEADERS["X-Hux-Subject"]} for g in body["grants"]) + assert policy.resolve_request(body, "network", False) == "allow" + assert policy.resolve_request(body, "shell", False) == "allow" + assert policy.resolve_request(body, "write_files", False) == "deny" + frozen["now"] = T0 + timedelta(days=3) + assert policy.resolve_request(body, "shell", False) == "ask", "expired grant falls back to the matrix" + assert policy.resolve_request(body, "network", False) == "allow" + frozen["now"] = T0 + timedelta(days=31) + assert policy.resolve_request(body, "network", False) == "deny", "safe denies network once the grant lapses" + + +def test_put_honours_if_match_and_audits_unconditional_writes(router, tmp_path): + put = {"scope": {"level": "global"}, "autonomy": "autonomous"} + status, body, headers = call(router, "PUT", "/hux/v1/policy", put) + assert (status, body["revision"], headers["ETag"]) == (200, 1, "1") + status, body, _ = call(router, "PUT", "/hux/v1/policy", put, {**HEADERS, "If-Match": "1"}) + assert (status, body["revision"]) == (200, 2) + status, body, _ = call(router, "PUT", "/hux/v1/policy", put, {**HEADERS, "If-Match": "1"}) + assert (status, body["code"]) == (409, "conflict") + status, body, _ = call(router, "PUT", "/hux/v1/policy", put) + assert (status, body["revision"]) == (200, 3) + rows = [(r["action"], r["outcome"], r.get("reason", "")) for r in audit.recent(store.TenantStore(tmp_path, identity.resolve(HEADERS, {})))] + assert rows == [("policy.write", "allow", ""), ("policy.write", "allow", ""), ("policy.write", "conflict", rows[2][2]), ("policy.write", "allow", "unconditional_write")] + assert "does not match" in rows[2][2] + + +@pytest.mark.parametrize("body,fragment", [ + ([], "JSON object"), + ({"scope": "global"}, "scope must be an object"), + ({"scope": {"level": "planet"}, "autonomy": "safe"}, "scope must be"), + ({"scope": {"level": "project"}, "autonomy": "safe"}, "scope_id required"), + ({"scope": {"level": "project", "scope_id": "prj_" + "a" * 70}, "autonomy": "safe"}, "scope_id required"), + ({"scope": {"level": "project", "scope_id": "../etc"}, "autonomy": "safe"}, "malformed id"), + ({"scope": {"level": "global"}, "autonomy": "yolo"}, "autonomy must be"), + ({"scope": {"level": "global"}, "autonomy": "safe", "grants": "all"}, "grants must be a list"), + ({"scope": {"level": "global"}, "autonomy": "safe", "grants": [{"capability": "teleport", "decision": "allow"}]}, "known capability"), + ({"scope": {"level": "global"}, "autonomy": "safe", "grants": [{"capability": "shell", "decision": "maybe"}]}, "grant decision"), + ({"scope": {"level": "global"}, "autonomy": "safe", "budgets": {"tokens_per_run": -1}}, "fails contract"), + ({"scope": {"level": "global"}, "autonomy": "safe", "budgets": {"unknown": 1}}, "fails contract"), +]) +def test_put_rejects_bad_bodies(router, body, fragment): + status, error, _ = call(router, "PUT", "/hux/v1/policy", body) + assert status == 400 and fragment in error["message"], error + valid(error) + + +def test_get_rejects_bad_scope_queries(router): + assert call(router, "GET", "/hux/v1/policy?scope=nope")[0] == 400 + assert call(router, "GET", "/hux/v1/policy?scope=conversation")[0] == 400 + + +def test_effective_policy_walks_conversation_project_global(router): + _, project, _ = call(router, "POST", "/hux/v1/projects", {"name": "Kitchen"}) + _, conversation, _ = call(router, "POST", "/hux/v1/conversations", {"title": "Cabinets", "project_id": project["id"]}) + _, loose, _ = call(router, "POST", "/hux/v1/conversations", {"title": "Unfiled"}) + conv_query = f"/hux/v1/policy?scope=conversation&scope_id={conversation['id']}" + assert call(router, "GET", conv_query)[1]["scope"] == {"level": "global"} + status, body, _ = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "project", "scope_id": project["id"]}, "autonomy": "autonomous"}) + assert status == 200 and valid(body)["id"] == f"pol_project.{project['id']}" + assert call(router, "GET", conv_query)[1]["scope"] == {"level": "project", "scope_id": project["id"]} + assert call(router, "GET", f"/hux/v1/policy?scope=conversation&scope_id={loose['id']}")[1]["scope"] == {"level": "global"} + assert call(router, "GET", "/hux/v1/policy?scope=conversation&scope_id=conv_unknown0001")[1]["scope"] == {"level": "global"} + status, body, _ = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "conversation", "scope_id": conversation["id"]}, "autonomy": "ask_first"}) + assert status == 200 + got = call(router, "GET", conv_query)[1] + assert got["autonomy"] == "ask_first" and got["scope"]["level"] == "conversation" + assert call(router, "GET", f"/hux/v1/policy?scope=project&scope_id={project['id']}")[1]["autonomy"] == "autonomous" + + +def test_second_subject_sees_its_own_default_not_the_first_tenants_policy(router): + call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"}) + status, body, _ = call(router, "GET", "/hux/v1/policy", headers=OTHER) + assert status == 200 and body["autonomy"] == "safe" and body["owner"] == OTHER["X-Hux-Subject"] + assert call(router, "GET", "/hux/v1/policy")[1]["autonomy"] == "autonomous" + + +def test_flag_off_hides_the_whole_card(tmp_path): + off = build_router(tmp_path, {"HUX_FLAGS": "hux.foundation"}) + status, body, _ = call(off, "GET", "/hux/v1/policy") + assert (status, body["code"]) == (404, "flag_off") + + +def test_helpers_round_trip_time_and_actors(): + assert policy.parse(policy.iso(T0)) == T0 + assert policy.now().tzinfo is timezone.utc + worker = identity.Identity("slot-3", HEADERS["X-Hux-Subject"], "worker", "worker") + assert policy.actor_for(worker) == {"type": "system", "id": "worker"} + assert policy.run_key("run_9f") == policy.run_key("run_9f") and len(policy.run_key("x")) == 32 + assert policy.public({"a": 1, "_b": 2, "revision": 3}) == {"a": 1} + assert policy.public({"a": 1, "_b": 2, "revision": 3}, revisioned=True) == {"a": 1, "revision": 3} diff --git a/testing/tests/test_hermes_hux_policy_receipts.py b/testing/tests/test_hermes_hux_policy_receipts.py new file mode 100644 index 00000000..8158e398 --- /dev/null +++ b/testing/tests/test_hermes_hux_policy_receipts.py @@ -0,0 +1,138 @@ +"""HUX-05 run budgets and cancellation receipts. + +Security obligations exercised: SO-40 (budget state per run against the +effective policy; crossing a limit emits ``budget.exhausted``), SO-41 (a +receipt says ``cancelled`` only when the process registry is empty and +carries every side effect the hook reports; a second stop returns the same +receipt). +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation" +if str(FOUNDATION) not in sys.path: + sys.path.insert(0, str(FOUNDATION)) + +from hux import audit, contracts, identity, store # noqa: E402 +from hux import events as hux_events # 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"} +WORKER = {**HEADERS, "X-Hux-Surface": "worker", "X-Hux-Trust": "worker", "X-Hux-Relay-Key": "wk"} +ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) +CONV = "conv_0001abcd" + + +@pytest.fixture +def router(tmp_path): + return build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_WORKER_KEY": "wk"}) + + +@pytest.fixture +def events(monkeypatch): + seen: list[dict] = [] + monkeypatch.setattr(hux_events, "emit", lambda store, identity, conversation_id, kind, summary, **extra: seen.append({"conversation_id": conversation_id, "kind": kind, "summary": summary, **extra})) + return seen + + +def call(router, method, path, body=None, headers=WORKER): + raw = b"" if body is None else json.dumps(body).encode() + response = router.dispatch(method, path, headers, raw) + return response.status, response.body + + +def valid(record): + problems = contracts.validate_record(record, SCHEMAS) + assert problems == [], problems + return record + + +# --- budgets ------------------------------------------------------------------------ + +def test_budget_starts_empty_against_the_effective_policy(router): + status, body = call(router, "GET", "/hux/v1/runs/run_9f/budget") + assert status == 200 and valid(body)["run_id"] == "run_9f" + assert body["spent"] == {k: 0 for k in ("tokens", "tool_calls", "wall_clock_seconds", "delegations", "spend_units", "subagents")} + assert body["limits"]["tokens_per_run"] == 200000 and body["exhausted"] == [] + assert "revision" not in body and not any(k.startswith("_") for k in body) + + +def test_increments_accumulate_and_exhaust_each_limit(router, events): + call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "safe", + "budgets": {"tokens_per_run": 100, "tool_calls_per_run": 3, "subagents_per_run": 0, "scope": {"paths": ["/work"]}}}, HEADERS) + status, body = call(router, "POST", "/hux/v1/runs/run_9f/budget", {"conversation_id": CONV, "tokens": 60, "tool_calls": 1}) + assert status == 200 and valid(body)["spent"]["tokens"] == 60 + assert body["exhausted"] == ["subagents_per_run"], "a zero limit is exhausted before any spend" + assert body["limits"] == {"tokens_per_run": 100, "tool_calls_per_run": 3, "subagents_per_run": 0}, "scope is policy detail, not a limit" + status, body = call(router, "POST", "/hux/v1/runs/run_9f/budget", {"tokens": 40, "tool_calls": 2, "delegations": 5}) + assert body["spent"] == {"tokens": 100, "tool_calls": 3, "wall_clock_seconds": 0, "delegations": 5, "spend_units": 0, "subagents": 0} + assert body["exhausted"] == ["tokens_per_run", "tool_calls_per_run", "subagents_per_run"] + assert [e["kind"] for e in events] == ["budget.exhausted", "budget.exhausted"] + assert events[1]["summary"] == "Budget exhausted: tokens_per_run, tool_calls_per_run" and events[1]["run_id"] == "run_9f" + assert events[1]["conversation_id"] == CONV, "the conversation learned on the first report sticks" + status, body = call(router, "POST", "/hux/v1/runs/run_9f/budget", {"tokens": 1}) + assert status == 200 and len(events) == 2, "already exhausted limits do not re-emit" + assert call(router, "GET", "/hux/v1/runs/run_9f/budget")[1] == body + + +def test_budget_follows_the_conversation_policy(router): + _, conversation = call(router, "POST", "/hux/v1/conversations", {"title": "Budgeted"}, HEADERS) + call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "conversation", "scope_id": conversation["id"]}, "autonomy": "safe", "budgets": {"spend_units": 1}}, HEADERS) + status, body = call(router, "POST", "/hux/v1/runs/run_c/budget", {"conversation_id": conversation["id"], "spend_units": 1}) + assert body["limits"] == {"spend_units": 1} and body["exhausted"] == ["spend_units"] + assert call(router, "GET", "/hux/v1/runs/run_c/budget")[1]["limits"] == {"spend_units": 1} + assert call(router, "GET", "/hux/v1/runs/run_c/budget", headers=OTHER)[1]["spent"]["spend_units"] == 0 + + +@pytest.mark.parametrize("body", [[], {"tokens": -1}, {"tokens": True}, {"tokens": "5"}, {"conversation_id": "x"}]) +def test_budget_rejects_bad_bodies(router, body): + status, error = call(router, "POST", "/hux/v1/runs/run_9f/budget", body) + assert status == 400 and valid(error)["code"] == "invalid" + + +# --- stop receipts ---------------------------------------------------------------- + +def test_stop_writes_a_receipt_and_a_second_stop_returns_it(router, events, tmp_path): + side_effects = [{"description": "Partial file left in workspace", "reverted": True, "evidence": {"kind": "file", "id": "notes.md"}}] + status, body = call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": True, "side_effects": side_effects, "conversation_id": CONV}, HEADERS) + assert status == 201 and valid(body)["outcome"] == "cancelled" + assert body["requested_by"] == {"type": "user", "id": HEADERS["X-Hux-Subject"]} + assert body["requested_at"] == body["acknowledged_at"] == body["completed_at"] + assert body["side_effects"] == side_effects and body["conversation_id"] == CONV and "revision" not in body + status, again = call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": False}) + assert (status, again) == (200, body) + assert [e["kind"] for e in events] == ["run.cancelled"] + assert events[0]["evidence"] == [{"kind": "run", "id": body["id"]}] and events[0]["run_id"] == "run_9f" + rows = [(r["action"], r.get("reason")) for r in audit.recent(store.TenantStore(tmp_path, identity.resolve(HEADERS, {}))) if r["action"] == "runs.stop"] + assert rows == [("runs.stop", "cancelled"), ("runs.stop", "replayed")] + + +def test_stop_outcomes_follow_the_process_registry(router): + status, body = call(router, "POST", "/hux/v1/runs/run_a/stop", {"process_registry_empty": False, "side_effects": []}) + assert status == 201 and valid(body)["outcome"] == "failed_to_cancel" and "completed_at" not in body + assert body["requested_by"] == {"type": "system", "id": "worker"} + status, body = call(router, "POST", "/hux/v1/runs/run_b/stop", {"already_complete": True}) + assert valid(body)["outcome"] == "already_complete" and "completed_at" in body + status, body = call(router, "POST", "/hux/v1/runs/run_c/stop", {}) + assert valid(body)["outcome"] == "failed_to_cancel", "no registry report means the stop is not proven" + + +@pytest.mark.parametrize("body", [[], {"side_effects": "none"}, {"side_effects": [{"description": ""}]}, {"conversation_id": "bad"}]) +def test_stop_rejects_bad_bodies(router, body): + status, error = call(router, "POST", "/hux/v1/runs/run_9f/stop", body) + assert status == 400 and valid(error)["code"] == "invalid" + + +def test_receipts_are_per_tenant(router): + call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": True}) + status, body = call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": False}, OTHER) + assert status == 201 and body["outcome"] == "failed_to_cancel", "same run id, other subject, its own receipt"