diff --git a/services/hermes/plugins/hux-runtime/NOTES.md b/services/hermes/plugins/hux-runtime/NOTES.md new file mode 100644 index 00000000..d2128d43 --- /dev/null +++ b/services/hermes/plugins/hux-runtime/NOTES.md @@ -0,0 +1,37 @@ +# HUX runtime integration + +This plugin wraps Hermes's real `tool_execution` middleware boundary. It is +inert unless `HUX_RUNTIME_ENABLED=1`; `HUX_TOOL_ENFORCEMENT=1` additionally +requires every exact tool call to be released by HUX before `next_call` runs. + +The process needs these read-only files: + +- `HUX_SUBJECT_FILE`: router-published `usr_<64 lowercase hex>` binding; +- `HUX_WORKER_KEY_FILE`: mode `0400`, used only for worker-trust loopback calls; +- `HUX_CONTEXT_KEY_FILE`: the WebUI-created, owner-only, single-link 32-byte + context key. Chat and the runtime use the same `hux.context.id.v1` helper. + +Set `HUX_TENANT_SLOT=slot-N`, `HUX_BASE_URL=http://127.0.0.1:8790`, and a +bounded `HUX_TIMEOUT_SECONDS` (default `3`). The HUX service and plugin must be +in the same pod. `HUX_PROJECT_SOURCE` defaults to `profile:default` and must +match the source used by WebUI context creation. No HUX port belongs in a +Service or ingress. + +Every turn first calls the worker-trust, idempotent +`POST /hux/v1/context/bootstrap`. The sidecar must verify every supplied ID +against the same context key before registering the deterministic project, +session, and conversation. Deployment remains default-off until that route is +present in the converged backend. + +Hermes currently exposes interruption through `on_session_end(interrupted=True)` +but does not give plugins the authoritative process-registry state or reverted +side-effect list required by `hux_hook.on_stop`. The plugin therefore records a +failed/interrupted run and never claims cancellation. A real bridge requires a +new post-interrupt hook fired only after the runtime has killed and reaped every +registered tool process, with `run_id`, `session_id`, `turn_id`, +`process_registry_empty`, and redacted side-effect receipts. + +Quality registration must include `services/hermes/plugins/hux-runtime/*.py` +as source and `testing/tests/test_hermes_hux_runtime_plugin.py` as its focused +test. The current focused gate is 54 passing tests with every production file +above 95% line and branch coverage. diff --git a/services/hermes/plugins/hux-runtime/__init__.py b/services/hermes/plugins/hux-runtime/__init__.py new file mode 100644 index 00000000..f1718fa7 --- /dev/null +++ b/services/hermes/plugins/hux-runtime/__init__.py @@ -0,0 +1,20 @@ +"""Register the default-off HUX runtime enforcement plugin.""" + +from __future__ import annotations + +import os +from typing import Any + +from .runtime import Runtime, UnavailableRuntime, enabled + + +def register(ctx: Any) -> None: + """Register a real execution wrapper only after explicit operator opt-in.""" + if not enabled(os.environ, "HUX_RUNTIME_ENABLED"): + return + try: + runtime: Runtime | UnavailableRuntime = Runtime.from_env() + except Exception: + runtime = UnavailableRuntime(enabled(os.environ, "HUX_TOOL_ENFORCEMENT")) + ctx.register_middleware("tool_execution", runtime.tool_execution) + ctx.register_hook("on_session_end", runtime.session_end) diff --git a/services/hermes/plugins/hux-runtime/context_ids.py b/services/hermes/plugins/hux-runtime/context_ids.py new file mode 100644 index 00000000..957622b6 --- /dev/null +++ b/services/hermes/plugins/hux-runtime/context_ids.py @@ -0,0 +1,105 @@ +"""Derive HUX entity ids from server-owned context and subject files.""" + +from __future__ import annotations + +import hashlib +import hmac +import os +import re +import stat +from pathlib import Path + +DOMAIN = "hux.context.id.v1" +KEY_BYTES = 32 +RAW_ID = re.compile(r"^[A-Za-z0-9._:@+-]{1,200}$") +SLOT = re.compile(r"^slot-[0-9]{1,3}$") +SUBJECT = re.compile(r"^usr_[0-9a-f]{64}$") +PAIRS = frozenset({("ses", "session"), ("conv", "conversation"), ("prj", "project"), ("run", "run")}) + + +class ContextUnavailable(ValueError): + """Trusted runtime context cannot be constructed.""" + + +def _read_owned_key(path: str | Path) -> bytes: + """Read one single-link, owner-only 32-byte context key without following links.""" + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(Path(path), flags) + except OSError as exc: + raise ContextUnavailable("HUX context key is unavailable") from exc + try: + info = os.fstat(descriptor) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_uid != os.geteuid() + or stat.S_IMODE(info.st_mode) != 0o600 + or info.st_nlink != 1 + or info.st_size != KEY_BYTES + ): + raise ContextUnavailable("HUX context key is unsafe") + value = os.read(descriptor, KEY_BYTES + 1) + except ContextUnavailable: + raise + except OSError as exc: + raise ContextUnavailable("HUX context key is unavailable") from exc + finally: + os.close(descriptor) + if len(value) != KEY_BYTES: + raise ContextUnavailable("HUX context key is invalid") + return value + + +def derive_hux_id( + key: bytes, + prefix: str, + purpose: str, + slot: str, + subject: str, + raw: str, +) -> str: + """Match WebUI's public ``hux.context.id.v1`` derivation exactly.""" + if (prefix, purpose) not in PAIRS: + raise ContextUnavailable("unsupported HUX id purpose") + if not SLOT.fullmatch(slot) or not SUBJECT.fullmatch(subject): + raise ContextUnavailable("HUX identity is malformed") + if not isinstance(raw, str) or not RAW_ID.fullmatch(raw): + raise ContextUnavailable("runtime context id is malformed") + if not isinstance(key, bytes) or len(key) != KEY_BYTES: + raise ContextUnavailable("HUX context key is invalid") + message = "\0".join((DOMAIN, purpose, slot, subject, raw)).encode("utf-8") + digest = hmac.new(key, message, hashlib.sha256).hexdigest()[:32] + return f"{prefix}_{digest}" + + +class ContextIds: + """Stable HUX ids for one Hermes tool call.""" + + def __init__(self, key_file: str | Path, slot: str, subject: str) -> None: + if not SLOT.fullmatch(slot) or not SUBJECT.fullmatch(subject): + raise ContextUnavailable("HUX identity is malformed") + self._key = _read_owned_key(key_file) + self._slot = slot + self._subject = subject + + def conversation(self, raw_session_id: str) -> str: + """Map the persisted Hermes/WebUI session to its shared conversation id.""" + return derive_hux_id( + self._key, "conv", "conversation", self._slot, self._subject, raw_session_id + ) + + def session(self, raw_session_id: str) -> str: + """Map the persisted Hermes/WebUI session to its shared session id.""" + return derive_hux_id( + self._key, "ses", "session", self._slot, self._subject, raw_session_id + ) + + def project(self, project_source: str) -> str: + """Map the server-selected project source to its shared project id.""" + return derive_hux_id( + self._key, "prj", "project", self._slot, self._subject, project_source + ) + + def run(self, raw_turn_id: str) -> str: + """Map one stable Hermes turn to its shared HUX run id.""" + return derive_hux_id(self._key, "run", "run", self._slot, self._subject, raw_turn_id) diff --git a/services/hermes/plugins/hux-runtime/plugin.yaml b/services/hermes/plugins/hux-runtime/plugin.yaml new file mode 100644 index 00000000..d90df250 --- /dev/null +++ b/services/hermes/plugins/hux-runtime/plugin.yaml @@ -0,0 +1,7 @@ +name: hux-runtime +version: "1" +description: Fail-closed HUX policy enforcement and redacted tool telemetry. +provides_hooks: + - on_session_end +provides_middleware: + - tool_execution diff --git a/services/hermes/plugins/hux-runtime/runtime.py b/services/hermes/plugins/hux-runtime/runtime.py new file mode 100644 index 00000000..84b30cc6 --- /dev/null +++ b/services/hermes/plugins/hux-runtime/runtime.py @@ -0,0 +1,300 @@ +"""Hermes execution middleware backed by the loopback HUX policy service.""" + +from __future__ import annotations + +import json +import os +import threading +import time +from dataclasses import dataclass +from typing import Any, Callable, Mapping + +from hux_hook import ( + HuxClient, + after_tool, + before_tool, + canonical_argument_hash, + emit, + record_spend, +) + +from .context_ids import ContextIds, ContextUnavailable +from .tool_policy import classify + +BLOCK_SCHEMA = "hux.tool_block.v1" +MAX_REPORTED_BYTES = 2**31 - 1 +TRUE = frozenset({"1", "true", "yes", "on"}) + + +def enabled(environ: Mapping[str, str], name: str) -> bool: + """Require an explicit truthy environment value.""" + return environ.get(name, "").strip().lower() in TRUE + + +def _block(reason: str, approval_id: str | None = None) -> str: + """Return a model-visible denial without request arguments or service details.""" + body: dict[str, Any] = { + "schema": BLOCK_SCHEMA, + "status": "blocked", + "error": "HUX policy did not release this tool call", + "reason": str(reason)[:120] or "hux_unavailable", + } + if approval_id: + body["approval_id"] = str(approval_id)[:120] + return json.dumps(body, sort_keys=True, separators=(",", ":")) + + +def _result_size(result: Any) -> int: + """Measure output for telemetry without persisting or returning its content.""" + if isinstance(result, bytes): + size = len(result) + elif isinstance(result, str): + size = len(result.encode("utf-8", errors="replace")) + else: + try: + size = len(json.dumps(result, ensure_ascii=False, default=str).encode("utf-8")) + except Exception: + size = 0 + return min(MAX_REPORTED_BYTES, max(0, size)) + + +def _result_ok(result: Any) -> bool: + """Detect the common Hermes error envelopes without inspecting free-form output.""" + value = result + if isinstance(result, str) and result.startswith("{"): + try: + value = json.loads(result) + except ValueError: + return True + if not isinstance(value, dict): + return True + status = str(value.get("status", "")).lower() + return "error" not in value and status not in {"blocked", "cancelled", "error", "failed"} + + +@dataclass(frozen=True) +class CallContext: + """Stable HUX scope resolved from host-provided middleware metadata.""" + + conversation_id: str + run_id: str + tool_call_id: str + raw_session_id: str + session_id: str + project_id: str + + +class Runtime: + """One process-wide HUX execution boundary.""" + + def __init__( + self, + client: HuxClient, + ids: ContextIds, + enforce: bool, + project_source: str = "profile:default", + ) -> None: + self.client = client + self.ids = ids + self.enforce = enforce + if not isinstance(project_source, str) or not project_source or len(project_source) > 200: + raise ContextUnavailable("HUX project source is malformed") + self.project_source = project_source + self._started: set[str] = set() + self._bootstrapped: set[str] = set() + self._guard = threading.Lock() + + @classmethod + def from_env(cls, environ: Mapping[str, str] | None = None) -> "Runtime": + """Build only from projected files and literal loopback configuration.""" + env = os.environ if environ is None else environ + slot = env.get("HUX_TENANT_SLOT", "") + subject_file = env.get("HUX_SUBJECT_FILE", "") + key_file = env.get("HUX_WORKER_KEY_FILE", "") + context_key_file = env.get("HUX_CONTEXT_KEY_FILE", "") + if not subject_file or not key_file or not context_key_file: + raise ContextUnavailable("required HUX runtime files are unavailable") + client = HuxClient( + env.get("HUX_BASE_URL", "http://127.0.0.1:8790"), + {"tenant_slot": slot, "surface": "worker", "trust": "worker"}, + key_file=key_file, + subject_file=subject_file, + timeout=float(env.get("HUX_TIMEOUT_SECONDS", "3")), + ) + ids = ContextIds(context_key_file, slot, client.identity["subject"]) + return cls( + client, + ids, + enabled(env, "HUX_TOOL_ENFORCEMENT"), + env.get("HUX_PROJECT_SOURCE", "profile:default"), + ) + + def _scope(self, metadata: Mapping[str, Any]) -> CallContext: + """Resolve IDs only from stable host session/turn values.""" + raw_session = metadata.get("session_id") + raw_turn = metadata.get("turn_id") + if not isinstance(raw_session, str) or not isinstance(raw_turn, str): + raise ContextUnavailable("Hermes session context is unavailable") + return CallContext( + self.ids.conversation(raw_session), + self.ids.run(raw_turn), + str(metadata.get("tool_call_id", ""))[:120], + raw_session, + self.ids.session(raw_session), + self.ids.project(self.project_source), + ) + + def _bootstrap(self, call: CallContext) -> None: + """Idempotently register shared deterministic context before all other HUX use.""" + with self._guard: + if call.conversation_id in self._bootstrapped: + return + self.client.post( + "/hux/v1/context/bootstrap", + { + "raw_session_id": call.raw_session_id, + "project_source": self.project_source, + "session_id": call.session_id, + "conversation_id": call.conversation_id, + "project_id": call.project_id, + }, + idempotency_key=f"context:{call.conversation_id}"[:120], + ) + self._bootstrapped.add(call.conversation_id) + + def _start_once(self, call: CallContext) -> None: + """Emit at most one run-start event per process and stable turn.""" + with self._guard: + first = call.run_id not in self._started + if first: + self._started.add(call.run_id) + if first: + emit( + self.client, + call.conversation_id, + "run.started", + "Hermes started a tool-using turn", + run_id=call.run_id, + idempotency_key=f"{call.run_id}:started", + ) + + def tool_execution( + self, + *, + tool_name: str, + args: dict[str, Any], + next_call: Callable[[dict[str, Any]], Any], + **metadata: Any, + ) -> Any: + """Gate the exact effective arguments, execute once, then emit bounded telemetry.""" + try: + call = self._scope(metadata) + argument_hash = canonical_argument_hash(tool_name, args) + policy = classify(tool_name) + self._bootstrap(call) + self._start_once(call) + emit( + self.client, + call.conversation_id, + "tool.call", + f"{tool_name} requested ({argument_hash[:23]})", + detail={"tool": tool_name, "capability": policy.capability}, + evidence=[{"kind": "tool_call", "id": call.tool_call_id or argument_hash[7:23], "hash": argument_hash}], + run_id=call.run_id, + correlation_id=call.tool_call_id or None, + idempotency_key=f"{call.run_id}:call:{call.tool_call_id or argument_hash[7:23]}"[:120], + ) + if self.enforce: + decision = before_tool( + self.client, + call.run_id, + call.conversation_id, + tool_name, + args, + policy.capability, + external=policy.external, + risk=policy.risk, + ) + if not decision.proceed: + return _block(decision.reason, decision.approval_id) + except Exception: + if self.enforce: + return _block("hux_unavailable") + return next_call(args) + + started = time.monotonic() + try: + result = next_call(args) + except BaseException: + self._after(call, tool_name, argument_hash, False, 0, started) + raise + self._after(call, tool_name, argument_hash, _result_ok(result), _result_size(result), started) + return result + + def _after( + self, + call: CallContext, + tool_name: str, + argument_hash: str, + ok: bool, + bytes_out: int, + started: float, + ) -> None: + """Best-effort status-only event and tool budget increment.""" + duration = max(0, int((time.monotonic() - started) * 1000)) + try: + after_tool( + self.client, + call.run_id, + call.conversation_id, + tool_name, + ok, + bytes_out, + argument_hash=argument_hash, + duration_ms=duration, + ) + except Exception: + pass + try: + record_spend( + self.client, + call.run_id, + call.conversation_id, + tool_calls=1, + wall_clock_seconds=max(0, duration // 1000), + ) + except Exception: + pass + + def session_end(self, **metadata: Any) -> None: + """Record run completion; never claim cancellation without process-registry evidence.""" + try: + call = self._scope(metadata) + self._bootstrap(call) + completed = bool(metadata.get("completed")) and not bool(metadata.get("interrupted")) + kind = "run.completed" if completed else "run.failed" + summary = "Hermes completed the turn" if completed else "Hermes ended the turn without a verified cancellation" + emit( + self.client, + call.conversation_id, + kind, + summary, + run_id=call.run_id, + idempotency_key=f"{call.run_id}:ended", + ) + except Exception: + pass + + +class UnavailableRuntime: + """Fail-closed middleware installed when an opted-in runtime is misconfigured.""" + + def __init__(self, enforce: bool) -> None: + self.enforce = enforce + + def tool_execution(self, *, args: dict[str, Any], next_call: Callable, **_metadata: Any) -> Any: + """Block enforcement mode; preserve telemetry-only fail-open behavior.""" + return _block("hux_unavailable") if self.enforce else next_call(args) + + def session_end(self, **_metadata: Any) -> None: + """There is no trusted scope in which to report telemetry.""" diff --git a/services/hermes/plugins/hux-runtime/tool_policy.py b/services/hermes/plugins/hux-runtime/tool_policy.py new file mode 100644 index 00000000..68862885 --- /dev/null +++ b/services/hermes/plugins/hux-runtime/tool_policy.py @@ -0,0 +1,54 @@ +"""Conservative mapping from Hermes tools to HUX capabilities.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class ToolPolicy: + """The HUX gate inputs for one tool family.""" + + capability: str + risk: str + external: bool + + +EXACT = { + "read_file": ToolPolicy("read_files", "low", False), + "read_many_files": ToolPolicy("read_files", "low", False), + "list_directory": ToolPolicy("read_files", "low", False), + "search_files": ToolPolicy("read_files", "low", False), + "session_search": ToolPolicy("read_files", "low", False), + "tool_search": ToolPolicy("read_files", "low", False), + "write_file": ToolPolicy("write_files", "medium", False), + "patch": ToolPolicy("write_files", "medium", False), + "memory": ToolPolicy("memory_write", "medium", False), + "delegate_task": ToolPolicy("delegate", "high", False), + "terminal": ToolPolicy("shell", "high", True), + "python": ToolPolicy("shell", "high", True), +} + +PREFIXES = ( + (("web_", "browser_", "http_", "mcp_"), ToolPolicy("network", "high", True)), + (("send_", "mail_", "email_", "slack_", "discord_", "telegram_"), ToolPolicy("send_message", "high", True)), + (("kubectl_", "flux_", "deploy_", "release_"), ToolPolicy("deploy", "high", True)), + (("image_", "video_", "vision_"), ToolPolicy("external_side_effect", "high", True)), + (("artifact_",), ToolPolicy("artifact_write", "medium", False)), + (("memory_",), ToolPolicy("memory_write", "medium", False)), + (("delegate_", "subagent_"), ToolPolicy("delegate", "high", False)), + (("write_", "edit_", "file_"), ToolPolicy("write_files", "medium", False)), +) + +UNKNOWN = ToolPolicy("external_side_effect", "high", True) + + +def classify(tool_name: str) -> ToolPolicy: + """Return a known mapping, treating every unknown tool as high-risk external.""" + name = tool_name.strip().lower() if isinstance(tool_name, str) else "" + if name in EXACT: + return EXACT[name] + for prefixes, policy in PREFIXES: + if name.startswith(prefixes): + return policy + return UNKNOWN diff --git a/testing/tests/test_hermes_hux_runtime_plugin.py b/testing/tests/test_hermes_hux_runtime_plugin.py new file mode 100644 index 00000000..17488e8a --- /dev/null +++ b/testing/tests/test_hermes_hux_runtime_plugin.py @@ -0,0 +1,488 @@ +"""Adversarial tests for the Hermes HUX execution middleware.""" + +from __future__ import annotations + +import hashlib +import hmac +import importlib +import importlib.util +import json +import os +import sys +import threading +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +PLUGIN = ROOT / "services" / "hermes" / "plugins" / "hux-runtime" +HOOK_ROOT = ROOT / "dockerfiles" / "hermes-worker-hux" +sys.path.insert(0, str(HOOK_ROOT)) +SPEC = importlib.util.spec_from_file_location( + "hermes_hux_runtime", + PLUGIN / "__init__.py", + submodule_search_locations=[str(PLUGIN)], +) +assert SPEC and SPEC.loader +PACKAGE = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = PACKAGE +SPEC.loader.exec_module(PACKAGE) +context_ids = importlib.import_module("hermes_hux_runtime.context_ids") +runtime_module = importlib.import_module("hermes_hux_runtime.runtime") +tool_policy = importlib.import_module("hermes_hux_runtime.tool_policy") + +SUBJECT = "usr_" + "a" * 64 +SLOT = "slot-2" +KEY = bytes(range(32)) +ARGS = {"path": "notes.txt", "content": "CANARY-secret-value"} + + +class FakeIds: + """Stable ids without touching a key file in middleware-only tests.""" + + def conversation(self, raw: str) -> str: + if not raw: + raise context_ids.ContextUnavailable("missing") + return "conv_" + "c" * 32 + + def session(self, raw: str) -> str: + if not raw: + raise context_ids.ContextUnavailable("missing") + return "ses_" + "b" * 32 + + def project(self, raw: str) -> str: + if not raw: + raise context_ids.ContextUnavailable("missing") + return "prj_" + "a" * 32 + + def run(self, raw: str) -> str: + if not raw: + raise context_ids.ContextUnavailable("missing") + return "run_" + "d" * 32 + + +def middleware_metadata() -> dict[str, str]: + return {"session_id": "session-1", "turn_id": "turn-1", "tool_call_id": "call-1"} + + +class FakeClient: + """Capture only the mandatory context bootstrap request.""" + + def __init__(self): + self.posts = [] + + def post(self, path, body, idempotency_key=None): + self.posts.append((path, body, idempotency_key)) + return SimpleNamespace(body={}) + + +def make_runtime(enforce: bool = True): + return runtime_module.Runtime(FakeClient(), FakeIds(), enforce) + + +def test_context_id_matches_frozen_webui_contract(): + raw = "session:one+two" + message = "\0".join( + ("hux.context.id.v1", "conversation", SLOT, SUBJECT, raw) + ).encode() + expected = "conv_" + hmac.new(KEY, message, hashlib.sha256).hexdigest()[:32] + assert context_ids.derive_hux_id(KEY, "conv", "conversation", SLOT, SUBJECT, raw) == expected + run = context_ids.derive_hux_id(KEY, "run", "run", SLOT, SUBJECT, "turn-9") + assert run.startswith("run_") and len(run) == 36 + + +@pytest.mark.parametrize( + ("prefix", "purpose", "slot", "subject", "raw", "key"), + [ + ("bad", "conversation", SLOT, SUBJECT, "s", KEY), + ("conv", "conversation", "tenant-2", SUBJECT, "s", KEY), + ("conv", "conversation", SLOT, "usr_short", "s", KEY), + ("conv", "conversation", SLOT, SUBJECT, "slash/not-allowed", KEY), + ("conv", "conversation", SLOT, SUBJECT, "s", b"short"), + ], +) +def test_context_id_rejects_noncanonical_inputs(prefix, purpose, slot, subject, raw, key): + with pytest.raises(context_ids.ContextUnavailable): + context_ids.derive_hux_id(key, prefix, purpose, slot, subject, raw) + + +def test_context_key_requires_owner_only_regular_single_link_file(tmp_path: Path): + path = tmp_path / "context-key" + path.write_bytes(KEY) + path.chmod(0o600) + ids = context_ids.ContextIds(path, SLOT, SUBJECT) + assert ids.conversation("session-1").startswith("conv_") + assert ids.session("session-1").startswith("ses_") + assert ids.project("profile:default").startswith("prj_") + assert ids.run("turn-1").startswith("run_") + path.chmod(0o640) + with pytest.raises(context_ids.ContextUnavailable, match="unsafe"): + context_ids.ContextIds(path, SLOT, SUBJECT) + path.chmod(0o600) + linked = tmp_path / "linked" + os.link(path, linked) + with pytest.raises(context_ids.ContextUnavailable, match="unsafe"): + context_ids.ContextIds(path, SLOT, SUBJECT) + linked.unlink() + path.write_bytes(b"short") + with pytest.raises(context_ids.ContextUnavailable): + context_ids.ContextIds(path, SLOT, SUBJECT) + + +def test_context_key_rejects_missing_and_symlink(tmp_path: Path): + with pytest.raises(context_ids.ContextUnavailable, match="unavailable"): + context_ids.ContextIds(tmp_path / "missing", SLOT, SUBJECT) + real = tmp_path / "real" + real.write_bytes(KEY) + real.chmod(0o600) + link = tmp_path / "link" + link.symlink_to(real) + with pytest.raises(context_ids.ContextUnavailable): + context_ids.ContextIds(link, SLOT, SUBJECT) + with pytest.raises(context_ids.ContextUnavailable, match="identity"): + context_ids.ContextIds(real, SLOT, "usr_" + "A" * 64) + + +def test_context_key_handles_short_read(monkeypatch, tmp_path: Path): + path = tmp_path / "key" + path.write_bytes(KEY) + path.chmod(0o600) + monkeypatch.setattr(context_ids.os, "read", lambda *_args: b"short") + with pytest.raises(context_ids.ContextUnavailable, match="invalid"): + context_ids.ContextIds(path, SLOT, SUBJECT) + + +@pytest.mark.parametrize( + ("name", "capability", "risk", "external"), + [ + ("read_file", "read_files", "low", False), + ("write_file", "write_files", "medium", False), + ("memory", "memory_write", "medium", False), + ("delegate_task", "delegate", "high", False), + ("terminal", "shell", "high", True), + ("web_search", "network", "high", True), + ("send_email", "send_message", "high", True), + ("kubectl_apply", "deploy", "high", True), + ("image_generate", "external_side_effect", "high", True), + ("artifact_create", "artifact_write", "medium", False), + ("memory_forget", "memory_write", "medium", False), + ("subagent_spawn", "delegate", "high", False), + ("edit_document", "write_files", "medium", False), + ("new_plugin_tool", "external_side_effect", "high", True), + (None, "external_side_effect", "high", True), + ], +) +def test_tool_mapping_is_conservative(name, capability, risk, external): + assert tool_policy.classify(name) == tool_policy.ToolPolicy(capability, risk, external) + + +def test_default_off_registers_nothing(monkeypatch): + hooks, middleware = {}, {} + ctx = SimpleNamespace( + register_hook=lambda name, callback: hooks.setdefault(name, callback), + register_middleware=lambda name, callback: middleware.setdefault(name, callback), + ) + monkeypatch.delenv("HUX_RUNTIME_ENABLED", raising=False) + PACKAGE.register(ctx) + assert not hooks and not middleware + + +@pytest.mark.parametrize(("enforce", "executes"), [("1", False), ("0", True)]) +def test_misconfigured_opt_in_closes_only_enforcement(monkeypatch, enforce, executes): + hooks, middleware = {}, {} + ctx = SimpleNamespace( + register_hook=lambda name, callback: hooks.setdefault(name, callback), + register_middleware=lambda name, callback: middleware.setdefault(name, callback), + ) + monkeypatch.setenv("HUX_RUNTIME_ENABLED", "true") + monkeypatch.setenv("HUX_TOOL_ENFORCEMENT", enforce) + monkeypatch.setattr(PACKAGE.Runtime, "from_env", classmethod(lambda cls: (_ for _ in ()).throw(ValueError()))) + PACKAGE.register(ctx) + called = [] + result = middleware["tool_execution"](args=ARGS, next_call=lambda value: called.append(value) or "ok") + assert bool(called) is executes + assert (result == "ok") is executes + assert set(hooks) == {"on_session_end"} + + +def test_from_env_reads_subject_only_from_file(tmp_path: Path): + subject = tmp_path / "subject" + subject.write_text(SUBJECT + "\n") + subject.chmod(0o400) + worker = tmp_path / "worker" + worker.write_text("worker-key") + worker.chmod(0o400) + context = tmp_path / "context" + context.write_bytes(KEY) + context.chmod(0o600) + env = { + "HUX_TENANT_SLOT": SLOT, + "HUX_SUBJECT_FILE": str(subject), + "HUX_WORKER_KEY_FILE": str(worker), + "HUX_CONTEXT_KEY_FILE": str(context), + "HUX_SUBJECT": "usr_" + "b" * 64, + "HUX_TOOL_ENFORCEMENT": "yes", + "HUX_TIMEOUT_SECONDS": "0.2", + } + instance = runtime_module.Runtime.from_env(env) + assert instance.client.identity == { + "tenant_slot": SLOT, + "subject": SUBJECT, + "surface": "worker", + "trust": "worker", + } + assert instance.enforce is True + assert instance.project_source == "profile:default" + + +def test_from_env_rejects_missing_file_contract(): + with pytest.raises(context_ids.ContextUnavailable): + runtime_module.Runtime.from_env({"HUX_TENANT_SLOT": SLOT}) + + +@pytest.mark.parametrize("project_source", [None, "", "x" * 201]) +def test_runtime_rejects_malformed_project_source(project_source): + with pytest.raises(context_ids.ContextUnavailable, match="project source"): + runtime_module.Runtime(FakeClient(), FakeIds(), False, project_source) + + +@pytest.mark.parametrize( + "metadata", + [ + {"session_id": None, "turn_id": "turn-1"}, + {"session_id": "session-1", "turn_id": None}, + ], +) +def test_scope_requires_host_owned_string_ids(metadata): + with pytest.raises(context_ids.ContextUnavailable, match="session context"): + make_runtime(False)._scope(metadata) + + +def test_denial_never_executes_and_exposes_no_arguments(monkeypatch): + instance = make_runtime() + calls, telemetry = [], [] + monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: telemetry.append((args[1:], kwargs))) + + def deny(_client, run_id, conversation_id, name, arguments, capability, **policy): + calls.append((run_id, conversation_id, name, arguments, capability, policy)) + return SimpleNamespace(proceed=False, reason="approval_required", approval_id="apr_safe") + + monkeypatch.setattr(runtime_module, "before_tool", deny) + downstream = [] + result = instance.tool_execution( + tool_name="write_file", + args=ARGS, + next_call=lambda args: downstream.append(args), + **middleware_metadata(), + ) + assert not downstream and calls[0][3] is ARGS + assert calls[0][-1] == {"external": False, "risk": "medium"} + assert json.loads(result) == { + "approval_id": "apr_safe", + "error": "HUX policy did not release this tool call", + "reason": "approval_required", + "schema": "hux.tool_block.v1", + "status": "blocked", + } + assert ARGS["content"] not in json.dumps(telemetry) + + +def test_bootstrap_uses_exact_ids_and_replays_in_process(monkeypatch): + instance = make_runtime(False) + monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {}) + monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: None) + monkeypatch.setattr(runtime_module, "record_spend", lambda *args, **kwargs: None) + for call_id in ("call-1", "call-2"): + instance.tool_execution( + tool_name="read_file", + args={}, + next_call=lambda _args: "ok", + session_id="session-1", + turn_id="turn-1", + tool_call_id=call_id, + ) + assert instance.client.posts == [ + ( + "/hux/v1/context/bootstrap", + { + "raw_session_id": "session-1", + "project_source": "profile:default", + "session_id": "ses_" + "b" * 32, + "conversation_id": "conv_" + "c" * 32, + "project_id": "prj_" + "a" * 32, + }, + "context:conv_" + "c" * 32, + ) + ] + + +def test_bootstrap_failure_blocks_enforcement_and_opens_telemetry(monkeypatch): + monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {}) + for enforce in (True, False): + instance = make_runtime(enforce) + instance.client.post = lambda *args, **kwargs: (_ for _ in ()).throw(OSError("offline")) + called = [] + result = instance.tool_execution( + tool_name="read_file", + args={}, + next_call=lambda value: called.append(value) or "ok", + **middleware_metadata(), + ) + assert bool(called) is (not enforce) + assert (result == "ok") is (not enforce) + + +def test_enforcement_failure_is_closed_but_telemetry_failure_is_open(monkeypatch): + for enforce in (True, False): + instance = make_runtime(enforce) + monkeypatch.setattr(runtime_module, "canonical_argument_hash", lambda *_: (_ for _ in ()).throw(TypeError())) + called = [] + result = instance.tool_execution( + tool_name="write_file", args=ARGS, next_call=lambda args: called.append(args) or "done", **middleware_metadata() + ) + assert bool(called) is (not enforce) + assert (result == "done") is (not enforce) + + +def test_released_call_executes_exact_object_once_and_reports_status_only(monkeypatch): + instance = make_runtime() + events, after, spend = [], [], [] + monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: events.append((args[1:], kwargs))) + monkeypatch.setattr( + runtime_module, + "before_tool", + lambda *args, **kwargs: SimpleNamespace(proceed=True, reason="released", approval_id="apr_1"), + ) + monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: after.append((args[1:], kwargs))) + monkeypatch.setattr(runtime_module, "record_spend", lambda *args, **kwargs: spend.append((args, kwargs))) + downstream = [] + + def execute(value): + downstream.append(value) + return {"ok": True, "payload": "RAW-RESULT-SECRET"} + + result = instance.tool_execution( + tool_name="write_file", args=ARGS, next_call=execute, **middleware_metadata() + ) + assert downstream == [ARGS] and downstream[0] is ARGS + assert result["payload"] == "RAW-RESULT-SECRET" + assert after[0][0][3] is True and after[0][0][4] > 0 + assert spend[0][1]["tool_calls"] == 1 + assert ARGS["content"] not in json.dumps(events) + assert "RAW-RESULT-SECRET" not in json.dumps(after) + + +def test_downstream_exception_is_reported_then_re_raised(monkeypatch): + instance = make_runtime() + after = [] + monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {}) + monkeypatch.setattr( + runtime_module, + "before_tool", + lambda *args, **kwargs: SimpleNamespace(proceed=True, reason="released", approval_id="apr_1"), + ) + monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: after.append(args)) + monkeypatch.setattr(runtime_module, "record_spend", lambda *args, **kwargs: None) + + def explode(_args): + raise RuntimeError("RAW-EXECUTION-SECRET") + + with pytest.raises(RuntimeError, match="RAW-EXECUTION-SECRET"): + instance.tool_execution(tool_name="terminal", args=ARGS, next_call=explode, **middleware_metadata()) + assert after[0][4] is False and after[0][5] == 0 + + +def test_after_telemetry_failure_never_changes_tool_result(monkeypatch): + instance = make_runtime(False) + monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {}) + monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError())) + assert instance.tool_execution( + tool_name="read_file", args={}, next_call=lambda _args: "result", **middleware_metadata() + ) == "result" + + +def test_run_start_is_single_fire_under_concurrency(monkeypatch): + instance = make_runtime(False) + events = [] + monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: events.append(args[2])) + call = runtime_module.CallContext( + "conv_" + "c" * 32, + "run_" + "d" * 32, + "call", + "session-1", + "ses_" + "b" * 32, + "prj_" + "a" * 32, + ) + threads = [threading.Thread(target=instance._start_once, args=(call,)) for _ in range(20)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert events == ["run.started"] + + +@pytest.mark.parametrize( + ("result", "ok"), + [ + ("plain", True), + ("{not json", True), + ({"status": "ok"}, True), + ({"error": "no"}, False), + (json.dumps({"status": "cancelled"}), False), + ({"status": "failed"}, False), + ], +) +def test_result_status_detection(result, ok): + assert runtime_module._result_ok(result) is ok + + +def test_result_size_is_bounded_and_tolerates_bad_values(monkeypatch): + class Bad: + def __str__(self): + raise ValueError + + assert runtime_module._result_size(b"abc") == 3 + assert runtime_module._result_size("é") == 2 + assert runtime_module._result_size(Bad()) == 0 + monkeypatch.setattr(runtime_module, "MAX_REPORTED_BYTES", 2) + assert runtime_module._result_size(b"xxx") == 2 + + +@pytest.mark.parametrize( + ("completed", "interrupted", "kind"), + [(True, False, "run.completed"), (True, True, "run.failed"), (False, False, "run.failed")], +) +def test_session_end_never_claims_unverified_cancellation(monkeypatch, completed, interrupted, kind): + instance = make_runtime(False) + events = [] + monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: events.append((args, kwargs))) + instance.session_end(completed=completed, interrupted=interrupted, **middleware_metadata()) + assert events[0][0][2] == kind + assert "cancel" not in events[0][0][3].lower() or "without a verified cancellation" in events[0][0][3] + + +def test_session_end_and_after_ignore_telemetry_errors(monkeypatch): + instance = make_runtime(False) + monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError())) + instance.session_end(session_id="bad session", turn_id="bad turn") + monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError())) + call = runtime_module.CallContext( + "conv_" + "c" * 32, + "run_" + "d" * 32, + "call", + "session-1", + "ses_" + "b" * 32, + "prj_" + "a" * 32, + ) + instance._after(call, "read_file", "sha256:" + "e" * 64, True, 1, 0.0) + + +def test_plugin_files_stay_small_and_stop_bridge_is_honest(): + for path in PLUGIN.glob("*.py"): + assert len(path.read_text().splitlines()) <= 500 + source = (PLUGIN / "runtime.py").read_text() + notes = (PLUGIN / "NOTES.md").read_text() + assert "on_stop(" not in source + assert "process_registry_empty" in notes + assert "HUX_SUBJECT" not in source.replace("HUX_SUBJECT_FILE", "")