From aeef0f848485f800542ce91329fa783a628d26ca Mon Sep 17 00:00:00 2001 From: jenkins Date: Mon, 24 Aug 2026 00:37:09 -0300 Subject: [PATCH] hermes(hux): worker-side hook library for approvals, gates, budgets, stop receipts and events Stdlib client the agent runtime calls around its tool loop; fails closed for side effects, fails open for telemetry, never carries raw arguments or outputs. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RNPhwu2bsaRNg3DETSAZoM --- .../hermes-worker-hux/hux_hook/NOTES.md | 83 ++++++ .../hermes-worker-hux/hux_hook/__init__.py | 24 ++ .../hermes-worker-hux/hux_hook/client.py | 178 ++++++++++++ .../hermes-worker-hux/hux_hook/hooks.py | 191 +++++++++++++ .../tests/test_hermes_hux_contract_hook.py | 254 ++++++++++++++++++ testing/tests/test_hermes_hux_policy_hook.py | 236 ++++++++++++++++ 6 files changed, 966 insertions(+) create mode 100644 dockerfiles/hermes-worker-hux/hux_hook/NOTES.md create mode 100644 dockerfiles/hermes-worker-hux/hux_hook/__init__.py create mode 100644 dockerfiles/hermes-worker-hux/hux_hook/client.py create mode 100644 dockerfiles/hermes-worker-hux/hux_hook/hooks.py create mode 100644 testing/tests/test_hermes_hux_contract_hook.py create mode 100644 testing/tests/test_hermes_hux_policy_hook.py diff --git a/dockerfiles/hermes-worker-hux/hux_hook/NOTES.md b/dockerfiles/hermes-worker-hux/hux_hook/NOTES.md new file mode 100644 index 00000000..4a4ccdf5 --- /dev/null +++ b/dockerfiles/hermes-worker-hux/hux_hook/NOTES.md @@ -0,0 +1,83 @@ +# hux_hook: wiring notes for the agent runtime patch + +`hux_hook` is a stdlib-only library the Hermes agent process imports. HUX +(`hermes-hux-foundation`, loopback `127.0.0.1:8790` in the same pod) is the +source of truth; the agent never decides an approval, never reads the tenant +ledger directly and never persists raw tool arguments or output. + +## Environment the agent process needs + +| Variable | Purpose | +| --- | --- | +| `HUX_BASE_URL` | default `http://127.0.0.1:8790`; the service is loopback-only | +| `HUX_TENANT_SLOT` | `slot-N`, same value the service was started with (`HUX_TENANT_SLOT` on the service side) | +| `HUX_SUBJECT` | `usr_` of the slot owner; the router derives it, the pod env carries it | +| `HUX_WORKER_KEY` | shared key mounted read-only under `/runtime-access`; sent as `X-Hux-Relay-Key` | +| `HUX_SURFACE` / `HUX_TRUST` | default `worker` / `worker`; the agent hook is never a human surface (SO-35) | +| `HUX_TIMEOUT_SECONDS` | default 5 | + +Headers the client sends (exactly `hux/identity.py`): `X-Hermes-Tenant-Identity`, +`X-Hux-Subject`, `X-Hux-Surface`, `X-Hux-Trust`, `X-Hux-Relay-Key`, plus +`Idempotency-Key` on creates and `If-Match` on revisioned PUTs. + +## Construction + +```python +from hux_hook import HuxClient +client = HuxClient(os.environ.get("HUX_BASE_URL", "http://127.0.0.1:8790"), + {"tenant_slot": os.environ["HUX_TENANT_SLOT"], "subject": os.environ["HUX_SUBJECT"], + "surface": "worker", "trust": "worker"}, + key=os.environ.get("HUX_WORKER_KEY"), timeout=float(os.environ.get("HUX_TIMEOUT_SECONDS", "5"))) +``` + +One client per process. `client.capabilities()` is cached; call +`client.forget_capabilities()` on SIGHUP or when the WebUI reports a flag change. + +## Where each call goes in the tool loop + +1. Run start: `emit(client, conversation_id, "run.started", "...", run_id=run_id)`. +2. Immediately before executing any tool that is not read-only: + `d = before_tool(client, run_id, conversation_id, tool_name, arguments, capability, external, risk)`. + - `d.proceed is True`: execute now, with exactly the `arguments` object that was hashed. + Re-serialising or "normalising" arguments after the gate breaks SO-37. + - `d.reason == "approval_required"`: do not execute. Surface `d.approval_id` to the UI + (chat posts `POST /hux/v1/approvals/{id}` with `once|session|always|deny` from a human surface), + park the turn, and call `before_tool` again with the same arguments after the decision. + The idempotent replay reaches the gate; a `once` approval releases exactly once. + - Any other reason (`approval_denied`, `budget_exhausted`, `hux_unavailable`, `autonomy_off`, + `service_error:*`, a gate reason): refuse the tool and tell the model why. Never fall back to + the upstream gateway approval prompt while `hux.autonomy` is on. + Capability mapping is the runtime's job: `read_files`, `write_files`, `shell`, `network`, + `web_search`, `send_message`, `memory_write`, `artifact_write`, `spend_tokens`, `delegate`, + `deploy`, `external_side_effect` (see `hux/rules.py`). `external=True` for anything that + leaves the tenant (messages, network, deploy). +3. After every tool: `after_tool(client, run_id, conversation_id, tool_name, ok, bytes_out, turn, + argument_hash=canonical_argument_hash(tool_name, arguments), duration_ms=..., exit_code=...)` + then `record_spend(client, run_id, conversation_id, tool_calls=1, tokens=)`. + Both are best-effort: they return `None` on failure and never raise. +4. Delegation: `record_spend(..., delegations=1)` / `subagents=1` when spawning; the child run + uses its own `run_id` and the same `conversation_id`. +5. Memory: before proposing a memory write call `memory_gate(client, conversation_id)`; + `False` means do not even propose. The write itself still goes through `POST /hux/v1/memory`, + which enforces forget, disable and topic rules server-side. +6. Stop: when the user cancels, kill the tool processes, then + `receipt = on_stop(client, run_id, conversation_id, process_registry_empty=, + side_effects=[{"description": ..., "reverted": bool}, ...])`. `None` means no receipt exists and + the stop is NOT done: retry, and never report "cancelled" to the user without a receipt (SO-41). + `process_registry_empty` must come from the gateway's process registry, not from a timer. +7. Run end: `emit(..., "run.completed" | "run.failed", ...)`. + +## What must never happen + +- Raw arguments, file contents, command output, prompts or secrets in any `summary`, `detail` + or `evidence`. The library only ever sends tool name, capability, hash, byte counts, status. +- Deciding an approval with the worker identity (the service answers 403; do not retry as `chat`). +- Executing a tool after `proceed=False` for any reason, including HUX being unreachable. +- Logging `HuxServiceError` bodies with request payloads: exceptions carry status/code/message only. + +## Routes used + +`GET /hux/v1/capabilities`, `POST /hux/v1/approvals`, `POST /hux/v1/runs/{id}/gate`, +`POST /hux/v1/runs/{id}/budget`, `POST /hux/v1/runs/{id}/stop`, +`POST /hux/v1/conversations/{id}/events`, `GET /hux/v1/privacy/policy`, +`GET /hux/v1/conversations/{id}`. diff --git a/dockerfiles/hermes-worker-hux/hux_hook/__init__.py b/dockerfiles/hermes-worker-hux/hux_hook/__init__.py new file mode 100644 index 00000000..4a1ec534 --- /dev/null +++ b/dockerfiles/hermes-worker-hux/hux_hook/__init__.py @@ -0,0 +1,24 @@ +"""Agent-side hook library for the per-tenant HUX service. + +Stdlib only. The Hermes agent process (tenant pod or Worker) calls these +functions around its tool loop so that HUX, not the agent, decides approvals, +gates, budgets, stop receipts and what lands on the activity timeline. +Side effects fail closed when the service is unreachable; telemetry fails open. +""" + +from hux_hook.client import HuxClient, HuxServiceError, HuxUnavailable +from hux_hook.hooks import ( + Decision, + after_tool, + before_tool, + canonical_argument_hash, + emit, + memory_gate, + on_stop, + record_spend, +) + +__all__ = [ + "Decision", "HuxClient", "HuxServiceError", "HuxUnavailable", "after_tool", "before_tool", + "canonical_argument_hash", "emit", "memory_gate", "on_stop", "record_spend", +] diff --git a/dockerfiles/hermes-worker-hux/hux_hook/client.py b/dockerfiles/hermes-worker-hux/hux_hook/client.py new file mode 100644 index 00000000..a54b9e6d --- /dev/null +++ b/dockerfiles/hermes-worker-hux/hux_hook/client.py @@ -0,0 +1,178 @@ +"""Thin urllib client for the HUX service on the pod loopback. + +The client owns three things: the identity headers ``hux.identity`` expects, +the mapping of ``hux.error.v1`` bodies to ``HuxServiceError``, and the +per-process capabilities cache. It never logs or embeds request bodies in +exceptions (SO-07, SO-11): an error carries status, code and the service's +own message only. +""" + +from __future__ import annotations + +import http.client +import json +import threading +import urllib.error +import urllib.request +from collections.abc import Mapping +from typing import Any + +HEADER_SLOT = "X-Hermes-Tenant-Identity" +HEADER_SUBJECT = "X-Hux-Subject" +HEADER_SURFACE = "X-Hux-Surface" +HEADER_TRUST = "X-Hux-Trust" +HEADER_KEY = "X-Hux-Relay-Key" +DEFAULT_BASE_URL = "http://127.0.0.1:8790" +MESSAGE_MAX = 280 + + +class HuxServiceError(Exception): + """The service answered with a ``hux.error.v1`` body (or a non-JSON failure).""" + + def __init__(self, status: int, code: str, message: str) -> None: + super().__init__(f"{status} {code}: {message}") + self.status = int(status) + self.code = str(code) + self.message = str(message)[:MESSAGE_MAX] + + +class HuxUnavailable(HuxServiceError): + """The service could not be reached at all; callers decide open or closed.""" + + def __init__(self, message: str = "hux service unreachable") -> None: + super().__init__(0, "unavailable", message) + + +class HuxResponse: + """Status, parsed JSON body and headers of one answer.""" + + def __init__(self, status: int, body: Any, headers: Mapping[str, str]) -> None: + self.status = status + self.body = body + self.headers = {k.lower(): v for k, v in headers.items()} + + def header(self, name: str) -> str: + """Case-insensitive header lookup; empty when absent.""" + return self.headers.get(name.lower(), "") + + +def _error_from(status: int, body: Any) -> HuxServiceError: + if isinstance(body, dict) and body.get("schema") == "hux.error.v1": + return HuxServiceError(int(body.get("status", status)), str(body.get("code", "invalid")), str(body.get("message", ""))) + return HuxServiceError(status, "invalid", f"unexpected response {status}") + + +class HuxClient: + """One tenant identity talking to one HUX base URL. + + ``identity`` carries ``tenant_slot``, ``subject``, ``surface`` and + ``trust``; ``key`` is the relay or worker shared key when ``trust`` needs + one. The agent hook normally runs as ``surface=worker, trust=worker``. + """ + + def __init__(self, base_url: str = DEFAULT_BASE_URL, identity: Mapping[str, str] | None = None, + key: str | None = None, timeout: float = 5) -> None: + identity = dict(identity or {}) + self.base_url = base_url.rstrip("/") + self.identity = { + "tenant_slot": str(identity.get("tenant_slot", "")), "subject": str(identity.get("subject", "")), + "surface": str(identity.get("surface", "worker")), "trust": str(identity.get("trust", "worker")), + } + self._key = key + self.timeout = timeout + self._capabilities: dict[str, Any] | None = None + self._guard = threading.Lock() + + # -- transport ------------------------------------------------------------- + + def headers(self, extra: Mapping[str, str] | None = None) -> dict[str, str]: + """Identity headers exactly as ``hux.identity.resolve`` reads them, plus ``extra``.""" + out = { + HEADER_SLOT: self.identity["tenant_slot"], HEADER_SUBJECT: self.identity["subject"], + HEADER_SURFACE: self.identity["surface"], HEADER_TRUST: self.identity["trust"], + "Accept": "application/json", + } + if self._key: + out[HEADER_KEY] = self._key + for name, value in (extra or {}).items(): + if value: + out[name] = str(value) + return out + + def request(self, method: str, path: str, body: Any = None, *, idempotency_key: str | None = None, + if_match: int | str | None = None, query: Mapping[str, str] | None = None) -> HuxResponse: + """Send one request; raise ``HuxServiceError`` on 4xx/5xx and ``HuxUnavailable`` on transport failure.""" + extra: dict[str, str] = {} + if idempotency_key: + extra["Idempotency-Key"] = idempotency_key + if if_match is not None: + extra["If-Match"] = str(if_match) + data = None + if body is not None: + data = json.dumps(body, sort_keys=True, separators=(",", ":")).encode() + extra["Content-Type"] = "application/json" + url = self.base_url + path + if query: + url += "?" + "&".join(f"{k}={urllib.request.quote(str(v), safe='')}" for k, v in query.items()) + req = urllib.request.Request(url, data=data, method=method, headers=self.headers(extra)) + try: + with urllib.request.urlopen(req, timeout=self.timeout) as raw: # noqa: S310 - loopback only + response = HuxResponse(raw.status, _decode(raw.read()), dict(raw.headers.items())) + except urllib.error.HTTPError as error: + payload = _decode(error.read()) + raise _error_from(error.code, payload) from None + except (urllib.error.URLError, OSError, TimeoutError) as error: + raise HuxUnavailable(f"hux service unreachable: {type(error).__name__}") from None + except http.client.InvalidURL: + # http.client refuses control characters and spaces in the path; treat it as our own bad request. + raise HuxServiceError(400, "invalid", "malformed request path") from None + except http.client.HTTPException as error: + raise HuxUnavailable(f"hux service unreachable: {type(error).__name__}") from None + if response.status >= 400: + raise _error_from(response.status, response.body) + return response + + def get(self, path: str, query: Mapping[str, str] | None = None) -> HuxResponse: + """``GET path``.""" + return self.request("GET", path, query=query) + + def post(self, path: str, body: Any, idempotency_key: str | None = None) -> HuxResponse: + """``POST path`` with an optional ``Idempotency-Key``.""" + return self.request("POST", path, body, idempotency_key=idempotency_key) + + def put(self, path: str, body: Any, if_match: int | str | None = None) -> HuxResponse: + """``PUT path`` with an optional ``If-Match`` revision.""" + return self.request("PUT", path, body, if_match=if_match) + + # -- capabilities ------------------------------------------------------------ + + def capabilities(self, refresh: bool = False) -> dict[str, Any]: + """Which cards are on, cached per process; unreachable or off means every card reads as off.""" + with self._guard: + if self._capabilities is not None and not refresh: + return self._capabilities + try: + body = self.get("/hux/v1/capabilities").body + except HuxServiceError as error: + return {"reachable": error.code != "unavailable", "cards": {}, "contract_version": ""} + cards = {c["card"]: bool(c.get("enabled")) for c in body.get("cards", []) if isinstance(c, dict) and "card" in c} + self._capabilities = {"reachable": True, "cards": cards, "contract_version": str(body.get("contract_version", ""))} + return self._capabilities + + def card_enabled(self, card: str) -> bool: + """True only when the service answered and reports ``card`` on.""" + return bool(self.capabilities()["cards"].get(card)) + + def forget_capabilities(self) -> None: + """Drop the cache so the next call re-reads flags (used after a reload signal).""" + with self._guard: + self._capabilities = None + + +def _decode(raw: bytes) -> Any: + if not raw: + return None + try: + return json.loads(raw) + except ValueError: + return None diff --git a/dockerfiles/hermes-worker-hux/hux_hook/hooks.py b/dockerfiles/hermes-worker-hux/hux_hook/hooks.py new file mode 100644 index 00000000..659fb33b --- /dev/null +++ b/dockerfiles/hermes-worker-hux/hux_hook/hooks.py @@ -0,0 +1,191 @@ +"""Pure hook functions the agent calls around its tool loop. + +Every function takes a ``HuxClient`` and never touches the agent's own state. +``before_tool`` is the only call whose answer the agent must obey: it asks for +an approval, then asks the gate, and fails closed on any doubt (SO-37, SO-39, +SO-40). ``after_tool``, ``record_spend`` and ``emit`` are telemetry: they never +raise into the loop and never carry raw arguments or tool output (SO-11). +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from typing import Any + +from hux_hook.client import HuxClient, HuxServiceError + +CARD_EVENTS = "HUX-01" +CARD_AUTONOMY = "HUX-05" +CARD_PRIVACY = "HUX-10" +SPEND_KEYS = ("tokens", "tool_calls", "wall_clock_seconds", "delegations", "spend_units", "subagents") +RISKS = ("low", "medium", "high") +PATH_SEGMENT = re.compile(r"^[A-Za-z0-9._:-]{1,120}$") +KEY_CHARS = re.compile(r"[^A-Za-z0-9._:-]") + + +@dataclass(frozen=True) +class Decision: + """What the agent may do next; ``proceed`` is False unless the gate released the call.""" + + proceed: bool + approval_id: str | None + reason: str + + +def canonical_json(value: Any) -> bytes: + """Canonical JSON: sorted keys, no whitespace, ASCII escapes; key order and formatting cannot change it.""" + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() + + +def canonical_argument_hash(tool_name: str, arguments: Any) -> str: + """``sha256:`` over the canonical JSON of ``{"tool": name, "arguments": ...}``; the gate compares this exact value.""" + return "sha256:" + hashlib.sha256(canonical_json({"tool": tool_name, "arguments": arguments})).hexdigest() + + +def call_ref(tool_name: str, argument_hash: str) -> str: + """Evidence id for one tool call: name plus a hash prefix, never the arguments.""" + return f"{KEY_CHARS.sub('-', tool_name)[:40]}:{argument_hash[7:23]}" + + +def idempotency_key(run_id: str, kind: str, token: str) -> str: + """A key that satisfies the contract pattern whatever the run id looks like.""" + key = KEY_CHARS.sub("-", f"{run_id}:{kind}:{token}") + return key[:120] if len(key) >= 8 else (key + ".pad").ljust(8, "0") + + +def _safe_summary(tool_name: str, capability: str, argument_hash: str, size: int) -> str: + return f"{tool_name} ({capability}) arguments {argument_hash[:23]} {size} bytes"[:280] + + +def before_tool(client: HuxClient, run_id: str, conversation_id: str, tool_name: str, arguments: Any, capability: str, + external: bool = False, risk: str = "medium", turn: int | None = None) -> Decision: + """Request (or replay) the approval for this exact call, then ask the gate. Fails closed. + + A pending approval returns ``proceed=False, reason="approval_required"`` with + the approval id so the UI can prompt; the agent calls again with the same + arguments once the human has decided and the idempotent replay reaches the gate. + """ + if not PATH_SEGMENT.match(str(run_id)): + return Decision(False, None, "invalid_run_id") + if not client.card_enabled(CARD_AUTONOMY): + return Decision(False, None, "hux_unavailable" if not client.capabilities()["reachable"] else "autonomy_off") + argument_hash = canonical_argument_hash(tool_name, arguments) + size = len(canonical_json(arguments)) + body = { + "run_id": run_id, "conversation_id": conversation_id, "capability": capability, + "request": { + "summary": _safe_summary(tool_name, capability, argument_hash, size), + "risk": risk if risk in RISKS else "high", "external": bool(external), + "evidence": [{"kind": "tool_call", "id": call_ref(tool_name, argument_hash), "hash": argument_hash}], + }, + } + try: + approval = client.post("/hux/v1/approvals", body, idempotency_key(run_id, "approval", argument_hash[7:39])).body + except HuxServiceError as error: + return Decision(False, None, _failure_reason(error)) + approval_id = approval.get("id") if isinstance(approval, dict) else None + status = approval.get("status") if isinstance(approval, dict) else None + if status == "pending": + return Decision(False, approval_id, "approval_required") + if status != "approved": + return Decision(False, approval_id, f"approval_{status or 'invalid'}") + gate_body = {"capability": capability, "argument_hash": argument_hash, "external": bool(external), "conversation_id": conversation_id} + try: + verdict = client.post(f"/hux/v1/runs/{run_id}/gate", gate_body).body + except HuxServiceError as error: + return Decision(False, approval_id, _failure_reason(error)) + if not isinstance(verdict, dict) or verdict.get("proceed") is not True: + reason = verdict.get("reason", "gate_blocked") if isinstance(verdict, dict) else "gate_invalid" + return Decision(False, approval_id, str(reason)[:280]) + return Decision(True, str(verdict.get("approval_id") or approval_id), "released") + + +def _failure_reason(error: HuxServiceError) -> str: + if error.code in {"unavailable", "budget_exhausted", "flag_off", "approval_required"}: + return "hux_unavailable" if error.code == "unavailable" else error.code + return f"service_error:{error.code}" + + +def emit(client: HuxClient, conversation_id: str, kind: str, summary: str, detail: dict[str, Any] | None = None, + evidence: list[dict[str, Any]] | None = None, sensitivity: str = "personal", run_id: str | None = None, + turn: int | None = None, correlation_id: str | None = None, idempotency_key: str | None = None) -> dict[str, Any] | None: + """Best-effort activity event; returns the stored record, or None when it could not be recorded.""" + body: dict[str, Any] = {"kind": kind, "summary": str(summary)[:280], "sensitivity": sensitivity} + if detail: + body["detail"] = detail + if evidence: + body["evidence"] = evidence + if run_id: + body["run_id"] = run_id + if turn is not None: + body["turn"] = int(turn) + if correlation_id: + body["correlation_id"] = correlation_id + try: + response = client.post(f"/hux/v1/conversations/{conversation_id}/events", body, idempotency_key) + except (HuxServiceError, TypeError, ValueError): + return None + return response.body if isinstance(response.body, dict) else None + + +def after_tool(client: HuxClient, run_id: str, conversation_id: str, tool_name: str, ok: bool, bytes_out: int, + turn: int | None = None, argument_hash: str | None = None, duration_ms: int | None = None, + exit_code: int | None = None) -> dict[str, Any] | None: + """Emit ``tool.result`` with status and sizes only; the output itself never leaves the agent.""" + detail: dict[str, Any] = {"tool": tool_name, "ok": bool(ok), "bytes": max(0, int(bytes_out))} + if duration_ms is not None: + detail["duration_ms"] = max(0, int(duration_ms)) + if exit_code is not None: + detail["exit_code"] = int(exit_code) + evidence = [{"kind": "tool_call", "id": call_ref(tool_name, argument_hash), "hash": argument_hash}] if argument_hash else None + summary = f"{tool_name} {'succeeded' if ok else 'failed'} ({detail['bytes']} bytes)" + key = idempotency_key(run_id, "result", f"{turn or 0}:{(argument_hash or 'none')[7:23]}") + return emit(client, conversation_id, "tool.result", summary, detail, evidence, run_id=run_id, turn=turn, idempotency_key=key) + + +def record_spend(client: HuxClient, run_id: str, conversation_id: str | None = None, **increments: int) -> dict[str, Any] | None: + """Add spend to the run budget; returns the new ``hux.budget_state.v1`` or None. Never raises.""" + body: dict[str, Any] = {k: max(0, int(v)) for k, v in increments.items() if k in SPEND_KEYS} + if conversation_id: + body["conversation_id"] = conversation_id + try: + response = client.post(f"/hux/v1/runs/{run_id}/budget", body) + except (HuxServiceError, TypeError, ValueError): + return None + return response.body if isinstance(response.body, dict) else None + + +def on_stop(client: HuxClient, run_id: str, conversation_id: str | None, process_registry_empty: bool, + side_effects: list[dict[str, Any]] | None = None, already_complete: bool = False) -> dict[str, Any] | None: + """Write the cancellation receipt (SO-41). None means no receipt exists and the stop is not done.""" + body: dict[str, Any] = {"process_registry_empty": bool(process_registry_empty), "side_effects": list(side_effects or [])} + if already_complete: + body["already_complete"] = True + if conversation_id: + body["conversation_id"] = conversation_id + try: + response = client.post(f"/hux/v1/runs/{run_id}/stop", body) + except HuxServiceError: + return None + return response.body if isinstance(response.body, dict) else None + + +def memory_gate(client: HuxClient, conversation_id: str) -> bool: + """May the agent propose a memory write from this conversation? Conservative False when in doubt. + + Requires the privacy card to answer (SO-27 header is passed through as a + hint only); a conversation whose mode is ``private`` never writes (SO-28). + The service still enforces forget and disable state on the write itself. + """ + try: + client.get("/hux/v1/privacy/policy") + except HuxServiceError: + return False + try: + record = client.get(f"/hux/v1/conversations/{conversation_id}").body + except HuxServiceError as error: + return error.status == 404 + return not (isinstance(record, dict) and record.get("mode") == "private") diff --git a/testing/tests/test_hermes_hux_contract_hook.py b/testing/tests/test_hermes_hux_contract_hook.py new file mode 100644 index 00000000..e5cf8a3f --- /dev/null +++ b/testing/tests/test_hermes_hux_contract_hook.py @@ -0,0 +1,254 @@ +"""HuxClient transport contract and the pure helpers of the agent hook. + +Security obligations exercised: SO-04 (the client emits exactly the identity +header vocabulary ``hux.identity`` accepts, with the worker key), SO-07 (errors +carry status, code and message only; request bodies are never embedded), +SO-27 and SO-28 (memory writes need the privacy card and are refused in a +private conversation), SO-37 (the argument hash is stable across key order +and whitespace), SO-50 (capabilities are read from the resolved flag chain). +""" + +from __future__ import annotations + +import ast +import http.client +import io +import sys +import threading +import urllib.error +import urllib.request +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation" +HOOK_ROOT = ROOT / "dockerfiles" / "hermes-worker-hux" +for entry in (FOUNDATION, HOOK_ROOT): + if str(entry) not in sys.path: + sys.path.insert(0, str(entry)) + +from hux import contracts, identity # noqa: E402 +from hux.http import serve # noqa: E402 +from hux.server import build_router # noqa: E402 +from hux_hook import HuxClient, HuxServiceError, HuxUnavailable, canonical_argument_hash, emit, memory_gate # noqa: E402 +from hux_hook import client as client_mod # noqa: E402 +from hux_hook import hooks # noqa: E402 + +SCHEMAS = contracts.load_all() +ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) +SUBJECT = "usr_0123456789abcdef" +WORKER = {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "worker", "trust": "worker"} +HUMAN = {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "chat", "trust": "router"} +OTHER = {**HUMAN, "subject": "usr_fedcba9876543210"} +CANARY = "CANARY-9c1d-SECRET" + + +def start(tmp_path: Path, flags: str = ALL_ON): + router = build_router(tmp_path, {"HUX_FLAGS": flags, "HUX_WORKER_KEY": "wk"}) + server = serve(router, "127.0.0.1", 0) + threading.Thread(target=server.serve_forever, daemon=True).start() + return f"http://127.0.0.1:{server.server_address[1]}", server + + +@pytest.fixture +def live(tmp_path): + base, server = start(tmp_path) + yield base, tmp_path + server.shutdown() + + +# --- headers and identity --------------------------------------------------------- + +def test_headers_match_the_service_vocabulary(): + """SO-04: the exact header names identity.resolve reads, key only when present, extras only when non-empty.""" + worker = HuxClient("http://127.0.0.1:1/", WORKER, key="wk") + sent = worker.headers({"Idempotency-Key": "run:approval:1", "If-Match": ""}) + assert identity.resolve(sent, {"HUX_WORKER_KEY": "wk"}) == identity.Identity("slot-3", SUBJECT, "worker", "worker") + assert sent["Idempotency-Key"] == "run:approval:1" and "If-Match" not in sent + assert worker.base_url == "http://127.0.0.1:1" + plain = HuxClient(identity={"tenant_slot": "slot-3", "subject": SUBJECT}).headers() + assert client_mod.HEADER_KEY not in plain and plain[client_mod.HEADER_TRUST] == "worker" + assert HuxClient().identity == {"tenant_slot": "", "subject": "", "surface": "worker", "trust": "worker"} + + +def test_error_mapping_and_no_body_leak(live): + """SO-07: a hux.error.v1 answer becomes HuxServiceError(status, code, message) and the body never appears in it.""" + base, _ = live + human = HuxClient(base, HUMAN) + with pytest.raises(HuxServiceError) as bad: + human.post("/hux/v1/approvals", {"conversation_id": "conv_0001abcd", "capability": "nope", "secret": CANARY}) + assert (bad.value.status, bad.value.code) == (400, "invalid") and CANARY not in str(bad.value) + with pytest.raises(HuxServiceError) as unauth: + HuxClient(base, WORKER, key="wrong").get("/hux/v1/capabilities") + assert unauth.value.code == "unauthorized" + with pytest.raises(HuxServiceError) as missing: + human.get("/hux/v1/no/such/route") + assert (missing.value.status, missing.value.code) == (404, "not_found") + assert HuxUnavailable().code == "unavailable" and HuxUnavailable().status == 0 + assert HuxServiceError(500, "x", "m" * 400).message == "m" * 280 + + +def test_transport_edge_cases(monkeypatch): + """Non-JSON failures, a 4xx delivered without an exception and socket errors map to typed errors.""" + client = HuxClient("http://127.0.0.1:1", WORKER, key="wk") + + def raise_http(*args, **kwargs): + raise urllib.error.HTTPError("u", 502, "bad gateway", {}, io.BytesIO(b"")) + + monkeypatch.setattr(urllib.request, "urlopen", raise_http) + with pytest.raises(HuxServiceError) as html: + client.get("/hux/v1/capabilities") + assert (html.value.status, html.value.code) == (502, "invalid") + + class Raw: + status = 418 + headers = {"X-Test": "1"} + + def read(self): + return b"{bad json" + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **k: Raw()) + with pytest.raises(HuxServiceError) as teapot: + client.put("/hux/v1/policy", {"x": 1}, if_match=3) + assert teapot.value.status == 418 + + def raise_socket(*args, **kwargs): + raise TimeoutError("slow") + + monkeypatch.setattr(urllib.request, "urlopen", raise_socket) + with pytest.raises(HuxUnavailable): + client.get("/hux/v1/capabilities", query={"a": "b c"}) + + def raise_half_closed(*args, **kwargs): + raise http.client.BadStatusLine("gone") + + monkeypatch.setattr(urllib.request, "urlopen", raise_half_closed) + with pytest.raises(HuxUnavailable): + client.get("/hux/v1/capabilities") + + +def test_put_with_if_match_and_get_with_query(live): + """Revisioned writes send If-Match; a stale revision is a conflict; queries reach the service.""" + base, _ = live + human = HuxClient(base, HUMAN) + first = human.put("/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "safe"}) + assert first.header("ETag") == "1" and first.header("Missing") == "" + second = human.put("/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"}, if_match=1) + assert second.body["revision"] == 2 + with pytest.raises(HuxServiceError) as stale: + human.put("/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "safe"}, if_match=1) + assert stale.value.code == "conflict" + assert human.get("/hux/v1/policy", {"scope": "global"}).body["autonomy"] == "autonomous" + + +# --- capabilities --------------------------------------------------------------------- + +def test_capabilities_cached_per_process(live, monkeypatch): + """SO-50: one capabilities read per process; refresh and forget re-read; failures are not cached.""" + base, _ = live + worker = HuxClient(base, WORKER, key="wk") + calls = [] + real_get = worker.get + monkeypatch.setattr(worker, "get", lambda path, query=None: calls.append(path) or real_get(path, query)) + assert worker.card_enabled("HUX-05") and worker.card_enabled("HUX-01") + assert worker.capabilities()["contract_version"] == contracts.load_flags()["contract_version"] + assert len(calls) == 1 + worker.capabilities(refresh=True) + worker.forget_capabilities() + worker.capabilities() + assert len(calls) == 3 + monkeypatch.setattr(worker, "get", lambda path, query=None: (_ for _ in ()).throw(HuxServiceError(404, "flag_off", "off"))) + worker.forget_capabilities() + assert worker.capabilities() == {"reachable": True, "cards": {}, "contract_version": ""} + dead = HuxClient("http://127.0.0.1:1", WORKER, key="wk", timeout=1) + assert dead.capabilities()["reachable"] is False and not dead.card_enabled("HUX-11") + + +# --- events --------------------------------------------------------------------------- + +def test_emit_is_best_effort_and_redaction_safe(live): + """SO-11: detail outside the allowlist is dropped by the service; failures return None and never raise.""" + base, root = live + human = HuxClient(base, HUMAN) + conv = human.post("/hux/v1/conversations", {"title": "t"}).body["id"] + worker = HuxClient(base, WORKER, key="wk") + record = emit(worker, conv, "tool.call", "shell call", {"tool": "shell", "arguments": {"cmd": CANARY}, "argument_bytes": 7}, + evidence=[{"kind": "run", "id": "run_1"}], run_id="run_1", turn=2, correlation_id="corr-1", idempotency_key="run_1:call:0001") + assert contracts.validate_record(record, SCHEMAS) == [] and record["detail"] == {"tool": "shell", "argument_bytes": 7} + assert record["provenance"]["actor"] == {"type": "system", "id": "hux-worker"} and record["correlation_id"] == "corr-1" + assert emit(worker, conv, "tool.call", "again", idempotency_key="run_1:call:0001")["id"] == record["id"] + assert emit(worker, conv, "not.a.kind", "x") is None + assert emit(worker, "conv_unknown0001", "tool.call", "x") is None + assert emit(worker, conv, "tool.call", "x", {"tool": object()}) is None + assert emit(HuxClient(base, OTHER), conv, "tool.call", "cross tenant") is None + private = human.post("/hux/v1/conversations", {"title": "p", "mode": "private"}).body["id"] + assert emit(worker, private, "tool.call", "private mode writes nothing") is None + assert CANARY not in "\n".join(p.read_text(errors="ignore") for p in root.rglob("*") if p.is_file()) + + +# --- memory gate ------------------------------------------------------------------------ + +def test_memory_gate(live, tmp_path): + """SO-27, SO-28: allowed only when the privacy card answers and the conversation is not private.""" + base, _ = live + human = HuxClient(base, HUMAN) + worker = HuxClient(base, WORKER, key="wk") + normal = human.post("/hux/v1/conversations", {"title": "n", "mode": "thoughtful"}).body["id"] + private = human.post("/hux/v1/conversations", {"title": "p", "mode": "private"}).body["id"] + assert memory_gate(worker, normal) is True + assert memory_gate(worker, private) is False + assert memory_gate(worker, "conv_notknown01") is True + assert memory_gate(worker, "bad id") is False + assert memory_gate(HuxClient("http://127.0.0.1:1", WORKER, key="wk", timeout=1), normal) is False + off_base, off_server = start(tmp_path / "off", "hux.foundation,hux.projects") + try: + assert memory_gate(HuxClient(off_base, WORKER, key="wk"), normal) is False + finally: + off_server.shutdown() + + +# --- pure helpers ------------------------------------------------------------------------- + +def test_hash_is_canonical(): + """SO-37: key order, whitespace and nesting order of dict keys do not change the hash; values do.""" + a = canonical_argument_hash("write", {"path": "n.md", "opts": {"b": 1, "a": [1, 2]}}) + b = canonical_argument_hash("write", {"opts": {"a": [1, 2], "b": 1}, "path": "n.md"}) + assert a == b and a.startswith("sha256:") and len(a) == 71 + assert canonical_argument_hash("write", {"path": "n.md ", "opts": {"b": 1, "a": [1, 2]}}) != a + assert canonical_argument_hash("other", {"path": "n.md", "opts": {"b": 1, "a": [1, 2]}}) != a + assert canonical_argument_hash("w", "raw string") == canonical_argument_hash("w", "raw string") + assert hooks.canonical_json({"z": "é", "a": None}) == b'{"a":null,"z":"\\u00e9"}' + + +def test_key_and_ref_helpers(): + """Idempotency keys always satisfy the contract pattern; call refs never contain arguments.""" + assert hooks.idempotency_key("r", "a", "") == "r:a:.pad" + assert hooks.idempotency_key("r", "a", "x") == "r:a:x.pad" + key = hooks.idempotency_key("run id/with spaces", "approval", "f" * 200) + assert len(key) == 120 and " " not in key and "/" not in key + assert hooks.call_ref("my tool/x", "sha256:" + "ab" * 32) == "my-tool-x:abababababababab" + assert hooks._failure_reason(HuxServiceError(404, "flag_off", "")) == "flag_off" + assert hooks._failure_reason(HuxServiceError(0, "unavailable", "")) == "hux_unavailable" + assert hooks._failure_reason(HuxServiceError(409, "conflict", "")) == "service_error:conflict" + + +def test_library_is_stdlib_only_small_and_documented(): + """Every module ≤ 500 lines, every public function and module documented, no third-party imports.""" + for path in sorted((HOOK_ROOT / "hux_hook").glob("*.py")): + source = path.read_text() + assert len(source.splitlines()) <= 500, path + tree = ast.parse(source) + assert ast.get_docstring(tree), path + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef | ast.ClassDef) and not node.name.startswith("_"): + assert ast.get_docstring(node), f"{path.name}:{node.name}" + if isinstance(node, ast.Import | ast.ImportFrom): + root = (node.names[0].name if isinstance(node, ast.Import) else node.module or "").split(".")[0] + assert root in {"http", "json", "threading", "urllib", "collections", "typing", "hashlib", "re", "dataclasses", "hux_hook", "__future__"}, root diff --git a/testing/tests/test_hermes_hux_policy_hook.py b/testing/tests/test_hermes_hux_policy_hook.py new file mode 100644 index 00000000..142c5545 --- /dev/null +++ b/testing/tests/test_hermes_hux_policy_hook.py @@ -0,0 +1,236 @@ +"""Agent-side HUX hook driven end-to-end against the real service over HTTP. + +Security obligations exercised: SO-35 (only the human surface decides; the +hook's worker identity cannot), SO-36 (a ``once`` approval releases the gate +exactly once), SO-37 (the gate compares the canonical argument hash the hook +recorded at request time), SO-39 (an external side effect under +``autonomous`` still waits for a human), SO-40 (an exhausted budget blocks new +approvals), SO-41 (a stop is done only when the receipt exists), SO-11 and +SO-07 (raw arguments and tool output never reach the tenant ledger). +""" + +from __future__ import annotations + +import json +import sys +import threading +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation" +HOOK_ROOT = ROOT / "dockerfiles" / "hermes-worker-hux" +for entry in (FOUNDATION, HOOK_ROOT): + if str(entry) not in sys.path: + sys.path.insert(0, str(entry)) + +from hux import contracts # noqa: E402 +from hux.http import serve # noqa: E402 +from hux.server import build_router # noqa: E402 +from hux_hook import HuxClient, HuxServiceError, after_tool, before_tool, canonical_argument_hash, on_stop, record_spend # noqa: E402 + +SCHEMAS = contracts.load_all() +ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) +SUBJECT = "usr_0123456789abcdef" +IDENTITY = {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "worker", "trust": "worker"} +HUMAN = {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "chat", "trust": "router"} +CANARY = "CANARY-7f3a-SECRET-VALUE" +ARGS = {"path": "notes.md", "content": CANARY} +RUN = "run_hook_1" + + +def start(tmp_path: Path, flags: str = ALL_ON) -> tuple[str, object]: + """Run the real service on an ephemeral loopback port for one test.""" + router = build_router(tmp_path, {"HUX_FLAGS": flags, "HUX_WORKER_KEY": "wk"}) + server = serve(router, "127.0.0.1", 0) + threading.Thread(target=server.serve_forever, daemon=True).start() + return f"http://127.0.0.1:{server.server_address[1]}", server + + +@pytest.fixture +def service(tmp_path): + base, server = start(tmp_path) + human = HuxClient(base, HUMAN) + conversation = human.post("/hux/v1/conversations", {"title": "hook test"}).body["id"] + yield {"base": base, "root": tmp_path, "agent": HuxClient(base, IDENTITY, key="wk"), "human": human, "conv": conversation} + server.shutdown() + + +def set_policy(human: HuxClient, autonomy: str, budgets: dict | None = None) -> None: + body = {"scope": {"level": "global"}, "autonomy": autonomy} + if budgets: + body["budgets"] = budgets + human.put("/hux/v1/policy", body) + + +def ledger_text(root: Path) -> str: + return "\n".join(p.read_text(errors="ignore") for p in root.rglob("*") if p.is_file()) + + +def events_of(human: HuxClient, conversation: str) -> list[dict]: + items = human.get(f"/hux/v1/conversations/{conversation}/events").body["items"] + for item in items: + assert contracts.validate_record(item, SCHEMAS) == [] + return items + + +# --- approval -> pending -> human decision -> gate once ------------------------------ + +def test_once_approval_releases_exactly_once(service): + """SO-35, SO-36, SO-37: pending until a human decides, then one release; the second gate is blocked.""" + agent, human, conv = service["agent"], service["human"], service["conv"] + first = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files", risk="low", turn=1) + assert (first.proceed, first.reason) == (False, "approval_required") + assert first.approval_id and first.approval_id.startswith("apr_") + record = human.get(f"/hux/v1/approvals/{first.approval_id}").body + assert contracts.validate_record(record, SCHEMAS) == [] + assert record["request"]["evidence"][0]["hash"] == canonical_argument_hash("write_file", ARGS) + assert CANARY not in json.dumps(record) + with pytest.raises(HuxServiceError) as denied: + agent.post(f"/hux/v1/approvals/{first.approval_id}", {"choice": "once"}) + assert denied.value.status == 403 + human.post(f"/hux/v1/approvals/{first.approval_id}", {"choice": "once"}) + second = before_tool(agent, RUN, conv, "write_file", {"content": CANARY, "path": "notes.md"}, "write_files", risk="low") + assert second == second.__class__(True, first.approval_id, "released") + third = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files", risk="low") + assert third.proceed is False and "consumed" in third.reason + kinds = [e["kind"] for e in events_of(human, conv)] + assert kinds.count("side_effect.released") == 1 and "side_effect.blocked" in kinds and "approval.requested" in kinds + assert CANARY not in ledger_text(service["root"]) + + +def test_different_arguments_after_approval_are_blocked(service): + """SO-37: an approval for one argument hash never releases another.""" + agent, human, conv = service["agent"], service["human"], service["conv"] + pending = before_tool(agent, RUN, conv, "shell", {"cmd": "ls"}, "shell") + human.post(f"/hux/v1/approvals/{pending.approval_id}", {"choice": "once"}) + other = before_tool(agent, RUN, conv, "shell", {"cmd": "rm -rf /"}, "shell") + assert other.proceed is False and other.reason == "approval_required" + + +def test_denied_and_session_choices(service): + """A denial fails closed; a session grant lets a later call in the same conversation auto-approve.""" + agent, human, conv = service["agent"], service["human"], service["conv"] + pending = before_tool(agent, RUN, conv, "shell", {"cmd": "ls"}, "shell") + human.post(f"/hux/v1/approvals/{pending.approval_id}", {"choice": "deny"}) + assert before_tool(agent, RUN, conv, "shell", {"cmd": "ls"}, "shell") == pending.__class__(False, pending.approval_id, "approval_denied") + pending2 = before_tool(agent, RUN, conv, "shell", {"cmd": "pwd"}, "shell") + human.post(f"/hux/v1/approvals/{pending2.approval_id}", {"choice": "session"}) + assert before_tool(agent, "run_hook_2", conv, "shell", {"cmd": "whoami"}, "shell").proceed is True + + +def test_external_side_effect_under_autonomous_still_asks(service): + """SO-39: autonomy never auto-allows an external side effect.""" + agent, human, conv = service["agent"], service["human"], service["conv"] + set_policy(human, "autonomous") + decision = before_tool(agent, RUN, conv, "send_email", {"to": "x@example.com"}, "send_message", external=True, risk="high") + assert (decision.proceed, decision.reason) == (False, "approval_required") + local = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files") + assert local.proceed is True and local.reason == "released" + assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").proceed is False + + +def test_budget_exhaustion_blocks_new_approvals(service): + """SO-40: once the run budget is spent the hook reports budget_exhausted and never proceeds.""" + agent, human, conv = service["agent"], service["human"], service["conv"] + set_policy(human, "autonomous", {"tool_calls_per_run": 2}) + state = record_spend(agent, RUN, conv, tool_calls=2, tokens=10, bogus=5) + assert state is not None and contracts.validate_record(state, SCHEMAS) == [] + assert state["exhausted"] == ["tool_calls_per_run"] + decision = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files") + assert decision == decision.__class__(False, None, "budget_exhausted") + assert record_spend(agent, RUN, conv, tokens=-3)["spent"]["tokens"] == 10 + assert record_spend(agent, RUN, tool_calls=1)["spent"]["tool_calls"] == 3 + + +def test_stop_receipt_is_written_once(service): + """SO-41: the receipt reports what really happened and a repeat returns the same record.""" + agent, human, conv = service["agent"], service["human"], service["conv"] + effects = [{"description": "partial notes.md", "reverted": True}] + receipt = on_stop(agent, RUN, conv, process_registry_empty=True, side_effects=effects) + assert contracts.validate_record(receipt, SCHEMAS) == [] + assert receipt["outcome"] == "cancelled" and receipt["side_effects"] == effects + assert on_stop(agent, RUN, conv, process_registry_empty=False)["id"] == receipt["id"] + failed = on_stop(agent, "run_hook_3", None, process_registry_empty=False) + assert failed["outcome"] == "failed_to_cancel" and "conversation_id" not in failed + assert on_stop(agent, "run_hook_4", conv, True, already_complete=True)["outcome"] == "already_complete" + assert [e["kind"] for e in events_of(human, conv)].count("run.cancelled") == 2 + + +def test_after_tool_records_status_only(service): + """SO-11: tool.result carries sizes and status; the output and arguments never land.""" + agent, conv = service["agent"], service["conv"] + digest = canonical_argument_hash("shell", {"cmd": CANARY}) + event = after_tool(agent, RUN, conv, "shell", True, 512, turn=3, argument_hash=digest, duration_ms=40, exit_code=0) + assert event["kind"] == "tool.result" and event["detail"] == {"tool": "shell", "ok": True, "bytes": 512, "duration_ms": 40, "exit_code": 0} + assert event["evidence"][0]["hash"] == digest and event["turn"] == 3 + replay = after_tool(agent, RUN, conv, "shell", True, 512, turn=3, argument_hash=digest) + assert replay["id"] == event["id"] + plain = after_tool(agent, RUN, conv, "shell", False, -1) + assert plain["detail"] == {"tool": "shell", "ok": False, "bytes": 0} and "evidence" not in plain + assert CANARY not in ledger_text(service["root"]) + + +def test_bad_ids_and_service_errors_fail_closed(service): + """Malformed run or conversation ids never reach a tool execution.""" + agent, conv = service["agent"], service["conv"] + assert before_tool(agent, "run/../x", conv, "shell", {}, "shell").reason == "invalid_run_id" + assert before_tool(agent, RUN, "not an id", "shell", {}, "shell").reason == "service_error:invalid" + assert before_tool(agent, RUN, conv, "shell", {}, "teleport").reason == "service_error:invalid" + + +def test_gate_failures_fail_closed(service, monkeypatch): + """A gate that errors, answers nonsense or says no keeps the tool from running.""" + agent, human, conv = service["agent"], service["human"], service["conv"] + set_policy(human, "autonomous") + real_post = agent.post + + def broken_gate(path, body, idempotency_key=None): + if path.endswith("/gate"): + raise HuxServiceError(500, "invalid", "boom") + return real_post(path, body, idempotency_key) + + monkeypatch.setattr(agent, "post", broken_gate) + decision = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files") + assert decision.proceed is False and decision.reason == "service_error:invalid" and decision.approval_id + + class Odd: + body = "not a dict" + + monkeypatch.setattr(agent, "post", lambda path, body, idempotency_key=None: Odd() if path.endswith("/gate") else real_post(path, body, idempotency_key)) + assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").reason == "gate_invalid" + + class NoProceed: + body = {"proceed": False} + + monkeypatch.setattr(agent, "post", lambda path, body, idempotency_key=None: NoProceed() if path.endswith("/gate") else real_post(path, body, idempotency_key)) + assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").reason == "gate_blocked" + monkeypatch.setattr(agent, "post", lambda path, body, idempotency_key=None: Odd()) + assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").reason == "approval_invalid" + + +def test_unreachable_service_fails_closed_for_gate_and_open_for_telemetry(tmp_path): + """No service: the gate says no, telemetry returns None, nothing raises into the loop.""" + base, server = start(tmp_path) + server.shutdown() + server.server_close() + agent = HuxClient(base, IDENTITY, key="wk", timeout=1) + decision = before_tool(agent, RUN, "conv_0001abcd", "shell", {"cmd": "ls"}, "shell") + assert decision == decision.__class__(False, None, "hux_unavailable") + assert after_tool(agent, RUN, "conv_0001abcd", "shell", True, 1) is None + assert record_spend(agent, RUN, "conv_0001abcd", tool_calls=1) is None + assert on_stop(agent, RUN, "conv_0001abcd", True) is None + + +def test_autonomy_flag_off_is_reported_and_fails_closed(tmp_path): + """SO-50: with hux.autonomy off, capabilities says so and the hook refuses every side effect.""" + base, server = start(tmp_path, "hux.foundation,hux.activity_timeline,hux.projects") + try: + agent = HuxClient(base, IDENTITY, key="wk") + caps = agent.capabilities() + assert caps["reachable"] and caps["cards"]["HUX-11"] and caps["cards"]["HUX-01"] and not caps["cards"]["HUX-05"] + decision = before_tool(agent, RUN, "conv_0001abcd", "shell", {"cmd": "ls"}, "shell") + assert decision == decision.__class__(False, None, "autonomy_off") + finally: + server.shutdown()