diff --git a/docs/architecture/image-vs-hsm-boundary.md b/docs/architecture/image-vs-hsm-boundary.md
index 250aafe..443d709 100644
--- a/docs/architecture/image-vs-hsm-boundary.md
+++ b/docs/architecture/image-vs-hsm-boundary.md
@@ -69,7 +69,7 @@ NimbleCoAI/hermes-agent is a fork of NousResearch/hermes-agent. Every feature th
 
 - **Useful to everyone** → upstream it (PR to NousResearch/hermes-agent) or cherry-pick to the public fork
 - **NimbleCo-specific** → keep it in HSM scaffolding where possible (plugins, hooks, config). If it *must* be in the image (e.g., security fix), accept the fork cost
-- **Experimental** → keep it in the private fork (hermes-swarm) until validated, then decide
+- **Experimental** → if it can be expressed through extension points (a plugin, a hook, a skill — no core edits), ship it as a **standalone, opt-in HSM artifact**: *not* in the image, and *not* in the default base bundle. Because it's unproven, it shouldn't ride the image everyone pulls or auto-load on every agent — install it per-agent to validate, then promote it (into the base bundle, or upstream) once proven. Only fall back to "carry it in the private fork" if it genuinely cannot be done without core edits.
 
 **Current fork divergence:**
 - Public fork removes ~25 upstream tools (platform-specific integrations not needed for NimbleCo)
@@ -101,6 +101,8 @@ NimbleCoAI/hermes-agent is a fork of NousResearch/hermes-agent. Every feature th
 | **Security defaults** | `HERMES_DM_POLICY=approved-only`, `HERMES_APPROVAL_ADMIN_ONLY=true`, `*_REQUIRE_MENTION=true` | Env vars that activate image-level features |
 | **Docker hardening** | `read_only: true`, `cap_drop: ALL`, user 10000:10000, 2GB/2CPU limits | Compose-level security |
 
+**Default bundle vs optional artifacts.** Not everything HSM *can* install is installed on every agent. The table above is the **default base bundle** — the sane, proven defaults every agent gets (see the *opinionated HSM base package* decision). HSM also hosts **optional / experimental artifacts** in the same `infra/templates/` tree that are installed **per-agent, opt-in** (a plugin enabled via `plugins.enabled`, a skill installed explicitly) and are deliberately kept *out* of the default bundle until validated. `human_escalation` (+ the `checkout` skill) is the first such optional artifact: an experimental human-in-the-loop purchasing capability — standalone, zero image footprint, not in the base bundle.
+
 ### The Boundary in Practice
 
 ```
@@ -156,3 +158,8 @@ NimbleCoAI/hermes-agent is a fork of NousResearch/hermes-agent. Every feature th
 
 **"I want agents to auto-solve CAPTCHAs during browser automation"**
 → HSM. The browser tool already returns `bot_detection_warning` (image). CAPTCHA solving depends on deployment-specific services (Camofox URL, CapSolver API key). Ship as a plugin (`captcha_cascade`) that registers a `captcha_solve` tool. The agent's skill teaches it when to call the tool. No fork changes needed — the image provides the detection, the plugin provides the resolution.
+
+**"I want agents to complete purchases with a human in the loop (relay a 2FA code, confirm the charge)"**
+→ HSM, standalone + experimental. Ship as a plugin (`human_escalation` — registers `escalate_to_human` + `check_pending_escalation` tools and a `pre_gateway_dispatch` resume hook) plus a `checkout` file skill, both under `infra/templates/`. It rides only existing extension points (tool registration, the dispatch hook, the `send_message` tool, session env vars) — **zero fork changes, no recurring rebase cost**. Because it's unproven *and* spends real money, it stays an **opt-in** artifact (not the base bundle) until validated on a real agent. Design note: it deliberately avoids a gateway patch — the escalation tool posts to chat and returns, and the hook rewrites the user's reply to resume — so the whole feature lives in `infra/templates/` with nothing in the image.
+
+> **Skill packaging caveat.** A skill that should *auto-trigger* (appear in the agent's `<available_skills>` index from its `description`) must ship as a **file skill** at `infra/templates/skills/<name>/SKILL.md` (like `checkout`, `captcha-escalation`, `ocr-and-documents`). A plugin can also expose a skill via `ctx.register_skill(...)`, but those are **opt-in explicit loads only** — *not* indexed and *not* auto-triggered (see `hermes_cli/plugins.py` `register_skill`). Use `register_skill` for reference docs the agent loads on demand; use a file skill when user intent should trigger it.
diff --git a/infra/templates/plugins/human_escalation/__init__.py b/infra/templates/plugins/human_escalation/__init__.py
new file mode 100644
index 0000000..d33d635
--- /dev/null
+++ b/infra/templates/plugins/human_escalation/__init__.py
@@ -0,0 +1,99 @@
+"""human_escalation plugin -- registers the escalation-bus tools and the resume hook.
+
+Standalone HSM artifact. Installed by HSM into a plugin directory (e.g.
+``/opt/data/.../plugins/human_escalation/`` or ``~/.hermes/plugins/human_escalation/``)
+where there is NO ``plugins.human_escalation`` parent package. The real loader
+(hermes-agent's ``PluginManager._load_directory_module``) imports this file as
+``hermes_plugins.human_escalation`` with ``submodule_search_locations=[plugin_dir]``
+and does NOT put plugin_dir on ``sys.path`` -- so ``register()`` resolves its
+siblings with a loader-agnostic importlib-by-__file__ fallback (the same pattern
+escalate.py / dispatch_hook.py already use to import store.py).
+"""
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+_ESCALATE_SCHEMA = {
+    "name": "escalate_to_human",
+    "description": (
+        "Ask the human (over the current chat) for something the agent cannot do alone, "
+        "then END THE TURN and wait for their reply. Use kind='code_request' to relay an "
+        "SMS/2FA/email code, kind='confirmation' to get explicit go-ahead before charging "
+        "(payload: {line_items:[...], total:'$X'}), kind='link_handoff' to send a VNC/QR/link "
+        "to finish a step (payload: {url, media_path?}), kind='freeform' for an open question. "
+        "Returns immediately with status='awaiting'; the reply is retrieved next turn via "
+        "check_pending_escalation."
+    ),
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "kind": {"type": "string", "enum": ["code_request", "confirmation", "link_handoff", "freeform"]},
+            "prompt": {"type": "string", "description": "The message to show the human."},
+            "payload": {"type": "object", "description": "kind-specific: confirmation={line_items,total}; link_handoff={url,media_path?}."},
+            "timeout_s": {"type": "integer", "description": "Seconds the escalation stays valid (default 300)."},
+        },
+        "required": ["kind", "prompt"],
+    },
+}
+
+_CHECK_SCHEMA = {
+    "name": "check_pending_escalation",
+    "description": (
+        "Call this at the START of a turn when you may be resuming a checkout. Returns the "
+        "pending escalation (kind + original prompt) for this chat and clears it; the user's "
+        "most recent message is the answer. Returns status='none' if nothing is pending."
+    ),
+    "parameters": {"type": "object", "properties": {}, "required": []},
+}
+
+
+def register(ctx):
+    # Loader-agnostic sibling imports. The real HSM loader imports this package as
+    # ``hermes_plugins.human_escalation`` with submodule_search_locations set but
+    # does NOT add the plugin dir to sys.path, so a bare ``from escalate import``
+    # works only when the dir happens to be importable. When it is not (the
+    # standalone install location), fall back to loading escalate.py /
+    # dispatch_hook.py directly by __file__ -- the same self-contained importlib
+    # pattern escalate.py and dispatch_hook.py use for store.py. This makes the
+    # plugin load regardless of the package name the loader assigns and with NO
+    # dependency on a ``plugins.human_escalation`` parent package.
+    try:
+        from escalate import escalate_to_human, check_pending_escalation
+        from dispatch_hook import pre_gateway_dispatch
+    except ImportError:
+        import importlib.util
+        from pathlib import Path
+
+        def _load(modname):
+            spec = importlib.util.spec_from_file_location(
+                modname, Path(__file__).with_name(modname + ".py")
+            )
+            m = importlib.util.module_from_spec(spec)
+            spec.loader.exec_module(m)
+            return m
+
+        _esc = _load("escalate")
+        _dh = _load("dispatch_hook")
+        escalate_to_human = _esc.escalate_to_human
+        check_pending_escalation = _esc.check_pending_escalation
+        pre_gateway_dispatch = _dh.pre_gateway_dispatch
+
+    ctx.register_tool(
+        name="escalate_to_human",
+        toolset="escalation",
+        schema=_ESCALATE_SCHEMA,
+        handler=escalate_to_human,
+        description="Escalate to the human over chat and wait for their reply.",
+        emoji="🙋",
+    )
+    ctx.register_tool(
+        name="check_pending_escalation",
+        toolset="escalation",
+        schema=_CHECK_SCHEMA,
+        handler=check_pending_escalation,
+        description="Retrieve and clear a pending escalation when resuming.",
+        emoji="📥",
+    )
+    ctx.register_hook("pre_gateway_dispatch", pre_gateway_dispatch)
+    logger.info("human_escalation: registered escalate_to_human + check_pending_escalation + resume hook")
diff --git a/infra/templates/plugins/human_escalation/dispatch_hook.py b/infra/templates/plugins/human_escalation/dispatch_hook.py
new file mode 100644
index 0000000..c3896a6
--- /dev/null
+++ b/infra/templates/plugins/human_escalation/dispatch_hook.py
@@ -0,0 +1,91 @@
+"""pre_gateway_dispatch hook: when a reply arrives for a conversation with a pending
+escalation, rewrite the inbound text to nudge the agent to resume the checkout.
+
+Zero-divergence resume mechanism -- lives entirely in the plugin, patches no gateway
+source. Registered against the ``pre_gateway_dispatch`` hook (hermes_cli/plugins.py),
+which fires once per inbound MessageEvent before auth/dispatch with kwargs
+``event``, ``gateway``, ``session_store``. Returning ``{"action":"rewrite","text":...}``
+replaces ``event.text``; returning ``None`` means normal flow.
+
+Runtime event shape (gateway/platforms/base.py::MessageEvent +
+gateway/session.py::SessionSource):
+
+    event.text                    -> str (message body)
+    event.source                  -> SessionSource
+    event.source.platform         -> Platform enum; ``.value`` is the platform string
+    event.source.chat_id          -> str
+
+The EscalationStore is keyed ``platform:chat_id`` where ``platform`` is the
+enum's ``.value`` string (gateway/run.py::_set_session_env stores
+``context.source.platform.value``), so we read ``source.platform.value`` here to
+match the key written by escalate_to_human.
+"""
+
+import os
+
+try:
+    # Preferred: package-style import when the plugin dir is on sys.path
+    # (runtime) or registered as a package in sys.modules (some harnesses).
+    from store import EscalationStore  # type: ignore[assignment]
+except ModuleNotFoundError:
+    # Fallback: load store.py relative to this file via importlib. Handles the
+    # importlib.util.spec_from_file_location + submodule_search_locations test
+    # pattern where the plugin dir is not on sys.path but __file__ is set.
+    import importlib.util as _ilu
+    _store_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "store.py")
+    _store_spec = _ilu.spec_from_file_location("_he_store", _store_path)
+    _store_mod = _ilu.module_from_spec(_store_spec)
+    _store_spec.loader.exec_module(_store_mod)
+    EscalationStore = _store_mod.EscalationStore
+
+
+def _platform_value(source) -> str:
+    """Return the platform string for a SessionSource.
+
+    ``source.platform`` is a ``Platform`` enum at runtime; its ``.value`` is the
+    string used as the EscalationStore key. Tolerates a plain-string platform
+    (synthetic/future events) by falling back to ``str``.
+    """
+    platform = getattr(source, "platform", None)
+    if platform is None:
+        return ""
+    value = getattr(platform, "value", None)
+    if value:
+        return str(value)
+    return str(platform) if platform else ""
+
+
+def _conversation(event):
+    """Resolve (platform, chat_id) from the inbound event.
+
+    The real MessageEvent carries these on a nested ``source`` (SessionSource).
+    Returns ("", "") safely when the shape is unexpected so the hook no-ops
+    rather than raising inside the gateway dispatch path.
+    """
+    source = getattr(event, "source", None)
+    if source is None:
+        return "", ""
+    platform = _platform_value(source)
+    chat_id = getattr(source, "chat_id", None) or ""
+    return platform, str(chat_id) if chat_id else ""
+
+
+def pre_gateway_dispatch(event=None, gateway=None, session_store=None, **_kw):
+    if event is None:
+        return None
+
+    platform, chat_id = _conversation(event)
+    if not platform or not chat_id:
+        return None
+
+    rec = EscalationStore().get_active(platform, chat_id)
+    if rec is None:
+        return None
+
+    text = getattr(event, "text", "") or ""
+    prefix = (
+        f"[RESUME CHECKOUT] You previously asked the user: \"{rec['prompt']}\". "
+        f"Their reply follows. First call check_pending_escalation to clear the pending "
+        f"state, then continue the checkout skill using their answer below.\n\n"
+    )
+    return {"action": "rewrite", "text": prefix + text}
diff --git a/infra/templates/plugins/human_escalation/escalate.py b/infra/templates/plugins/human_escalation/escalate.py
new file mode 100644
index 0000000..f8551aa
--- /dev/null
+++ b/infra/templates/plugins/human_escalation/escalate.py
@@ -0,0 +1,180 @@
+"""Chat-send helpers for escalate_to_human + check_pending_escalation handlers.
+
+Approach A' (no-patch async): escalate_to_human posts a prompt to the user's chat
+and RETURNS -- the turn ends. The reply resolves via check_pending_escalation on the
+next turn, nudged by the pre_gateway_dispatch hook (dispatch_hook.py).
+
+This module intentionally contains ONLY the three primitive helpers. The tool
+handlers (escalate_to_human, check_pending_escalation) live in Tasks 3 and 4.
+"""
+
+import json
+import os
+from typing import Optional
+
+try:
+    from gateway.session_context import get_session_env
+except ImportError:  # outside the hermes runtime (unit tests)
+    def get_session_env(name: str, default: str = "") -> str:
+        return os.environ.get(name, default)
+
+try:
+    # Preferred: package-style import when the plugin dir is a proper package
+    # on sys.path (runtime) or when the module is registered in sys.modules
+    # as a package (some test harnesses do this).
+    from store import EscalationStore  # type: ignore[assignment]
+except ModuleNotFoundError:
+    # Fallback: load store.py relative to this file via importlib. This handles
+    # the importlib.util.spec_from_file_location + submodule_search_locations
+    # test pattern where the plugin dir is not on sys.path but __file__ is set.
+    import importlib.util as _ilu
+    import sys as _sys
+    _store_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "store.py")
+    _store_spec = _ilu.spec_from_file_location("_he_store", _store_path)
+    _store_mod = _ilu.module_from_spec(_store_spec)
+    _store_spec.loader.exec_module(_store_mod)
+    EscalationStore = _store_mod.EscalationStore
+
+
+def _session_target() -> str:
+    """Return the send_message target string for the active session.
+
+    Format: ``platform:chat_id`` or ``platform:chat_id:thread_id`` when a
+    thread is present. Returns an empty string when platform or chat_id are
+    not available in the session environment (e.g. CLI context).
+    """
+    platform = get_session_env("HERMES_SESSION_PLATFORM", "")
+    chat_id = get_session_env("HERMES_SESSION_CHAT_ID", "")
+    thread_id = get_session_env("HERMES_SESSION_THREAD_ID", "")
+    if not platform or not chat_id:
+        return ""
+    target = f"{platform}:{chat_id}"
+    if thread_id:
+        target = f"{target}:{thread_id}"
+    return target
+
+
+def _dispatch_send_message(target: str, message: str) -> str:
+    """Dispatch the built-in send_message tool directly.
+
+    Implementation note: ``send_message_tool`` (tools/send_message_tool.py,
+    line 150) is a plain synchronous function with signature
+    ``send_message_tool(args, **kw) -> str``. It accepts ``action``, ``target``,
+    and ``message`` in the args dict and returns a JSON string. Calling it
+    directly avoids needing the tool to be registered in the registry (which
+    requires full agent startup) while remaining fully correct at runtime.
+
+    This function is isolated so tests can monkeypatch it without importing any
+    gateway code.
+    """
+    try:
+        from tools.send_message_tool import send_message_tool
+    except ImportError as exc:
+        raise RuntimeError(
+            "_dispatch_send_message: tools.send_message_tool is unavailable in this runtime context"
+        ) from exc
+    return send_message_tool({"action": "send", "target": target, "message": message})
+
+
+def _send_to_chat(target: str, message: str) -> str:
+    """Send *message* to *target* via the built-in send_message tool."""
+    return _dispatch_send_message(target, message)
+
+
+_DEFAULT_TIMEOUT_S = 300
+
+_RESUME_INSTRUCTION = (
+    "I have sent your request to the user and recorded a pending escalation. "
+    "END YOUR TURN NOW and wait for their reply -- do not poll or loop. When they "
+    "respond, call check_pending_escalation to retrieve and clear it, then continue "
+    "from where you left off."
+)
+
+
+def _render_message(kind: str, prompt: str, payload: Optional[dict]) -> str:
+    payload = payload or {}
+    parts = [prompt]
+    if kind == "confirmation":
+        items = payload.get("line_items") or []
+        if items:
+            parts.append("")
+            parts.extend(f"  • {it}" for it in items)
+        total = payload.get("total")
+        if total:
+            parts.append("")
+            parts.append(f"TOTAL: {total}")
+        parts.append("")
+        parts.append("Reply YES to confirm and charge, or anything else to cancel.")
+    elif kind == "link_handoff":
+        url = payload.get("url")
+        if url:
+            parts.append("")
+            parts.append(url)
+        media = payload.get("media_path")
+        if media:
+            parts.append(f"MEDIA:{media}")
+        parts.append("")
+        parts.append("Reply DONE when finished.")
+    elif kind == "code_request":
+        parts.append("")
+        parts.append("Reply with the code.")
+    return "\n".join(parts)
+
+
+def escalate_to_human(args: dict, **_kw) -> str:
+    kind = (args.get("kind") or "freeform").strip()
+    prompt = args.get("prompt") or ""
+    payload = args.get("payload") or {}
+    try:
+        timeout_s = int(args.get("timeout_s") or _DEFAULT_TIMEOUT_S)
+    except (ValueError, TypeError):
+        timeout_s = _DEFAULT_TIMEOUT_S
+
+    if kind not in {"code_request", "confirmation", "link_handoff", "freeform"}:
+        return json.dumps({"error": f"unknown kind: {kind}"})
+    if not prompt:
+        return json.dumps({"error": "prompt is required"})
+
+    target = _session_target()
+    if not target:
+        return json.dumps({"error": "no active chat session (HERMES_SESSION_PLATFORM/CHAT_ID unset)"})
+
+    platform = get_session_env("HERMES_SESSION_PLATFORM", "")
+    chat_id = get_session_env("HERMES_SESSION_CHAT_ID", "")
+    user_id = get_session_env("HERMES_SESSION_USER_ID", "")
+
+    store = EscalationStore()
+    escal_id = store.create(platform, chat_id, user_id, kind, prompt, timeout_s=timeout_s)
+
+    message = _render_message(kind, prompt, payload)
+    send_result = _send_to_chat(target, message)
+
+    return json.dumps({
+        "status": "awaiting",
+        "escal_id": escal_id,
+        "kind": kind,
+        "instruction": _RESUME_INSTRUCTION,
+        "send_result": send_result,
+    })
+
+
+def check_pending_escalation(args: dict, **_kw) -> str:
+    platform = get_session_env("HERMES_SESSION_PLATFORM", "")
+    chat_id = get_session_env("HERMES_SESSION_CHAT_ID", "")
+    if not platform or not chat_id:
+        return json.dumps({"status": "none"})
+
+    store = EscalationStore()
+    rec = store.get_active(platform, chat_id)
+    if rec is None:
+        return json.dumps({"status": "none"})
+    store.resolve(platform, chat_id)  # clear it; the user's latest message is the answer
+    return json.dumps({
+        "status": "found",
+        "kind": rec["kind"],
+        "prompt": rec["prompt"],
+        "escal_id": rec["escal_id"],
+        "note": ("The user's most recent message is the answer to this escalation. "
+                 "For kind=confirmation, proceed to submit/charge ONLY if the reply is "
+                 "affirmative (yes/confirm/y); otherwise abort and report nothing was charged."),
+    })
diff --git a/infra/templates/plugins/human_escalation/plugin.yaml b/infra/templates/plugins/human_escalation/plugin.yaml
new file mode 100644
index 0000000..4b8a29b
--- /dev/null
+++ b/infra/templates/plugins/human_escalation/plugin.yaml
@@ -0,0 +1,10 @@
+name: human_escalation
+version: 1.0.0
+description: "Human-in-the-loop escalation bus over chat -- relay codes, confirm charges, hand off links. Used by the checkout skill."
+author: NimbleCo
+kind: standalone
+provides_tools:
+  - escalate_to_human
+  - check_pending_escalation
+provides_hooks:
+  - pre_gateway_dispatch
diff --git a/infra/templates/plugins/human_escalation/store.py b/infra/templates/plugins/human_escalation/store.py
new file mode 100644
index 0000000..240d5c8
--- /dev/null
+++ b/infra/templates/plugins/human_escalation/store.py
@@ -0,0 +1,103 @@
+"""File-based pending-escalation store.
+
+One active escalation per conversation (keyed platform:chat_id). Records persist
+under HERMES_HOME so a process restart mid-wait does not lose the pending state.
+Mirrors the atomic-write + 0600 pattern used by gateway/pairing.py.
+"""
+
+import json
+import os
+import tempfile
+import time
+import uuid
+from pathlib import Path
+from typing import Optional, Dict, Any
+
+try:
+    from hermes_constants import get_hermes_dir
+    _ESCALATION_DIR = get_hermes_dir("escalations", "escalations")
+    _HAS_HERMES_CONSTANTS = True
+except Exception:
+    # Fallback when imported outside the hermes runtime (e.g. unit tests that
+    # only set HERMES_HOME). Keeps the store self-contained and testable.
+    _ESCALATION_DIR = Path(os.environ.get("HERMES_HOME", str(Path.home() / ".hermes"))) / "escalations"
+    _HAS_HERMES_CONSTANTS = False
+
+
+def _key(platform: str, chat_id: str) -> str:
+    return f"{platform}:{chat_id}"
+
+
+class EscalationStore:
+    def __init__(self, base_dir: Optional[Path] = None):
+        if base_dir is not None:
+            self._dir = Path(base_dir)
+        elif not _HAS_HERMES_CONSTANTS and os.environ.get("HERMES_HOME"):
+            # Re-resolve from HERMES_HOME only when hermes_constants was not
+            # importable (e.g. unit tests that set HERMES_HOME for isolation).
+            # When constants ARE available, honor get_hermes_dir's path, which
+            # may legitimately differ from $HERMES_HOME/escalations in prod.
+            self._dir = Path(os.environ["HERMES_HOME"]) / "escalations"
+        else:
+            self._dir = _ESCALATION_DIR
+        self._path = self._dir / "pending.json"
+
+    def _load(self) -> Dict[str, Any]:
+        if self._path.exists():
+            try:
+                return json.loads(self._path.read_text())
+            except (json.JSONDecodeError, OSError):
+                return {}
+        return {}
+
+    def _write(self, data: Dict[str, Any]) -> None:
+        self._dir.mkdir(parents=True, exist_ok=True)
+        fd, tmp = tempfile.mkstemp(dir=str(self._dir), suffix=".tmp")
+        try:
+            with os.fdopen(fd, "w") as f:
+                json.dump(data, f, indent=2)
+            # Restrict perms before the file becomes visible at the target
+            # path — chmod-after-replace leaves a TOCTOU window where the
+            # record (user_id/prompt) is world-readable under a lax umask.
+            os.chmod(tmp, 0o600)
+            os.replace(tmp, self._path)
+        except BaseException:
+            try:
+                os.unlink(tmp)
+            except OSError:
+                pass
+            raise
+
+    def create(self, platform: str, chat_id: str, user_id: str, kind: str,
+               prompt: str, timeout_s: int = 300) -> str:
+        escal_id = uuid.uuid4().hex
+        data = self._load()
+        data[_key(platform, chat_id)] = {
+            "escal_id": escal_id,
+            "platform": platform,
+            "chat_id": chat_id,
+            "user_id": user_id,
+            "kind": kind,
+            "prompt": prompt,
+            "created_at": time.time(),
+            "timeout_s": timeout_s,
+            "status": "pending",
+        }
+        self._write(data)
+        return escal_id
+
+    def get_active(self, platform: str, chat_id: str) -> Optional[Dict[str, Any]]:
+        data = self._load()
+        rec = data.get(_key(platform, chat_id))
+        if not rec or rec.get("status") != "pending":
+            return None
+        if time.time() - rec["created_at"] > rec["timeout_s"]:
+            return None
+        return rec
+
+    def resolve(self, platform: str, chat_id: str) -> Optional[Dict[str, Any]]:
+        data = self._load()
+        rec = data.pop(_key(platform, chat_id), None)
+        if rec is not None:
+            self._write(data)
+        return rec
diff --git a/infra/templates/plugins/human_escalation/test_human_escalation.py b/infra/templates/plugins/human_escalation/test_human_escalation.py
new file mode 100644
index 0000000..56da429
--- /dev/null
+++ b/infra/templates/plugins/human_escalation/test_human_escalation.py
@@ -0,0 +1,364 @@
+"""Co-located tests for the human_escalation HSM plugin.
+
+Consolidates the four source suites (store / escalate / hook / integration)
+from hermes-agent. In this standalone location the plugin dir IS this test
+file's own directory, so modules are loaded by __file__ via importlib --
+matching the captcha_cascade test convention (no hermes-agent imports needed).
+"""
+
+import importlib.util
+import json
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+# The plugin dir is this test file's own directory.
+_PKG = Path(__file__).parent
+_STORE_PATH = _PKG / "store.py"
+
+
+# ---------------------------------------------------------------------------
+# store.py
+# ---------------------------------------------------------------------------
+
+
+def _load_store(monkeypatch, tmp_path):
+    """Import store.py fresh with HERMES_HOME pointed at a temp dir."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    spec = importlib.util.spec_from_file_location("he_store_under_test", _STORE_PATH)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod.EscalationStore()
+
+
+def test_create_then_get_active_returns_record(monkeypatch, tmp_path):
+    store = _load_store(monkeypatch, tmp_path)
+    escal_id = store.create("signal", "chat1", "user1", "code_request",
+                            "Reply with the SMS code", timeout_s=300)
+    assert escal_id
+    rec = store.get_active("signal", "chat1")
+    assert rec is not None
+    assert rec["kind"] == "code_request"
+    assert rec["prompt"] == "Reply with the SMS code"
+    assert rec["status"] == "pending"
+
+
+def test_get_active_none_when_no_record(monkeypatch, tmp_path):
+    store = _load_store(monkeypatch, tmp_path)
+    assert store.get_active("signal", "nope") is None
+
+
+def test_expired_record_is_not_active(monkeypatch, tmp_path):
+    store = _load_store(monkeypatch, tmp_path)
+    store.create("signal", "chat1", "user1", "confirmation", "Confirm $5", timeout_s=1)
+    data = store._load()
+    data["signal:chat1"]["created_at"] = time.time() - 10
+    store._write(data)
+    assert store.get_active("signal", "chat1") is None
+
+
+def test_resolve_clears_record_and_returns_it(monkeypatch, tmp_path):
+    store = _load_store(monkeypatch, tmp_path)
+    store.create("signal", "chat1", "user1", "code_request", "code?", timeout_s=300)
+    rec = store.resolve("signal", "chat1")
+    assert rec is not None and rec["kind"] == "code_request"
+    assert store.get_active("signal", "chat1") is None
+
+
+def test_new_create_overwrites_prior_active_for_same_conversation(monkeypatch, tmp_path):
+    store = _load_store(monkeypatch, tmp_path)
+    store.create("signal", "chat1", "user1", "code_request", "first", timeout_s=300)
+    store.create("signal", "chat1", "user1", "confirmation", "second", timeout_s=300)
+    rec = store.get_active("signal", "chat1")
+    assert rec["prompt"] == "second"
+
+
+# ---------------------------------------------------------------------------
+# escalate.py
+# ---------------------------------------------------------------------------
+
+
+def _load_escalate(monkeypatch, tmp_path):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    spec = importlib.util.spec_from_file_location(
+        "he_escalate_under_test", _PKG / "escalate.py",
+        submodule_search_locations=[str(_PKG)],
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+class TestSessionTarget:
+    def test_target_from_platform_and_chat(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        monkeypatch.setenv("HERMES_SESSION_PLATFORM", "signal")
+        monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "group:abc")
+        monkeypatch.delenv("HERMES_SESSION_THREAD_ID", raising=False)
+        assert mod._session_target() == "signal:group:abc"
+
+    def test_target_includes_thread_when_present(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        monkeypatch.setenv("HERMES_SESSION_PLATFORM", "telegram")
+        monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "-100123")
+        monkeypatch.setenv("HERMES_SESSION_THREAD_ID", "17")
+        assert mod._session_target() == "telegram:-100123:17"
+
+    def test_target_empty_when_no_session(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        for k in ("HERMES_SESSION_PLATFORM", "HERMES_SESSION_CHAT_ID"):
+            monkeypatch.delenv(k, raising=False)
+        assert mod._session_target() == ""
+
+
+class TestSendToChat:
+    def test_send_dispatches_send_message(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        calls = {}
+        monkeypatch.setattr(mod, "_dispatch_send_message",
+                            lambda target, message: calls.update(target=target, message=message) or '{"ok":true}')
+        out = mod._send_to_chat("signal:c1", "hello")
+        assert calls["target"] == "signal:c1"
+        assert calls["message"] == "hello"
+        assert out == '{"ok":true}'
+
+
+class TestEscalateToHuman:
+    def _prep(self, mod, monkeypatch):
+        monkeypatch.setenv("HERMES_SESSION_PLATFORM", "signal")
+        monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "c1")
+        monkeypatch.setenv("HERMES_SESSION_USER_ID", "u1")
+        sent = {}
+        monkeypatch.setattr(mod, "_send_to_chat",
+                            lambda target, message: sent.update(target=target, message=message) or '{"ok":true}')
+        return sent
+
+    def test_code_request_writes_record_and_sends(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        sent = self._prep(mod, monkeypatch)
+        out = json.loads(mod.escalate_to_human({"kind": "code_request", "prompt": "Reply with the SMS code"}))
+        assert out["status"] == "awaiting"
+        assert out["escal_id"]
+        assert "SMS code" in sent["message"]
+        rec = mod.EscalationStore().get_active("signal", "c1")
+        assert rec["kind"] == "code_request"
+
+    def test_confirmation_renders_line_items_and_total(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        sent = self._prep(mod, monkeypatch)
+        mod.escalate_to_human({
+            "kind": "confirmation",
+            "prompt": "Confirm this purchase",
+            "payload": {"line_items": ["2x GA"], "total": "$94.50"},
+        })
+        assert "2x GA" in sent["message"]
+        assert "$94.50" in sent["message"]
+
+    def test_link_handoff_includes_url(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        sent = self._prep(mod, monkeypatch)
+        mod.escalate_to_human({
+            "kind": "link_handoff",
+            "prompt": "Finish this step here",
+            "payload": {"url": "https://vnc.example/x"},
+        })
+        assert "https://vnc.example/x" in sent["message"]
+
+    def test_errors_when_no_session(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        for k in ("HERMES_SESSION_PLATFORM", "HERMES_SESSION_CHAT_ID"):
+            monkeypatch.delenv(k, raising=False)
+        out = json.loads(mod.escalate_to_human({"kind": "freeform", "prompt": "hi"}))
+        assert "error" in out
+
+    def test_non_numeric_timeout_falls_back_to_default(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        self._prep(mod, monkeypatch)
+        out = json.loads(mod.escalate_to_human({"kind": "freeform", "prompt": "hi", "timeout_s": "fast"}))
+        assert out["status"] == "awaiting"  # did not crash
+        rec = mod.EscalationStore().get_active("signal", "c1")
+        assert rec["timeout_s"] == 300
+
+
+class TestCheckPending:
+    def test_returns_found_and_clears(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        monkeypatch.setenv("HERMES_SESSION_PLATFORM", "signal")
+        monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "c1")
+        mod.EscalationStore().create("signal", "c1", "u1", "code_request", "code?", timeout_s=300)
+        out = json.loads(mod.check_pending_escalation({}))
+        assert out["status"] == "found"
+        assert out["kind"] == "code_request"
+        out2 = json.loads(mod.check_pending_escalation({}))
+        assert out2["status"] == "none"
+
+    def test_none_when_nothing_pending(self, monkeypatch, tmp_path):
+        mod = _load_escalate(monkeypatch, tmp_path)
+        monkeypatch.setenv("HERMES_SESSION_PLATFORM", "signal")
+        monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "c1")
+        out = json.loads(mod.check_pending_escalation({}))
+        assert out["status"] == "none"
+
+
+# ---------------------------------------------------------------------------
+# dispatch_hook.py
+# ---------------------------------------------------------------------------
+
+
+def _load_hook(monkeypatch, tmp_path):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    spec = importlib.util.spec_from_file_location(
+        "he_hook_under_test", _PKG / "dispatch_hook.py",
+        submodule_search_locations=[str(_PKG)],
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def _event(platform="signal", chat_id="c1", text="123456"):
+    """Mirror the real gateway MessageEvent shape: platform/chat_id live on a
+    nested ``source`` (SessionSource), and ``source.platform`` is an enum whose
+    ``.value`` is the platform string used as the EscalationStore key.
+    """
+    source = SimpleNamespace(
+        platform=SimpleNamespace(value=platform),
+        chat_id=chat_id,
+    )
+    return SimpleNamespace(source=source, text=text)
+
+
+def test_no_pending_returns_none(monkeypatch, tmp_path):
+    mod = _load_hook(monkeypatch, tmp_path)
+    assert mod.pre_gateway_dispatch(event=_event(), gateway=None, session_store=None) is None
+
+
+def test_pending_rewrites_text_with_resume_prefix(monkeypatch, tmp_path):
+    mod = _load_hook(monkeypatch, tmp_path)
+    mod.EscalationStore().create("signal", "c1", "u1", "code_request",
+                                 "Reply with the SMS code", timeout_s=300)
+    out = mod.pre_gateway_dispatch(event=_event(text="123456"), gateway=None, session_store=None)
+    assert out is not None and out["action"] == "rewrite"
+    assert "123456" in out["text"]
+    assert "check_pending_escalation" in out["text"]
+    assert "Reply with the SMS code" in out["text"]
+
+
+def test_pending_for_other_conversation_is_ignored(monkeypatch, tmp_path):
+    mod = _load_hook(monkeypatch, tmp_path)
+    mod.EscalationStore().create("signal", "OTHER", "u1", "code_request", "x", timeout_s=300)
+    assert mod.pre_gateway_dispatch(event=_event(chat_id="c1"), gateway=None, session_store=None) is None
+
+
+def test_none_event_returns_none(monkeypatch, tmp_path):
+    mod = _load_hook(monkeypatch, tmp_path)
+    assert mod.pre_gateway_dispatch(event=None, gateway=None, session_store=None) is None
+
+
+def test_missing_source_returns_none(monkeypatch, tmp_path):
+    mod = _load_hook(monkeypatch, tmp_path)
+    mod.EscalationStore().create("signal", "c1", "u1", "code_request", "x", timeout_s=300)
+    bad = SimpleNamespace(text="hi")  # no .source attribute
+    assert mod.pre_gateway_dispatch(event=bad, gateway=None, session_store=None) is None
+
+
+def test_plain_string_platform_also_supported(monkeypatch, tmp_path):
+    """Defensive: if a future/synthetic event carries a plain string platform
+    (no ``.value``), the hook should still resolve it correctly."""
+    mod = _load_hook(monkeypatch, tmp_path)
+    mod.EscalationStore().create("signal", "c1", "u1", "code_request",
+                                 "Reply with the SMS code", timeout_s=300)
+    source = SimpleNamespace(platform="signal", chat_id="c1")
+    event = SimpleNamespace(source=source, text="123456")
+    out = mod.pre_gateway_dispatch(event=event, gateway=None, session_store=None)
+    assert out is not None and out["action"] == "rewrite"
+    assert "123456" in out["text"]
+
+
+# ---------------------------------------------------------------------------
+# integration: full A' cycle (escalate -> hook rewrite -> check_pending)
+# ---------------------------------------------------------------------------
+
+
+def _load(name, file):
+    spec = importlib.util.spec_from_file_location(name, _PKG / file,
+                                                  submodule_search_locations=[str(_PKG)])
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def _reply_event(platform="signal", chat_id="c1", text="445566"):
+    # Mirror the real MessageEvent shape the hook reads: event.source.platform.value / .chat_id
+    return SimpleNamespace(
+        text=text,
+        source=SimpleNamespace(platform=SimpleNamespace(value=platform), chat_id=chat_id),
+    )
+
+
+def test_full_escalate_reply_resume_cycle(monkeypatch, tmp_path):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setenv("HERMES_SESSION_PLATFORM", "signal")
+    monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "c1")
+    monkeypatch.setenv("HERMES_SESSION_USER_ID", "u1")
+
+    escalate = _load("he_e", "escalate.py")
+    hook = _load("he_h", "dispatch_hook.py")
+    monkeypatch.setattr(escalate, "_send_to_chat", lambda target, message: '{"ok":true}')
+
+    # Confirm both separately-loaded modules share the same pending.json path.
+    e_path = escalate.EscalationStore()._path
+    h_path = hook.EscalationStore()._path
+    assert e_path == h_path, (
+        f"Store-path mismatch: escalate resolves {e_path}, hook resolves {h_path}. "
+        "The cross-module store-path is inconsistent — the plugin's store path is "
+        "unstable in the real runtime too."
+    )
+
+    # 1. Agent escalates for a code.
+    out = json.loads(escalate.escalate_to_human({"kind": "code_request", "prompt": "Reply with the SMS code"}))
+    assert out["status"] == "awaiting"
+
+    # 2. User replies "445566" -- hook rewrites it with a resume nudge.
+    ev = _reply_event(text="445566")
+    rewritten = hook.pre_gateway_dispatch(event=ev, gateway=None, session_store=None)
+    assert rewritten is not None and rewritten["action"] == "rewrite"
+    assert "445566" in rewritten["text"]
+    assert "check_pending_escalation" in rewritten["text"]
+
+    # 3. Resumed turn: check_pending_escalation returns the code request and clears it.
+    found = json.loads(escalate.check_pending_escalation({}))
+    assert found["status"] == "found"
+    assert found["kind"] == "code_request"
+
+    # 4. Idempotent: a second check finds nothing pending.
+    assert json.loads(escalate.check_pending_escalation({}))["status"] == "none"
+
+
+def test_confirmation_decline_path_records_then_resolves(monkeypatch, tmp_path):
+    """A confirmation escalation is recorded, the hook nudges on reply, and check_pending
+    returns kind=confirmation so the skill can enforce the affirmative-only gate."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setenv("HERMES_SESSION_PLATFORM", "signal")
+    monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "c1")
+    monkeypatch.setenv("HERMES_SESSION_USER_ID", "u1")
+
+    escalate = _load("he_e2", "escalate.py")
+    hook = _load("he_h2", "dispatch_hook.py")
+    monkeypatch.setattr(escalate, "_send_to_chat", lambda target, message: '{"ok":true}')
+
+    out = json.loads(escalate.escalate_to_human({
+        "kind": "confirmation", "prompt": "Confirm this purchase",
+        "payload": {"line_items": ["2x GA"], "total": "$94.50"},
+    }))
+    assert out["status"] == "awaiting" and out["kind"] == "confirmation"
+
+    ev = _reply_event(text="no")
+    rewritten = hook.pre_gateway_dispatch(event=ev, gateway=None, session_store=None)
+    assert rewritten["action"] == "rewrite" and "no" in rewritten["text"]
+
+    found = json.loads(escalate.check_pending_escalation({}))
+    assert found["status"] == "found" and found["kind"] == "confirmation"
+    # The skill is responsible for interpreting "no" as non-affirmative -> abort.
diff --git a/infra/templates/skills/checkout/SKILL.md b/infra/templates/skills/checkout/SKILL.md
new file mode 100644
index 0000000..92c562b
--- /dev/null
+++ b/infra/templates/skills/checkout/SKILL.md
@@ -0,0 +1,66 @@
+---
+name: checkout
+description: Use when asked to buy or purchase something online (tickets, products, bookings) where the flow may need a human for a captcha, a 2FA/SMS code, or a final charge confirmation. Drives an online checkout end-to-end with human-in-the-loop escalation over chat.
+version: 1.0.0
+author: NimbleCo
+license: MIT
+metadata:
+  hermes:
+    tags: [purchase, checkout, browser, human-in-the-loop, tickets]
+    related_skills: [captcha-cascade]
+---
+
+# Checkout — human-in-the-loop purchasing
+
+Drive an online purchase to completion using the Camofox browser tools, escalating to
+the human over chat whenever you hit something you cannot (or must not) do alone.
+
+## Resume check — DO THIS FIRST, EVERY TURN
+
+Before anything else, call `check_pending_escalation`.
+- If it returns `status: "found"`, you are resuming. The user's most recent message is the
+  answer to `prompt`. Apply it:
+  - `code_request` → the message is the code; type it into the page and continue.
+  - `confirmation` → continue to submit/charge ONLY if the reply is affirmative (yes / confirm
+    / y / yep). Anything else → abort: do not submit, tell the user nothing was charged.
+  - `link_handoff` → an affirmative ("done") means the user finished the step; re-read the
+    page and continue.
+- Re-orient by reading the live browser tab (it is still open from before). Use the browser
+  read tools to see the current page state, then proceed from where you left off.
+- If it returns `status: "none"`, this is a fresh request — start at step 1.
+
+## The flow
+
+1. **Locate** the item/event (navigate, search).
+2. **Select** options — quantity, tier, seats, date.
+3. **Fill** purchaser details. Retrieve stored details/payment info via the approved
+   credential path (the swarm-tool proxy / `~/.hermes` credential files) — NEVER hardcode or
+   ask the user to paste raw card numbers in chat.
+4. **Captcha?** Use the `captcha_solve` tool (captcha-cascade). If it cannot solve and returns
+   a VNC URL, escalate it: `escalate_to_human(kind="link_handoff", prompt="Please solve the
+   captcha here, then reply DONE", payload={"url": <vnc_url>})` and END YOUR TURN.
+5. **Need an SMS / 2FA / emailed code?** `escalate_to_human(kind="code_request",
+   prompt="Reply with the verification code <site> just sent you")` and END YOUR TURN.
+6. **Reach the order review.** Read the exact line items and total from the page.
+7. **Confirm the charge — ALWAYS.** `escalate_to_human(kind="confirmation", prompt="Confirm
+   this purchase", payload={"line_items": [...], "total": "$NN.NN"})` and END YOUR TURN.
+8. **Submit only on an affirmative confirmation.** Never click final pay/submit without having
+   received `kind=confirmation` answered affirmatively this purchase.
+9. **Report** the outcome: success + confirmation number, or the reason it stopped.
+
+## Hard rules
+
+- **Never submit payment without an affirmative `confirmation` escalation for THIS purchase.**
+  This is non-negotiable — it is the one guaranteed human gate.
+- After any `escalate_to_human` call, END YOUR TURN. Do not poll, sleep, or loop waiting for a
+  reply — the reply arrives as a new message and you resume via the Resume check above.
+- On timeout (the user does not reply and a later turn finds no pending escalation but an
+  unfinished checkout), report honestly: "timed out waiting for you — nothing was charged."
+- Never expose raw credentials or full card numbers in chat.
+
+## Moshtix (worked example, instance #1)
+
+Moshtix ticketing is the first target. Capture concrete selectors/flow notes here as you learn
+them (event search → ticket-type select → quantity → checkout → captcha → code → review →
+confirm). Treat anything site-specific as notes under this section; the flow above is the
+reusable core.