From c28866fa87c13eefff087b18ed5364dbecb1e048 Mon Sep 17 00:00:00 2001
From: Wesley Matos <wesleymatosdev@gmail.com>
Date: Tue, 1 Sep 2026 04:55:37 -0300
Subject: [PATCH 1/2] fix(hooks): register configured hooks in TUI gateway and
 serve backends
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Every long-lived agent runtime must register the user's shell hooks and
outbound webhooks at startup. The CLI (_prepare_agent_startup) and the
messaging gateway (gateway/run.py) both do; three backends did neither:

* tui_gateway.entry.main — 'hermes --tui' stdio backend (and the
  dashboard chat PTY, which spawns the same entry point)
* tui_gateway.ws.handle_ws — the dashboard / desktop WebSocket sidecar
* hermes_cli.web_server._lifespan — 'hermes serve' / dashboard backend

Net effect: a webhook configured in config.yaml fired in 'hermes --cli'
but silently never fired from a TUI, dashboard, or desktop session —
including the on_session_end event integration points rely on for
completion signaling. Nine open PRs (#67084, #78805, #81409, #83997,
whole class instead.

Add agent/hook_registration.ensure_hooks_registered() — once-per-process
guard, shell-hook consent semantics unchanged (flag / env / config
opt-in, fail-closed on non-TTY stdin), outbound registration unchanged,
fail-soft: a broken hook config never takes down a backend. Wire it into
all three backends. The two inline call sites in main.py and gateway/run.py
behave identically and can migrate later without behavior change.

Regression coverage in tests/tui_gateway/test_hook_registration_parity.py:
registration content, explicit-cfg path, once-per-process guard, failure
isolation, and all three entry-point wirings (7 tests).
---
 agent/hook_registration.py                    |  72 +++++++
 hermes_cli/web_server.py                      |  13 ++
 .../test_hook_registration_parity.py          | 203 ++++++++++++++++++
 tui_gateway/entry.py                          |  11 +
 tui_gateway/server.py                         |  23 ++
 tui_gateway/ws.py                             |   9 +
 6 files changed, 331 insertions(+)
 create mode 100644 agent/hook_registration.py
 create mode 100644 tests/tui_gateway/test_hook_registration_parity.py

diff --git a/agent/hook_registration.py b/agent/hook_registration.py
new file mode 100644
index 000000000000..6fdb9964739e
--- /dev/null
+++ b/agent/hook_registration.py
@@ -0,0 +1,72 @@
+"""Once-per-process registration of user-configured hooks.
+
+Every long-lived agent runtime must register the user's shell hooks and
+outbound webhooks at startup, or events configured in config.yaml silently
+never fire for sessions driven through that backend. The call sites:
+
+* ``hermes --cli`` / oneshot — ``hermes_cli.main._prepare_agent_startup``
+  (registers inline, predates this module)
+* messaging gateway — ``gateway/run.py`` (registers inline)
+* TUI stdio backend — ``tui_gateway.entry.main`` → ``tui_gateway.server``
+* TUI WebSocket sidecar (dashboard chat / desktop) —
+  ``tui_gateway.ws.handle_ws`` → ``tui_gateway.server``
+* ``hermes serve`` / dashboard backend — ``hermes_cli.web_server._lifespan``
+
+The two inline call sites predate this module and behave identically; they
+can migrate to :func:`ensure_hooks_registered` later without behavior
+change. Consent semantics are owned by ``agent.shell_hooks`` (flag / env /
+config opt-in, fail-closed on non-TTY stdin) and neither helper ever
+prompts on a backend's piped stdio. Both registrations are idempotent and
+fail-soft: a broken hook config must never take down a backend.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+
+logger = logging.getLogger(__name__)
+
+_ensured_lock = threading.Lock()
+_ensured = False
+
+
+def reset_for_tests() -> None:
+    """Clear the once-per-process guard (test isolation only)."""
+    global _ensured
+    with _ensured_lock:
+        _ensured = False
+
+
+def ensure_hooks_registered(cfg=None, *, accept_hooks: bool = False) -> None:
+    """Register shell hooks + outbound webhooks exactly once per process.
+
+    *cfg* defaults to a fresh ``hermes_cli.config.load_config()`` read.
+    *accept_hooks* is passed through to shell-hook registration — callers
+    that own a CLI consent flag pass it; backend entry points keep the
+    default ``False`` and let the helper resolve opt-in from env/config.
+
+    Never raises. Repeat calls are no-ops (the underlying registrations
+    are independently idempotent too, so a caller that must bypass the
+    guard can invoke ``agent.shell_hooks`` / ``agent.outbound_webhooks``
+    directly, as the CLI and gateway already do).
+    """
+    global _ensured
+    with _ensured_lock:
+        if _ensured:
+            return
+        _ensured = True
+    try:
+        if cfg is None:
+            from hermes_cli.config import load_config
+
+            cfg = load_config()
+        from agent import outbound_webhooks, shell_hooks
+
+        shell_hooks.register_from_config(cfg, accept_hooks=accept_hooks)
+        outbound_webhooks.register_from_config(cfg)
+    except Exception:
+        logger.debug(
+            "shell-hook / outbound-webhook registration failed at startup",
+            exc_info=True,
+        )
\ No newline at end of file
diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py
index d95f8182a696..d3769fca70f0 100644
--- a/hermes_cli/web_server.py
+++ b/hermes_cli/web_server.py
@@ -167,6 +167,19 @@ async def _lifespan(app: "FastAPI"):
 
     record_boot_fingerprint()
 
+    # Shell-hook / outbound-webhook registration — parity with the CLI
+    # (hermes_cli.main._prepare_agent_startup) and the messaging gateway
+    # (gateway/run.py). Once-per-process via agent.hook_registration;
+    # consent and failure semantics live inside. Without this, hooks
+    # configured in config.yaml fired on --cli but never for dashboard /
+    # Desktop sessions driven through this backend.
+    try:
+        from agent.hook_registration import ensure_hooks_registered
+
+        ensure_hooks_registered()
+    except Exception:
+        _log.warning("hook registration failed at serve startup", exc_info=True)
+
     # Hosted Bot rooms belong to the backend process. Recovery may need a
     # contended state.db migration, so keep it off the pre-yield path: Group
     # Chat must degrade on its own rather than block every Desktop feature.
diff --git a/tests/tui_gateway/test_hook_registration_parity.py b/tests/tui_gateway/test_hook_registration_parity.py
new file mode 100644
index 000000000000..4fad6c84512f
--- /dev/null
+++ b/tests/tui_gateway/test_hook_registration_parity.py
@@ -0,0 +1,203 @@
+"""Regression tests for backend hook-registration parity.
+
+Every long-lived agent runtime must register user-configured shell hooks
+and outbound webhooks at startup. The CLI (``_prepare_agent_startup``) and
+the messaging gateway (``gateway/run.py``) always did; the TUI gateway
+backends and the serve/dashboard backend historically did not, so a
+webhook configured in config.yaml fired in ``hermes --cli`` but silently
+never fired from ``hermes --tui``, the dashboard chat PTY, the desktop WS
+sidecar, or ``hermes serve``.
+
+Verifies:
+
+* ``agent.hook_registration.ensure_hooks_registered`` registers both
+  shell hooks and outbound webhooks from the loaded config.
+* It is once-per-process (repeat calls are no-ops).
+* A config/load failure never breaks backend startup.
+* All three backend entry points call it: ``tui_gateway.entry.main``
+  (stdio TUI + dashboard chat PTY), ``tui_gateway.ws.handle_ws``
+  (dashboard / desktop WS sidecar), and ``web_server._lifespan``
+  (serve / dashboard backend).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import io
+
+import pytest
+
+import agent.hook_registration as hook_registration
+
+
+@pytest.fixture(autouse=True)
+def _fresh_hook_registration_guard():
+    """Start every test with a clean once-per-process guard."""
+    hook_registration.reset_for_tests()
+    yield
+    hook_registration.reset_for_tests()
+
+
+class TestEnsureHooksRegistered:
+    def test_registers_both_shell_hooks_and_outbound_webhooks(self, monkeypatch):
+        import agent.outbound_webhooks as ow
+        import agent.shell_hooks as sh
+
+        calls: dict[str, object] = {}
+
+        def _shell(cfg, *, accept_hooks):
+            calls["shell"] = (cfg, accept_hooks)
+            return []
+
+        def _outbound(cfg):
+            calls["outbound"] = cfg
+            return []
+
+        monkeypatch.setattr(sh, "register_from_config", _shell)
+        monkeypatch.setattr(ow, "register_from_config", _outbound)
+
+        cfg = {"hooks": {}}
+        monkeypatch.setattr("hermes_cli.config.load_config", lambda: cfg)
+
+        hook_registration.ensure_hooks_registered()
+
+        # Consent is delegated to the helper's own resolution (env/config),
+        # never force-enabled from a backend.
+        assert calls["shell"] == (cfg, False)
+        assert calls["outbound"] is cfg
+
+    def test_repeat_calls_are_noops(self, monkeypatch):
+        import agent.outbound_webhooks as ow
+        import agent.shell_hooks as sh
+
+        n = {"shell": 0, "outbound": 0}
+        monkeypatch.setattr(
+            sh, "register_from_config",
+            lambda cfg, *, accept_hooks: n.__setitem__("shell", n["shell"] + 1),
+        )
+        monkeypatch.setattr(
+            ow, "register_from_config", lambda cfg: n.__setitem__("outbound", n["outbound"] + 1)
+        )
+
+        hook_registration.ensure_hooks_registered()
+        hook_registration.ensure_hooks_registered()
+        hook_registration.ensure_hooks_registered()
+
+        assert n == {"shell": 1, "outbound": 1}
+
+    def test_failure_does_not_raise(self, monkeypatch):
+        """A broken config read must never take down backend startup."""
+
+        def _boom():
+            raise RuntimeError("malformed config")
+
+        monkeypatch.setattr("hermes_cli.config.load_config", _boom)
+
+        # Must not raise.
+        hook_registration.ensure_hooks_registered()
+
+    def test_explicit_cfg_skips_config_read(self, monkeypatch):
+        """A caller-provided cfg is used as-is (no load_config round trip)."""
+        import agent.outbound_webhooks as ow
+        import agent.shell_hooks as sh
+
+        seen: dict[str, object] = {}
+        monkeypatch.setattr(sh, "register_from_config", lambda cfg, *, accept_hooks: seen.setdefault("shell", cfg))
+        monkeypatch.setattr(ow, "register_from_config", lambda cfg: seen.setdefault("outbound", cfg))
+        monkeypatch.setattr(
+            "hermes_cli.config.load_config",
+            lambda: (_ for _ in ()).throw(AssertionError("load_config must not be called")),
+        )
+
+        cfg = {"hooks": {}}
+        hook_registration.ensure_hooks_registered(cfg)
+        assert seen == {"shell": cfg, "outbound": cfg}
+
+
+class TestEntryPointWiring:
+    """All three backends must call the registration helper at startup —
+    the same wiring pattern the heartbeat refresher and orphan sweep
+    follow."""
+
+    def _stub_entry_main_common(self, monkeypatch):
+        from tui_gateway import entry, server
+
+        monkeypatch.setattr(entry, "_install_sidecar_publisher", lambda: None)
+        monkeypatch.setattr(entry, "ensure_mcp_discovery_started", lambda: None)
+        monkeypatch.setattr(entry, "resolve_skin", lambda: "default")
+        monkeypatch.setattr(entry.server, "_ensure_skin_watcher", lambda: None)
+        monkeypatch.setattr(entry.server, "_schedule_startup_orphan_sweep", lambda: None)
+        monkeypatch.setattr(entry, "_log_exit", lambda reason: None)
+        monkeypatch.setattr(entry, "handle_spurious_eof", lambda *a: False)
+        monkeypatch.setattr(entry, "write_json", lambda _payload: True)
+        monkeypatch.setattr(entry.sys, "stdin", io.StringIO(""))
+
+        import hermes_cli.model_switch as ms
+
+        monkeypatch.setattr(ms, "prewarm_picker_cache_async", lambda: None)
+        return entry, server
+
+    def test_entry_main_registers_hooks(self, monkeypatch):
+        entry, server = self._stub_entry_main_common(monkeypatch)
+
+        started = {"n": 0}
+        monkeypatch.setattr(
+            server, "_register_hooks_from_config",
+            lambda: started.__setitem__("n", started["n"] + 1),
+        )
+
+        entry.main()
+        assert started["n"] == 1
+
+    def test_handle_ws_registers_hooks(self, monkeypatch):
+        from tui_gateway import server
+        from tui_gateway import ws as ws_mod
+
+        started = {"n": 0}
+        monkeypatch.setattr(
+            server, "_register_hooks_from_config",
+            lambda: started.__setitem__("n", started["n"] + 1),
+        )
+        monkeypatch.setattr(server, "resolve_skin", lambda: "default")
+        monkeypatch.setattr(server, "_ensure_skin_watcher", lambda: None)
+        monkeypatch.setattr(server, "register_live_transport", lambda *_a, **_k: None)
+        monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 0)
+
+        class FakeWS:
+            async def accept(self):
+                pass
+
+            async def send_text(self, line):
+                pass
+
+            async def receive_text(self):
+                raise ws_mod._WebSocketDisconnect()
+
+            async def close(self):
+                pass
+
+        asyncio.run(ws_mod.handle_ws(FakeWS()))
+        assert started["n"] == 1
+
+    def test_serve_lifespan_registers_hooks(self, monkeypatch):
+        """The dashboard/serve backend registers hooks during startup."""
+        import hermes_cli.web_server as web_server_mod
+
+        started = {"n": 0}
+
+        def _ensure(*_a, **_k):
+            started["n"] += 1
+
+        # The lifespan does a lazy `from agent.hook_registration import
+        # ensure_hooks_registered` — patch the module attribute the import
+        # resolves against.
+        monkeypatch.setattr(hook_registration, "ensure_hooks_registered", _ensure)
+        # Neutralize the other lifespan startup work we don't assert on.
+        monkeypatch.setattr(web_server_mod, "_warm_gateway_module", lambda: None)
+
+        from fastapi.testclient import TestClient
+
+        with TestClient(web_server_mod.app, raise_server_exceptions=False):
+            pass
+
+        assert started["n"] == 1
\ No newline at end of file
diff --git a/tui_gateway/entry.py b/tui_gateway/entry.py
index 0c032b882cad..eb9e4ead7461 100644
--- a/tui_gateway/entry.py
+++ b/tui_gateway/entry.py
@@ -250,6 +250,17 @@ def main():
             logger.warning("%s failed", what, exc_info=True)
 
     # Backgrounded so a dead MCP server can't freeze startup; _make_agent briefly joins it.
+
+    # Shell-hook / outbound-webhook registration — parity with the CLI
+    # (hermes_cli.main._prepare_agent_startup) and the messaging gateway
+    # (gateway/run.py). Idempotent + once-per-process; consent and failure
+    # semantics live inside. Without this, hooks configured in config.yaml
+    # fired on --cli but silently never fired from --tui sessions.
+    try:
+        server._register_hooks_from_config()
+    except Exception:
+        logger.warning("hook registration failed at TUI gateway startup", exc_info=True)
+
     ensure_mcp_discovery_started()
 
     # change_events: clients demote legacy polls; replay_epoch: WS restart detection.
diff --git a/tui_gateway/server.py b/tui_gateway/server.py
index 99ef6639f111..5335b4944b2e 100644
--- a/tui_gateway/server.py
+++ b/tui_gateway/server.py
@@ -367,6 +367,29 @@ def _loop():
     threading.Thread(target=_loop, daemon=True).start()
 
 
+# Hook-registration parity with the other backend entry points: the CLI
+# (``hermes_cli.main._prepare_agent_startup``) and the messaging gateway
+# (``gateway/run.py``) both register user-configured shell hooks and
+# outbound webhooks at startup. The TUI gateway backends historically did
+# neither, so a webhook configured in config.yaml fired in ``hermes --cli``
+# but silently never fired from ``hermes --tui`` — or from the dashboard
+# chat PTY / desktop WS sidecar, which share these backends.
+#
+# The once-per-process guard, consent semantics (flag / env / config
+# opt-in, fail-closed on non-TTY stdin) and failure isolation live in
+# :mod:`agent.hook_registration`, shared with the serve/dashboard path.
+
+def _register_hooks_from_config() -> None:
+    """Register user shell hooks + outbound webhooks for this backend.
+
+    Thin delegate to ``agent.hook_registration.ensure_hooks_registered``;
+    called from both TUI-gateway entry points (``entry.main`` and
+    ``ws.handle_ws``). Never raises.
+    """
+    from agent.hook_registration import ensure_hooks_registered
+
+    ensure_hooks_registered()
+
 atexit.register(_shutdown_sessions)
 _start_idle_reaper()
 
diff --git a/tui_gateway/ws.py b/tui_gateway/ws.py
index 2f8d52c215a1..30999b5c80c2 100644
--- a/tui_gateway/ws.py
+++ b/tui_gateway/ws.py
@@ -317,6 +317,15 @@ def _error(code: int, message: str, req_id: Any) -> dict:
                 start()
             except Exception:
                 _log.warning("%s failed", what, exc_info=True)
+        # Shell-hook / outbound-webhook registration — parity with the CLI
+        # and messaging-gateway startup paths. Same once-per-process pass;
+        # a stdio TUI that already registered is a no-op here. Without
+        # this, hooks configured in config.yaml never fired for sessions
+        # driven through the dashboard / desktop WS sidecar.
+        try:
+            server._register_hooks_from_config()
+        except Exception:
+            _log.warning("hook registration failed at TUI WS startup", exc_info=True)
         if not ready_ok:
             disconnect_reason = "ready_send_failed"
             send_failures += 1

From c20f708c45f1a58227b39420cb138b8ae13716a0 Mon Sep 17 00:00:00 2001
From: Wesley Matos <wesleymatosdev@gmail.com>
Date: Mon, 28 Sep 2026 12:45:47 -0300
Subject: [PATCH 2/2] Address PR review feedback (#111315)

- Key hook registration per Hermes home (hermes_home_key) instead of a
  process-global guard, so one backend process serving multiple profiles
  registers each profile's hooks independently
- Mark a home complete only after registration succeeds; concurrent
  callers for the same home wait, failures stay retryable
- Register config hooks after plugin discovery (plugin block decisions
  win ties), matching gateway startup ordering
- Register BOTH shell hooks and outbound webhooks in the profile-scoped
  agent build (secondary profiles previously dropped hooks.outbound)
- Register hooks in the slash worker process against its session profile
---
 agent/hook_registration.py                    | 104 +++++++++-------
 .../test_hook_registration_parity.py          | 112 +++++++++++++++++-
 tui_gateway/server.py                         |   7 +-
 tui_gateway/slash_worker.py                   |  12 +-
 4 files changed, 183 insertions(+), 52 deletions(-)

diff --git a/agent/hook_registration.py b/agent/hook_registration.py
index 6fdb9964739e..803dcbfdc34b 100644
--- a/agent/hook_registration.py
+++ b/agent/hook_registration.py
@@ -1,72 +1,88 @@
-"""Once-per-process registration of user-configured hooks.
+"""Profile-scoped registration of user-configured hooks.
 
 Every long-lived agent runtime must register the user's shell hooks and
 outbound webhooks at startup, or events configured in config.yaml silently
-never fire for sessions driven through that backend. The call sites:
-
-* ``hermes --cli`` / oneshot — ``hermes_cli.main._prepare_agent_startup``
-  (registers inline, predates this module)
-* messaging gateway — ``gateway/run.py`` (registers inline)
-* TUI stdio backend — ``tui_gateway.entry.main`` → ``tui_gateway.server``
-* TUI WebSocket sidecar (dashboard chat / desktop) —
-  ``tui_gateway.ws.handle_ws`` → ``tui_gateway.server``
-* ``hermes serve`` / dashboard backend — ``hermes_cli.web_server._lifespan``
-
-The two inline call sites predate this module and behave identically; they
-can migrate to :func:`ensure_hooks_registered` later without behavior
-change. Consent semantics are owned by ``agent.shell_hooks`` (flag / env /
-config opt-in, fail-closed on non-TTY stdin) and neither helper ever
-prompts on a backend's piped stdio. Both registrations are idempotent and
-fail-soft: a broken hook config must never take down a backend.
+never fire for sessions driven through that backend. Registrations are keyed
+by ``hermes_home_key()`` at call time because one gateway process can serve
+multiple profiles. A home is marked complete only after registration finishes,
+so concurrent callers for that profile wait rather than observing an early
+no-op. Plugin discovery runs before config hooks because plugin block decisions
+must win ties, matching gateway startup.
+
+Consent semantics are owned by ``agent.shell_hooks`` (flag / env / config
+opt-in, fail-closed on non-TTY stdin) and neither helper ever prompts on a
+backend's piped stdio. Both registrations are idempotent and fail-soft: a
+broken hook config must never take down a backend.
 """
 
 from __future__ import annotations
 
 import logging
 import threading
+from pathlib import Path
 
 logger = logging.getLogger(__name__)
 
 _ensured_lock = threading.Lock()
-_ensured = False
+_completed: set[str] = set()
+_inflight_locks: dict[str, threading.Lock] = {}
 
 
 def reset_for_tests() -> None:
-    """Clear the once-per-process guard (test isolation only)."""
-    global _ensured
+    """Clear profile-scoped registration state (test isolation only)."""
     with _ensured_lock:
-        _ensured = False
+        _completed.clear()
+        _inflight_locks.clear()
 
 
-def ensure_hooks_registered(cfg=None, *, accept_hooks: bool = False) -> None:
-    """Register shell hooks + outbound webhooks exactly once per process.
+def ensure_hooks_registered(
+    cfg=None, *, accept_hooks: bool = False, home: str | Path | None = None
+) -> None:
+    """Register shell hooks + outbound webhooks once for the current profile home.
 
     *cfg* defaults to a fresh ``hermes_cli.config.load_config()`` read.
     *accept_hooks* is passed through to shell-hook registration — callers
     that own a CLI consent flag pass it; backend entry points keep the
     default ``False`` and let the helper resolve opt-in from env/config.
+    *home* is an optional explicit profile home for callers that need to
+    register outside the currently bound profile scope.
 
-    Never raises. Repeat calls are no-ops (the underlying registrations
-    are independently idempotent too, so a caller that must bypass the
-    guard can invoke ``agent.shell_hooks`` / ``agent.outbound_webhooks``
-    directly, as the CLI and gateway already do).
+    Never raises. Repeat calls for a completed home are no-ops; callers for
+    a home whose registration is in progress wait for that work to finish.
     """
-    global _ensured
+    from hermes_constants import hermes_home_key
+
+    home_key = hermes_home_key(home)
     with _ensured_lock:
-        if _ensured:
+        if home_key in _completed:
+            return
+        registration_lock = _inflight_locks.setdefault(home_key, threading.Lock())
+
+    with registration_lock:
+        with _ensured_lock:
+            if home_key in _completed:
+                return
+        try:
+            # Plugin block decisions win ties with declarative hook config.
+            from hermes_cli.plugins import discover_plugins
+
+            discover_plugins()
+        except Exception:
+            logger.debug("plugin discovery failed before hook registration", exc_info=True)
+        try:
+            if cfg is None:
+                from hermes_cli.config import load_config
+
+                cfg = load_config()
+            from agent import outbound_webhooks, shell_hooks
+
+            shell_hooks.register_from_config(cfg, accept_hooks=accept_hooks)
+            outbound_webhooks.register_from_config(cfg)
+        except Exception:
+            logger.debug(
+                "shell-hook / outbound-webhook registration failed at startup",
+                exc_info=True,
+            )
             return
-        _ensured = True
-    try:
-        if cfg is None:
-            from hermes_cli.config import load_config
-
-            cfg = load_config()
-        from agent import outbound_webhooks, shell_hooks
-
-        shell_hooks.register_from_config(cfg, accept_hooks=accept_hooks)
-        outbound_webhooks.register_from_config(cfg)
-    except Exception:
-        logger.debug(
-            "shell-hook / outbound-webhook registration failed at startup",
-            exc_info=True,
-        )
\ No newline at end of file
+        with _ensured_lock:
+            _completed.add(home_key)
diff --git a/tests/tui_gateway/test_hook_registration_parity.py b/tests/tui_gateway/test_hook_registration_parity.py
index 4fad6c84512f..7fb11afaa96c 100644
--- a/tests/tui_gateway/test_hook_registration_parity.py
+++ b/tests/tui_gateway/test_hook_registration_parity.py
@@ -24,6 +24,7 @@
 
 import asyncio
 import io
+import threading
 
 import pytest
 
@@ -31,9 +32,10 @@
 
 
 @pytest.fixture(autouse=True)
-def _fresh_hook_registration_guard():
-    """Start every test with a clean once-per-process guard."""
+def _fresh_hook_registration_guard(monkeypatch):
+    """Start every test with a clean profile-scoped registration guard."""
     hook_registration.reset_for_tests()
+    monkeypatch.setattr("hermes_cli.plugins.discover_plugins", lambda: None)
     yield
     hook_registration.reset_for_tests()
 
@@ -44,15 +46,23 @@ def test_registers_both_shell_hooks_and_outbound_webhooks(self, monkeypatch):
         import agent.shell_hooks as sh
 
         calls: dict[str, object] = {}
+        order: list[str] = []
+
+        def _discover_plugins():
+            order.append("plugins")
 
         def _shell(cfg, *, accept_hooks):
+            order.append("shell")
             calls["shell"] = (cfg, accept_hooks)
             return []
 
         def _outbound(cfg):
+            order.append("outbound")
             calls["outbound"] = cfg
             return []
 
+        monkeypatch.setattr("hermes_cli.plugins.discover_plugins", _discover_plugins)
+
         monkeypatch.setattr(sh, "register_from_config", _shell)
         monkeypatch.setattr(ow, "register_from_config", _outbound)
 
@@ -65,6 +75,7 @@ def _outbound(cfg):
         # never force-enabled from a backend.
         assert calls["shell"] == (cfg, False)
         assert calls["outbound"] is cfg
+        assert order == ["plugins", "shell", "outbound"]
 
     def test_repeat_calls_are_noops(self, monkeypatch):
         import agent.outbound_webhooks as ow
@@ -113,6 +124,103 @@ def test_explicit_cfg_skips_config_read(self, monkeypatch):
         hook_registration.ensure_hooks_registered(cfg)
         assert seen == {"shell": cfg, "outbound": cfg}
 
+    def test_registration_is_scoped_to_each_home(self, monkeypatch, tmp_path):
+        """Each profile home registers once without suppressing another profile."""
+        import agent.outbound_webhooks as ow
+        import agent.shell_hooks as sh
+
+        observed: list[tuple[str, str]] = []
+        monkeypatch.setattr(
+            sh,
+            "register_from_config",
+            lambda cfg, *, accept_hooks: observed.append(("shell", cfg["name"])),
+        )
+        monkeypatch.setattr(
+            ow,
+            "register_from_config",
+            lambda cfg: observed.append(("outbound", cfg["name"])),
+        )
+        home_a = tmp_path / "profile-a"
+        home_b = tmp_path / "profile-b"
+        home_a.mkdir()
+        home_b.mkdir()
+
+        hook_registration.ensure_hooks_registered({"name": "a"}, home=home_a)
+        hook_registration.ensure_hooks_registered({"name": "a-repeat"}, home=home_a)
+        hook_registration.ensure_hooks_registered({"name": "b"}, home=home_b)
+
+        assert observed == [
+            ("shell", "a"), ("outbound", "a"),
+            ("shell", "b"), ("outbound", "b"),
+        ]
+
+    def test_inflight_call_waits_and_failed_registration_retries(self, monkeypatch, tmp_path):
+        """A home is complete only after its registration succeeds."""
+        import agent.outbound_webhooks as ow
+        import agent.shell_hooks as sh
+
+        started = threading.Event()
+        release = threading.Event()
+        second_returned = threading.Event()
+        monkeypatch.setattr(ow, "register_from_config", lambda cfg: None)
+
+        def _blocking_shell(cfg, *, accept_hooks):
+            started.set()
+            assert release.wait(2)
+
+        monkeypatch.setattr(sh, "register_from_config", _blocking_shell)
+        home = tmp_path / "profile"
+        home.mkdir()
+        first = threading.Thread(
+            target=lambda: hook_registration.ensure_hooks_registered({"hooks": {}}, home=home)
+        )
+        second = threading.Thread(
+            target=lambda: (
+                hook_registration.ensure_hooks_registered({"hooks": {}}, home=home),
+                second_returned.set(),
+            )
+        )
+        first.start()
+        assert started.wait(2)
+        second.start()
+        assert not second_returned.wait(2)
+        release.set()
+        first.join(timeout=2)
+        second.join(timeout=2)
+        assert not first.is_alive()
+        assert not second.is_alive()
+
+        retries = {"n": 0}
+
+        def _broken_load_config():
+            retries["n"] += 1
+            raise RuntimeError("bad config")
+
+        monkeypatch.setattr("hermes_cli.config.load_config", _broken_load_config)
+        failed_home = tmp_path / "failed-profile"
+        failed_home.mkdir()
+        hook_registration.ensure_hooks_registered(home=failed_home)
+        hook_registration.ensure_hooks_registered(home=failed_home)
+        assert retries["n"] == 2
+
+    def test_slash_worker_registers_hooks_for_its_process(self, monkeypatch):
+        """The worker's independent runtime performs hook registration."""
+        from tui_gateway import slash_worker
+
+        calls = {"hooks": 0}
+        monkeypatch.setattr(
+            "hermes_cli.mcp_startup.start_background_mcp_discovery", lambda **kwargs: None
+        )
+        monkeypatch.setattr("hermes_cli.mcp_startup.wait_for_mcp_discovery", lambda: None)
+        monkeypatch.setattr(
+            hook_registration,
+            "ensure_hooks_registered",
+            lambda: calls.__setitem__("hooks", calls["hooks"] + 1),
+        )
+
+        slash_worker._prepare_slash_worker_runtime()
+        assert calls == {"hooks": 1}
+
 
 class TestEntryPointWiring:
     """All three backends must call the registration helper at startup —
diff --git a/tui_gateway/server.py b/tui_gateway/server.py
index 5335b4944b2e..e893aef62daf 100644
--- a/tui_gateway/server.py
+++ b/tui_gateway/server.py
@@ -2363,9 +2363,10 @@ def _make_agent(
         with contextlib.suppress(Exception):
             importlib.import_module(_mod).wait_for_mcp_discovery()
     cfg = _load_cfg()
-    # Load hooks alongside the same profile config used to construct this agent.
-    from agent.shell_hooks import register_from_config
-    register_from_config(cfg)
+    # The build scope binds this session's HERMES_HOME; register both hook kinds
+    # from the same profile config without a second config read.
+    from agent.hook_registration import ensure_hooks_registered
+    ensure_hooks_registered(cfg)
     system_prompt = _startup_system_prompt(cfg, session_id or key)
     model, runtime = _resolve_agent_model_runtime(model_override, provider_override)
     _pr = _load_provider_routing()
diff --git a/tui_gateway/slash_worker.py b/tui_gateway/slash_worker.py
index a478adec129f..8017c9188a64 100644
--- a/tui_gateway/slash_worker.py
+++ b/tui_gateway/slash_worker.py
@@ -43,14 +43,20 @@ def _is_orphaned(original_ppid, getppid=os.getppid) -> bool:
 
 
 def _prepare_slash_worker_runtime() -> None:
-    """Start bounded MCP discovery before HermesCLI snapshots tools: each slash_worker child is its
-    own process — the parent ``hermes serve`` discovery thread does not populate this registry.
+    """Start MCP discovery and profile-scoped hook registration before HermesCLI snapshots tools.
 
-    See #61891.
+    Each slash-worker child is its own process, so its session profile needs
+    independent startup work; the parent process cannot populate its state.
     """
     from hermes_cli.mcp_startup import start_background_mcp_discovery, wait_for_mcp_discovery
     start_background_mcp_discovery(logger=logger, thread_name="slash-worker-mcp-discovery")
     wait_for_mcp_discovery()
+    try:
+        from agent.hook_registration import ensure_hooks_registered
+
+        ensure_hooks_registered()
+    except Exception:
+        logger.debug("slash-worker hook registration failed at startup", exc_info=True)
 
 
 def _start_parent_death_watchdog(original_ppid) -> None: