titan/atlas-iac
3
0
Fork 0
Code Issues Pull Requests 9 Actions Packages Projects Releases Wiki Activity

hermes(hux): worker-side hook library for approvals, gates, budgets, stop receipts and events

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RNPhwu2bsaRNg3DETSAZoM
This commit is contained in:
jenkins 2026-08-24 00:37:09 -03:00
parent 1b1a14e972
commit aeef0f8484
6 changed files with 966 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
dockerfiles/hermes-worker-hux/hux_hook/NOTES.md Normal file
View File

@ -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_<hash>` 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=<delta>)`.
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=<real registry check>,
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}`.

24
dockerfiles/hermes-worker-hux/hux_hook/__init__.py Normal file
View File

@ -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",
]

178
dockerfiles/hermes-worker-hux/hux_hook/client.py Normal file
View File

@ -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

191
dockerfiles/hermes-worker-hux/hux_hook/hooks.py Normal file
View File

@ -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:<hex>`` 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")

254
testing/tests/test_hermes_hux_contract_hook.py Normal file
View File

@ -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"<html>"))
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

236
testing/tests/test_hermes_hux_policy_hook.py Normal file
View File

@ -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()