diff --git a/services/hermes/plugins/hux-runtime/hux_hook/__init__.py b/services/hermes/plugins/hux-runtime/hux_hook/__init__.py
new file mode 100644
index 00000000..374e3a74
--- /dev/null
+++ b/services/hermes/plugins/hux-runtime/hux_hook/__init__.py
@@ -0,0 +1,24 @@
+"""Agent-side hook library for the per-tenant HUX service.
+
+Stdlib only. The Hermes agent process (tenant pod or Worker) calls these
+functions around its tool loop so that HUX, not the agent, decides approvals,
+gates, budgets, stop receipts and what lands on the activity timeline.
+Side effects fail closed when the service is unreachable; telemetry fails open.
+"""
+
+from .client import HuxClient, HuxServiceError, HuxUnavailable
+from .hooks import (
+    Decision,
+    after_tool,
+    before_tool,
+    canonical_argument_hash,
+    emit,
+    memory_gate,
+    on_stop,
+    record_spend,
+)
+
+__all__ = [
+    "Decision", "HuxClient", "HuxServiceError", "HuxUnavailable", "after_tool", "before_tool",
+    "canonical_argument_hash", "emit", "memory_gate", "on_stop", "record_spend",
+]
diff --git a/services/hermes/plugins/hux-runtime/hux_hook/client.py b/services/hermes/plugins/hux-runtime/hux_hook/client.py
new file mode 100644
index 00000000..6188069e
--- /dev/null
+++ b/services/hermes/plugins/hux-runtime/hux_hook/client.py
@@ -0,0 +1,300 @@
+"""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 hmac
+import json
+import os
+import re
+import stat
+import threading
+import urllib.error
+import urllib.parse
+import urllib.request
+from collections.abc import Mapping
+from pathlib import Path
+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
+MAX_RESPONSE_BYTES = 4 * 1024 * 1024
+MAX_KEY_BYTES = 4096
+MAX_SUBJECT_BYTES = 128
+LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1"})
+SUBJECT_RE = re.compile(r"^usr_[0-9a-f]{16,64}$")
+
+
+class _RejectRedirect(urllib.request.HTTPRedirectHandler):
+    """Never replay tenant identity or worker credentials to a redirect target."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ANN001, ANN201, ARG002
+        """Refuse every redirect rather than replaying the original headers."""
+        return None
+
+
+def _validated_base_url(raw: str) -> str:
+    """Return a canonical literal-loopback HTTP origin or reject it."""
+    parts = urllib.parse.urlsplit(raw)
+    if (
+        parts.scheme != "http"
+        or parts.hostname not in LOOPBACK_HOSTS
+        or parts.username is not None
+        or parts.password is not None
+        or parts.query
+        or parts.fragment
+        or parts.path not in {"", "/"}
+        or parts.port is None
+    ):
+        raise ValueError("HUX base URL must be a literal loopback HTTP origin with an explicit port")
+    host = f"[{parts.hostname}]" if parts.hostname == "::1" else parts.hostname
+    return f"http://{host}:{parts.port}"
+
+
+def _validated_path(path: str) -> str:
+    """Accept only canonical relative HUX API paths owned by this client."""
+    if not isinstance(path, str):
+        raise HuxServiceError(400, "invalid", "malformed request path")
+    decoded = urllib.parse.unquote(path)
+    if (
+        not path.startswith("/hux/v1/")
+        or path.startswith("//")
+        or any(char in path for char in "?#\\\r\n")
+        or any(segment in {".", ".."} for segment in decoded.split("/"))
+    ):
+        raise HuxServiceError(400, "invalid", "malformed request path")
+    return path
+
+
+def _key_from_file(key_file: str | Path) -> str:
+    """Read one 0400 regular-file credential without accepting weak permissions or unbounded data."""
+    path = Path(key_file)
+    try:
+        mode = path.stat().st_mode
+        if not stat.S_ISREG(mode) or stat.S_IMODE(mode) != 0o400:
+            raise ValueError("HUX key file must be a 0400 regular file")
+        data = path.read_bytes()
+        if not data or len(data) > MAX_KEY_BYTES:
+            raise ValueError("HUX key file is empty or oversized")
+        value = data.decode("utf-8", errors="strict").strip()
+        if not value:
+            raise ValueError("HUX key file is empty")
+        return value
+    except OSError as error:
+        raise ValueError("HUX key file is unavailable") from error
+    except UnicodeDecodeError as error:
+        raise ValueError("HUX key file is not UTF-8") from error
+
+
+def _subject_from_file(subject_file: str | Path) -> str:
+    """Read one router-published subject binding from a read-only shared file."""
+    path = Path(subject_file)
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except OSError as error:
+        raise ValueError("HUX subject file is unavailable") from error
+    try:
+        mode = os.fstat(descriptor).st_mode
+        if not stat.S_ISREG(mode) or stat.S_IMODE(mode) not in {0o400, 0o440}:
+            raise ValueError("HUX subject file must be a 0400 or 0440 regular file")
+        with os.fdopen(descriptor, "rb", closefd=False) as subject_stream:
+            data = subject_stream.read(MAX_SUBJECT_BYTES + 1)
+    except OSError as error:
+        raise ValueError("HUX subject file is unavailable") from error
+    finally:
+        os.close(descriptor)
+    if not data or len(data) > MAX_SUBJECT_BYTES:
+        raise ValueError("HUX subject file is empty or oversized")
+    try:
+        value = data.decode("utf-8", errors="strict").strip()
+    except UnicodeDecodeError as error:
+        raise ValueError("HUX subject file is not UTF-8") from error
+    if not SUBJECT_RE.fullmatch(value):
+        raise ValueError("HUX subject file is malformed")
+    return value
+
+
+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, *, key_file: str | Path | None = None,
+                 subject_file: str | Path | None = None) -> None:
+        identity = dict(identity or {})
+        configured_subject_file = subject_file if subject_file is not None else os.environ.get("HUX_SUBJECT_FILE")
+        if configured_subject_file:
+            bound_subject = _subject_from_file(configured_subject_file)
+            asserted_subject = str(identity.get("subject", ""))
+            if asserted_subject and not hmac.compare_digest(asserted_subject, bound_subject):
+                raise ValueError("HUX identity subject conflicts with the trusted binding")
+            identity["subject"] = bound_subject
+        self.base_url = _validated_base_url(base_url)
+        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")),
+        }
+        if key is not None and key_file is not None:
+            raise ValueError("set either key or key_file, not both")
+        self._key = _key_from_file(key_file) if key_file is not None else key
+        if isinstance(timeout, bool) or not isinstance(timeout, int | float) or not 0.1 <= float(timeout) <= 30:
+            raise ValueError("timeout must be between 0.1 and 30 seconds")
+        self.timeout = float(timeout)
+        self._opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), _RejectRedirect())
+        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 + _validated_path(path)
+        if query:
+            url += "?" + urllib.parse.urlencode({str(k): str(v) for k, v in query.items()})
+        req = urllib.request.Request(url, data=data, method=method, headers=self.headers(extra))
+        try:
+            with self._opener.open(req, timeout=self.timeout) as raw:  # noqa: S310 - validated literal loopback only
+                response = HuxResponse(raw.status, _decode(_bounded_read(raw)), dict(raw.headers.items()))
+        except urllib.error.HTTPError as error:
+            payload = _decode(_bounded_read(error))
+            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
+
+
+def _bounded_read(raw: Any) -> bytes:
+    """Read one bounded response so a compromised sidecar cannot exhaust the worker."""
+    data = raw.read(MAX_RESPONSE_BYTES + 1)
+    if len(data) > MAX_RESPONSE_BYTES:
+        raise HuxUnavailable("hux service returned an oversized response")
+    return data
diff --git a/services/hermes/plugins/hux-runtime/hux_hook/hooks.py b/services/hermes/plugins/hux-runtime/hux_hook/hooks.py
new file mode 100644
index 00000000..b65391a8
--- /dev/null
+++ b/services/hermes/plugins/hux-runtime/hux_hook/hooks.py
@@ -0,0 +1,193 @@
+"""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 .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")
+    try:
+        # A zero-spend write pins the trusted worker's run -> conversation
+        # binding before a scoped session grant can create a new approval.
+        client.post(f"/hux/v1/runs/{run_id}/budget", {"conversation_id": conversation_id})
+    except HuxServiceError as error:
+        return Decision(False, None, _failure_reason(error))
+    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:
+        state = client.get(f"/hux/v1/conversations/{conversation_id}/privacy").body
+    except HuxServiceError:
+        return False
+    return bool(isinstance(state, dict) and state.get("memory_writes_allowed"))
diff --git a/services/hermes/plugins/hux-runtime/runtime.py b/services/hermes/plugins/hux-runtime/runtime.py
index 84b30cc6..237ac2fd 100644
--- a/services/hermes/plugins/hux-runtime/runtime.py
+++ b/services/hermes/plugins/hux-runtime/runtime.py
@@ -9,7 +9,7 @@ import time
 from dataclasses import dataclass
 from typing import Any, Callable, Mapping
 
-from hux_hook import (
+from .hux_hook import (
     HuxClient,
     after_tool,
     before_tool,
diff --git a/testing/tests/test_hermes_hux_runtime_vendor_parity.py b/testing/tests/test_hermes_hux_runtime_vendor_parity.py
new file mode 100644
index 00000000..77305bd4
--- /dev/null
+++ b/testing/tests/test_hermes_hux_runtime_vendor_parity.py
@@ -0,0 +1,89 @@
+"""Keep the Kustomize-local HUX hook package identical to its canonical source."""
+
+from __future__ import annotations
+
+import ast
+import importlib.util
+import inspect
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+CANONICAL = ROOT / "dockerfiles" / "hermes-worker-hux" / "hux_hook"
+VENDORED = ROOT / "services" / "hermes" / "plugins" / "hux-runtime" / "hux_hook"
+FILES = ("__init__.py", "client.py", "hooks.py")
+
+
+def _canonical_bytes(path: Path, vendored: bool) -> bytes:
+    """Normalize only the package relocation imports before byte comparison."""
+    value = path.read_bytes()
+    if vendored:
+        value = value.replace(b"from .client import", b"from hux_hook.client import")
+        value = value.replace(b"from .hooks import", b"from hux_hook.hooks import")
+    return value
+
+
+class _CanonicalImports(ast.NodeTransformer):
+    """Normalize the same relative imports for structural comparison."""
+
+    def visit_ImportFrom(self, node: ast.ImportFrom):  # noqa: N802
+        if node.level == 1 and node.module in {"client", "hooks"}:
+            node.level = 0
+            node.module = f"hux_hook.{node.module}"
+        return node
+
+
+def _canonical_ast(value: bytes) -> str:
+    tree = _CanonicalImports().visit(ast.parse(value))
+    return ast.dump(ast.fix_missing_locations(tree), include_attributes=False)
+
+
+def _load(name: str, root: Path):
+    """Load one package under a unique name so both copies coexist."""
+    spec = importlib.util.spec_from_file_location(
+        name,
+        root / "__init__.py",
+        submodule_search_locations=[str(root)],
+    )
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_vendored_hook_bytes_and_ast_match_canonical_source():
+    """Any canonical change requires an intentional vendor refresh in the same diff."""
+    for name in FILES:
+        canonical = _canonical_bytes(CANONICAL / name, False)
+        vendored = _canonical_bytes(VENDORED / name, True)
+        assert vendored == canonical, f"refresh vendored hux_hook/{name}"
+        assert _canonical_ast((VENDORED / name).read_bytes()) == _canonical_ast(
+            (CANONICAL / name).read_bytes()
+        )
+
+
+def test_vendored_hook_exports_and_signatures_match_canonical_api():
+    """Protect the runtime from a byte-equal but incorrectly loaded package surface."""
+    for name in tuple(sys.modules):
+        if name == "hux_hook" or name.startswith("hux_hook."):
+            sys.modules.pop(name)
+    canonical = _load("hux_hook", CANONICAL)
+    vendored = _load("vendored_hux_hook", VENDORED)
+    assert vendored.__all__ == canonical.__all__
+    for name in canonical.__all__:
+        canonical_value = getattr(canonical, name)
+        vendored_value = getattr(vendored, name)
+        assert type(vendored_value).__name__ == type(canonical_value).__name__
+        assert inspect.signature(vendored_value) == inspect.signature(canonical_value)
+
+
+def test_runtime_uses_only_its_kustomize_local_hook_package():
+    """A rendered plugin must not depend on the Docker build-context path."""
+    source = (VENDORED.parent / "runtime.py").read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    imports = [node for node in ast.walk(tree) if isinstance(node, ast.ImportFrom)]
+    assert any(node.level == 1 and node.module == "hux_hook" for node in imports)
+    assert not any(node.level == 0 and node.module == "hux_hook" for node in imports)
+    for path in (*CANONICAL.glob("*.py"), *VENDORED.glob("*.py")):
+        assert len(path.read_text(encoding="utf-8").splitlines()) <= 500