feat(hermes): gate tool runtime through HUX

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.

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)

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)

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

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."""

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

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", "")