security(hux): private mode denies web, messaging, shell and delegation

HUX-06/HUX-10: a conversation whose stored friendly mode is private now
has network, web_search, send_message, shell and delegate refused by
both the approval resolver and the pre-side-effect gate, regardless of
autonomy policy — matching the mode catalog's tool contract. Memory
writes were already refused by the privacy state. Missing or unbound
conversations keep the existing matrix behaviour.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BvMSXH8VH2tMWXanb8SJdf

dockerfiles/hermes-hux-foundation/hux/budgets.py

@@ -179,6 +179,10 @@ def gate(request: Request) -> Response:
         raise Invalid("argument_hash must be sha256:<hex>")
     external = bool(body.get("external", False))
     conversation_id = run_conversation(request.store, run_id)
+    if policy.private_mode_denies(request.store, conversation_id, capability):
+        request.audit("gate.check", f"{run_id}:{capability}", outcome="deny", reason="private_mode")
+        policy.emit(request.store, request.identity, conversation_id, "side_effect.blocked", f"{capability} blocked: private mode", run_id=run_id)
+        return Response(200, {"proceed": False, "reason": "private_mode"})
     state = budget_state(request.store, request.identity, run_id, conversation_id)
     if state["exhausted"]:  # F6: an exhausted run releases nothing, whatever was approved
         request.audit("gate.check", f"{run_id}:{capability}", outcome="deny", reason="budget_exhausted")

dockerfiles/hermes-hux-foundation/hux/policy.py

@@ -290,6 +290,25 @@ def resolve_request(policy: dict[str, Any], capability: str, external: bool) ->
     return decision
 
 
+PRIVATE_DENIED = frozenset({"network", "web_search", "send_message", "shell", "delegate"})
+
+
+def private_mode_denies(store: TenantStore, conversation_id: str | None, capability: str) -> bool:
+    """Private conversations never release web, shell, messaging or delegation.
+
+    The mode catalog says so (``rules.MODE_CATALOG['private']['tools']``);
+    memory writes are already refused by the privacy state. The conversation
+    record's ``mode`` field is written only by the HUX-06 selection route.
+    """
+    if capability not in PRIVATE_DENIED or not conversation_id:
+        return False
+    try:
+        record = store.get("conversations", conversation_id)
+    except Exception:
+        return False
+    return record.get("mode") == "private"
+
+
 def create_approval(request: Request) -> Response:
     """``POST /hux/v1/approvals``: the agent hook asks before a gated action."""
     require_worker(request.identity, "approval requests")
@@ -322,6 +341,8 @@ def create_approval(request: Request) -> Response:
         raise BudgetExhausted("run budget exhausted", state["exhausted"])
     policy = effective_policy(request.store, request.identity, "conversation", conversation_id)
     decision = resolve_request(policy, capability, external)
+    if private_mode_denies(request.store, conversation_id, capability):
+        decision = "deny"
     stamp = now()
     record: dict[str, Any] = {
         "schema": "hux.approval.v1", "id": new_id("apr"), "run_id": run_id, "conversation_id": conversation_id,

testing/tests/test_hermes_hux_policy_private_mode.py

@@ -0,0 +1,128 @@
+"""Private mode denies web, messaging, shell and delegation upstream.
+
+HUX-06/HUX-10: selecting the private friendly mode must change what the
+agent may actually do — not merely relabel the conversation. The approval
+resolver and the pre-side-effect gate both refuse the denied capability
+set for a conversation whose stored mode is ``private``, whatever the
+autonomy policy says.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
+if str(FOUNDATION) not in sys.path:
+    sys.path.insert(0, str(FOUNDATION))
+
+from hux import contracts, policy  # noqa: E402
+from hux.server import build_router  # noqa: E402
+
+HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
+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"])
+HASH = "sha256:" + "ab" * 32
+
+
+def call(router, method, path, body=None, headers=None):
+    raw = b"" if body is None else json.dumps(body).encode()
+    response = router.dispatch(method, path, {**HEADERS, **(headers or {})}, raw)
+    return response.status, response.body
+
+
+@pytest.fixture
+def scoped(tmp_path):
+    router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk", "HUX_WORKER_KEY": "wk"})
+    _, project = call(router, "POST", "/hux/v1/projects", {"name": "P"})
+    _, conversation = call(router, "POST", "/hux/v1/conversations", {"title": "C", "project_id": project["id"]})
+    status, _ = call(
+        router, "POST", "/hux/v1/runs/run_9f/budget",
+        {"conversation_id": conversation["id"]}, WORKER,
+    )
+    assert status == 200
+    return router, project["id"], conversation["id"]
+
+
+def _select_mode(router, project_id, conversation_id, mode):
+    status, body = call(
+        router, "PUT",
+        f"/hux/v1/projects/{project_id}/conversations/{conversation_id}/mode",
+        {"project_id": project_id, "mode": mode},
+        {"If-Match": "0", "Idempotency-Key": f"mode-{mode}-0001"},
+    )
+    assert status == 200, body
+
+
+def _approval(router, conversation_id, capability, external=False):
+    body = {
+        "run_id": "run_9f", "conversation_id": conversation_id, "capability": capability,
+        "request": {"summary": f"do {capability}", "risk": "low", "external": external,
+                    "evidence": [{"kind": "tool_call", "id": "call-7", "hash": HASH}]},
+    }
+    return call(router, "POST", "/hux/v1/approvals", body, WORKER)
+
+
+def _gate(router, capability, external=False):
+    return call(
+        router, "POST", "/hux/v1/runs/run_9f/gate",
+        {"capability": capability, "argument_hash": HASH, "external": external,
+         "conversation_id": "ignored"},
+        WORKER,
+    )
+
+
+def test_private_mode_denies_the_catalog_denied_capabilities(scoped):
+    router, project_id, conversation_id = scoped
+    status, _ = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"})
+    assert status == 200
+    _select_mode(router, project_id, conversation_id, "private")
+    for capability in sorted(policy.PRIVATE_DENIED):
+        status, body = _approval(router, conversation_id, capability)
+        assert status == 201, body
+        assert body["status"] == "denied", (capability, body)
+        assert body["decision"]["by"] == {"type": "system", "id": "policy"}
+    # Reading files and writing artifacts stay governed by the normal matrix.
+    status, body = _approval(router, conversation_id, "read_files")
+    assert status == 201 and body["status"] == "approved"
+
+
+def test_gate_refuses_private_mode_even_with_an_earlier_approval(scoped):
+    router, project_id, conversation_id = scoped
+    status, _ = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"})
+    assert status == 200
+    status, approved = _approval(router, conversation_id, "network")
+    assert status == 201 and approved["status"] == "approved"
+    _select_mode(router, project_id, conversation_id, "private")
+    status, verdict = _gate(router, "network")
+    assert status == 200
+    assert verdict == {"proceed": False, "reason": "private_mode"}
+
+
+def test_non_private_modes_do_not_touch_the_matrix(scoped):
+    router, project_id, conversation_id = scoped
+    status, _ = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"})
+    assert status == 200
+    _select_mode(router, project_id, conversation_id, "fast")
+    status, body = _approval(router, conversation_id, "network")
+    assert status == 201 and body["status"] == "approved"
+    status, verdict = _gate(router, "network")
+    assert status == 200 and verdict["proceed"] is True
+
+
+def test_unbound_or_missing_conversations_fail_open_to_the_matrix(tmp_path):
+    router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk", "HUX_WORKER_KEY": "wk"})
+    status, _ = call(
+        router, "POST", "/hux/v1/runs/run_9f/budget",
+        {"conversation_id": "conv_0001abcd"}, WORKER,
+    )
+    assert status == 200
+    status, _ = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"})
+    assert status == 200
+    # No stored conversation record: the private check never blocks.
+    status, body = _approval(router, "conv_0001abcd", "network")
+    assert status == 201 and body["status"] == "approved"