feat(plugins): install from a subdirectory within a repo

Support installing a plugin that lives in a subdirectory of a larger
repo (docs/tests at root, plugin in a subdir) without forcing a
dedicated single-plugin repo.

Identifier syntax:
  owner/repo/path/to/plugin        (shorthand + subpath)
  <url>.git/path/to/plugin         (.git boundary on GitHub-style URLs)
  <url>#path/to/plugin             (explicit fragment, any scheme)

_resolve_git_url now returns (git_url, subdir); _install_plugin_core
reads the manifest from and moves only the subdir, so root-level docs
and tests no longer leak into ~/.hermes/plugins. _resolve_subdir_within
guards against path traversal, missing dirs, and non-directories.

Both the CLI (hermes plugins install) and the dashboard install endpoint
inherit this for free since they share _install_plugin_core. Dashboard
install hint + placeholder updated to advertise the subdir syntax.

.github/workflows/deploy-site.yml

@@ -59,12 +59,22 @@ jobs:
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         run: |
-          # Always rebuild — the file isn't committed (gitignored), so a
-          # fresh checkout starts without it and we want the freshest crawl
-          # in every deploy. Failure is non-fatal: extract-skills.py will
-          # fall back to the legacy snapshot cache and the Skills Hub page
-          # still renders, just without the latest community catalog.
-          python3 scripts/build_skills_index.py || echo "Skills index build failed (non-fatal)"
+          # Rebuild the unified catalog. The file is gitignored, so a fresh
+          # checkout starts without it and we want the freshest crawl in
+          # every deploy.
+          #
+          # This MUST be fatal. build_skills_index.py runs a health check and
+          # exits non-zero WITHOUT writing the output file when a source
+          # collapses (e.g. a GitHub API rate limit zeroes the github /
+          # claude-marketplace / well-known taps all at once). Letting the
+          # deploy continue would either (a) ship a degenerate index missing
+          # whole hubs — the June 2026 regression where OpenAI/Anthropic/
+          # HuggingFace/NVIDIA tabs vanished — or (b) fall through to a
+          # local-only catalog. Failing here keeps the last good deployment
+          # live (GitHub Pages serves the previous build) instead of
+          # publishing a broken catalog. Re-run the workflow once the
+          # transient rate limit clears.
+          python3 scripts/build_skills_index.py
 
       - name: Extract skill metadata for dashboard
         run: python3 website/scripts/extract-skills.py

apps/desktop/electron/main.cjs

@@ -119,6 +119,20 @@ if (REMOTE_DISPLAY_REASON) {
     `[hermes] remote display detected (${REMOTE_DISPLAY_REASON}); disabling GPU hardware acceleration to prevent flicker`
   )
 }
+
+// Keep the renderer running at full speed while the window is in the background
+// or occluded. The chat transcript streams to screen through a
+// requestAnimationFrame-gated flush; Chromium pauses rAF (and clamps timers)
+// for backgrounded/occluded renderers, so without these the live answer stalls
+// whenever the window loses focus (switching to your editor mid-turn, detached
+// devtools, another window covering it) and only paints on refocus or refresh.
+// `backgroundThrottling: false` on the BrowserWindow covers the blurred case;
+// these process-level switches additionally stop Chromium from backgrounding or
+// occlusion-throttling the renderer. Must run before app `ready`.
+app.commandLine.appendSwitch('disable-renderer-backgrounding')
+app.commandLine.appendSwitch('disable-backgrounding-occluded-windows')
+app.commandLine.appendSwitch('disable-background-timer-throttling')
+
 const SOURCE_REPO_ROOT = path.resolve(APP_ROOT, '../..')
 
 // Build-time install stamp -- the git ref this .exe was built against.
@@ -4689,7 +4703,16 @@ function createWindow() {
       webviewTag: true,
       sandbox: true,
       nodeIntegration: false,
-      devTools: true
+      devTools: true,
+      // Keep timers + requestAnimationFrame running at full speed when the
+      // window is blurred/occluded. The chat transcript streams to the screen
+      // through a requestAnimationFrame-gated flush (useSessionStateCache),
+      // so with Chromium's default background throttling the live answer
+      // stalls whenever this window isn't focused (e.g. you switch to your
+      // editor mid-turn, or open detached devtools) and only appears once you
+      // refocus or refresh. A streaming chat app must render in the
+      // background, so opt out — matching the secondary windows above.
+      backgroundThrottling: false
     }
   })
 

cli-config.yaml.example

@@ -528,6 +528,15 @@ session_reset:
   idle_minutes: 1440   # Inactivity timeout in minutes (default: 1440 = 24 hours)
   at_hour: 4           # Daily reset hour, 0-23 local time (default: 4 AM)
 
+# Maximum number of simultaneously active chat sessions across CLI, TUI,
+# dashboard chat, and messaging gateway. Set to null, 0, or omit to allow
+# unlimited concurrent sessions. When the limit is reached, new sessions get a
+# clean error while existing active sessions keep their normal behavior. This
+# top-level key takes precedence over gateway.max_concurrent_sessions. The cap
+# is a best-effort single-host/profile runtime guard; Hermes fails open if the
+# local runtime lease registry cannot be read or locked.
+max_concurrent_sessions: null
+
 # When true, group/channel chats use one session per participant when the platform
 # provides a user ID. This is the secure default and prevents users in the same
 # room from sharing context, interrupts, and token costs. Set false only if you

cli.py

@@ -3462,6 +3462,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         self._image_counter = 0
         self.preloaded_skills: list[str] = []
         self._startup_skills_line_shown = False
+        self._active_session_lease = None
 
         # Voice mode state (also reinitialized inside run() for interactive TUI).
         self._voice_lock = threading.Lock()
@@ -3490,6 +3491,45 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         self._background_tasks: Dict[str, threading.Thread] = {}
         self._background_task_counter = 0
 
+    def _claim_active_session(self, surface: str = "cli", *, stderr: bool = False) -> bool:
+        """Claim a global active-session slot for this CLI process."""
+        if self._active_session_lease is not None:
+            return True
+        try:
+            from hermes_cli.active_sessions import try_acquire_active_session
+
+            lease, message = try_acquire_active_session(
+                session_id=self.session_id,
+                surface=surface,
+                config=self.config,
+            )
+        except Exception as exc:
+            logger.warning("Failed to claim active session slot: %s", exc)
+            return True
+        if message:
+            if stderr:
+                print(message, file=sys.stderr)
+            else:
+                self._console_print(f"[bold red]{message}[/]")
+            return False
+        self._active_session_lease = lease
+        try:
+            atexit.register(self._release_active_session)
+        except Exception:
+            pass
+        return True
+
+    def _release_active_session(self) -> None:
+        lease = getattr(self, "_active_session_lease", None)
+        if lease is None:
+            return
+        try:
+            lease.release()
+        except Exception:
+            logger.debug("Failed to release active session slot", exc_info=True)
+        finally:
+            self._active_session_lease = None
+
     def _invalidate(self, min_interval: float = 0.25) -> None:
         """Throttled UI repaint for high-frequency background updates.
 
@@ -10497,6 +10537,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
 
     def run(self):
         """Run the interactive CLI loop with persistent input at bottom."""
+        if not self._claim_active_session("cli"):
+            return
+
         # Detect light/dark terminal mode now (before pt grabs the tty).
         # Caches the result so subsequent _hex_to_ansi / style calls
         # don't risk re-querying mid-render.
@@ -12918,6 +12961,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
                     pass
             _run_cleanup()
             self._print_exit_summary()
+            self._release_active_session()
 
         # Deferred relaunch: /update sets _pending_relaunch so the exec
         # happens here — after prompt_toolkit has exited and fully restored
@@ -13281,219 +13325,224 @@ def main(
     
     # Handle single query mode
     if query or image:
-        query, single_query_images = _collect_query_images(query, image)
-        # Kanban workers spawn with ``hermes chat -q "work kanban task <id>"``;
-        # the actual task description lives in the task body. Mirror the
-        # gateway/CLI behaviour for inbound images by scanning the body for
-        # local image paths and http(s) image URLs and attaching them to the
-        # worker's first turn. Without this, users who paste a screenshot
-        # path or URL into a kanban task body never get it routed to the
-        # model's vision input.
-        single_query_image_urls: list[str] = []
-        _kanban_task_id = os.environ.get("HERMES_KANBAN_TASK", "").strip()
-        if _kanban_task_id:
-            try:
-                from hermes_cli import kanban_db as _kb
-                from agent.image_routing import extract_image_refs as _extract_refs
-
-                _conn = _kb.connect()
+        if not cli._claim_active_session("cli", stderr=bool(quiet)):
+            sys.exit(1)
+        try:
+            query, single_query_images = _collect_query_images(query, image)
+            # Kanban workers spawn with ``hermes chat -q "work kanban task <id>"``;
+            # the actual task description lives in the task body. Mirror the
+            # gateway/CLI behaviour for inbound images by scanning the body for
+            # local image paths and http(s) image URLs and attaching them to the
+            # worker's first turn. Without this, users who paste a screenshot
+            # path or URL into a kanban task body never get it routed to the
+            # model's vision input.
+            single_query_image_urls: list[str] = []
+            _kanban_task_id = os.environ.get("HERMES_KANBAN_TASK", "").strip()
+            if _kanban_task_id:
                 try:
-                    _task = _kb.get_task(_conn, _kanban_task_id)
-                finally:
-                    try:
-                        _conn.close()
-                    except Exception:
-                        pass
-                _body = getattr(_task, "body", "") if _task is not None else ""
-                if _body:
-                    _kb_paths, _kb_urls = _extract_refs(_body)
-                    if _kb_paths:
-                        # Dedupe against any --image the user already passed.
-                        _seen = {str(p) for p in single_query_images}
-                        for _p in _kb_paths:
-                            if _p not in _seen:
-                                _seen.add(_p)
-                                single_query_images.append(Path(_p))
-                    if _kb_urls:
-                        single_query_image_urls.extend(_kb_urls)
-            except Exception as _exc:
-                # Best-effort enrichment; never block worker startup on it.
-                logger.debug("kanban image-ref extraction failed: %s", _exc)
-        if quiet:
-            # Quiet mode: suppress banner, spinner, tool previews.
-            # Only print the final response and parseable session info.
-            cli.tool_progress_mode = "off"
-            if cli._ensure_runtime_credentials():
-                effective_query: Any = query
-                if single_query_images or single_query_image_urls:
-                    # Honour the same image-routing decision used by the
-                    # interactive path. With a vision-capable model (incl.
-                    # custom-provider models declared via
-                    # `model.supports_vision: true`), attach images natively
-                    # as image_url content parts. Otherwise fall back to the
-                    # text-pipeline (vision_analyze pre-description).
-                    _img_mode = "text"
-                    _build_parts = None
-                    try:
-                        from agent.image_routing import (
-                            build_native_content_parts as _build_parts,  # noqa: F811
-                        )
-                        from agent.image_routing import decide_image_input_mode
-                        from hermes_cli.config import load_config
+                    from hermes_cli import kanban_db as _kb
+                    from agent.image_routing import extract_image_refs as _extract_refs
 
-                        _img_mode = decide_image_input_mode(
-                            (cli.provider or "").strip(),
-                            (cli.model or "").strip(),
-                            load_config(),
-                        )
-                    except Exception:
-                        _img_mode = "text"
-
-                    if _img_mode == "native" and _build_parts is not None:
+                    _conn = _kb.connect()
+                    try:
+                        _task = _kb.get_task(_conn, _kanban_task_id)
+                    finally:
                         try:
-                            _parts, _skipped = _build_parts(
-                                query if isinstance(query, str) else "",
-                                [str(p) for p in single_query_images],
-                                image_urls=list(single_query_image_urls) or None,
+                            _conn.close()
+                        except Exception:
+                            pass
+                    _body = getattr(_task, "body", "") if _task is not None else ""
+                    if _body:
+                        _kb_paths, _kb_urls = _extract_refs(_body)
+                        if _kb_paths:
+                            # Dedupe against any --image the user already passed.
+                            _seen = {str(p) for p in single_query_images}
+                            for _p in _kb_paths:
+                                if _p not in _seen:
+                                    _seen.add(_p)
+                                    single_query_images.append(Path(_p))
+                        if _kb_urls:
+                            single_query_image_urls.extend(_kb_urls)
+                except Exception as _exc:
+                    # Best-effort enrichment; never block worker startup on it.
+                    logger.debug("kanban image-ref extraction failed: %s", _exc)
+            if quiet:
+                # Quiet mode: suppress banner, spinner, tool previews.
+                # Only print the final response and parseable session info.
+                cli.tool_progress_mode = "off"
+                if cli._ensure_runtime_credentials():
+                    effective_query: Any = query
+                    if single_query_images or single_query_image_urls:
+                        # Honour the same image-routing decision used by the
+                        # interactive path. With a vision-capable model (incl.
+                        # custom-provider models declared via
+                        # `model.supports_vision: true`), attach images natively
+                        # as image_url content parts. Otherwise fall back to the
+                        # text-pipeline (vision_analyze pre-description).
+                        _img_mode = "text"
+                        _build_parts = None
+                        try:
+                            from agent.image_routing import (
+                                build_native_content_parts as _build_parts,  # noqa: F811
                             )
-                            if any(p.get("type") == "image_url" for p in _parts):
-                                effective_query = _parts
-                            else:
-                                # All images unreadable — text fallback.
-                                # ``_preprocess_images_with_vision`` only knows
-                                # about local files; URLs would be lost there,
-                                # so keep the original query text intact when
-                                # only URLs were supplied.
+                            from agent.image_routing import decide_image_input_mode
+                            from hermes_cli.config import load_config
+
+                            _img_mode = decide_image_input_mode(
+                                (cli.provider or "").strip(),
+                                (cli.model or "").strip(),
+                                load_config(),
+                            )
+                        except Exception:
+                            _img_mode = "text"
+
+                        if _img_mode == "native" and _build_parts is not None:
+                            try:
+                                _parts, _skipped = _build_parts(
+                                    query if isinstance(query, str) else "",
+                                    [str(p) for p in single_query_images],
+                                    image_urls=list(single_query_image_urls) or None,
+                                )
+                                if any(p.get("type") == "image_url" for p in _parts):
+                                    effective_query = _parts
+                                else:
+                                    # All images unreadable — text fallback.
+                                    # ``_preprocess_images_with_vision`` only knows
+                                    # about local files; URLs would be lost there,
+                                    # so keep the original query text intact when
+                                    # only URLs were supplied.
+                                    if single_query_images:
+                                        effective_query = cli._preprocess_images_with_vision(
+                                            query, single_query_images, announce=False,
+                                        )
+                            except Exception:
                                 if single_query_images:
                                     effective_query = cli._preprocess_images_with_vision(
                                         query, single_query_images, announce=False,
                                     )
-                        except Exception:
-                            if single_query_images:
-                                effective_query = cli._preprocess_images_with_vision(
-                                    query, single_query_images, announce=False,
-                                )
-                    elif single_query_images:
-                        effective_query = cli._preprocess_images_with_vision(
-                            query,
-                            single_query_images,
-                            announce=False,
-                        )
-                turn_route = cli._resolve_turn_agent_config(effective_query)
-                if turn_route["signature"] != cli._active_agent_route_signature:
-                    cli.agent = None
-                if cli._init_agent(
-                    model_override=turn_route["model"],
-                    runtime_override=turn_route["runtime"],
-                    request_overrides=turn_route.get("request_overrides"),
-                ):
-                    cli.agent.quiet_mode = True
-                    cli.agent.suppress_status_output = True
-                    # Suppress streaming display callbacks so stdout stays
-                    # machine-readable (no styled "Hermes" box, no tool-gen
-                    # status lines).  The response is printed once below.
-                    cli.agent.stream_delta_callback = None
-                    cli.agent.tool_gen_callback = None
-                    try:
-                        result = cli.agent.run_conversation(
-                            user_message=effective_query,
-                            conversation_history=cli.conversation_history,
-                        )
-                    except KeyboardInterrupt:
-                        _emit_interrupted_session_end(cli, reason="keyboard_interrupt")
-                        print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
-                        sys.exit(130)
-                    # Sync session_id if mid-run compression created a
-                    # continuation session. The exit line below reports
-                    # session_id to stderr for automation wrappers; without
-                    # this sync it would point at the ended parent.
-                    if (
-                        getattr(cli.agent, "session_id", None)
-                        and cli.agent.session_id != cli.session_id
+                        elif single_query_images:
+                            effective_query = cli._preprocess_images_with_vision(
+                                query,
+                                single_query_images,
+                                announce=False,
+                            )
+                    turn_route = cli._resolve_turn_agent_config(effective_query)
+                    if turn_route["signature"] != cli._active_agent_route_signature:
+                        cli.agent = None
+                    if cli._init_agent(
+                        model_override=turn_route["model"],
+                        runtime_override=turn_route["runtime"],
+                        request_overrides=turn_route.get("request_overrides"),
                     ):
-                        cli.session_id = cli.agent.session_id
-                    response = result.get("final_response", "") if isinstance(result, dict) else str(result)
-                    # Surface backend errors that produced no visible output
-                    # (e.g. invalid model slug → provider 4xx). Mirrors the
-                    # interactive CLI path. Write to stderr so piped stdout
-                    # stays clean for automation wrappers.
-                    if (
-                        not response
-                        and isinstance(result, dict)
-                        and result.get("error")
-                        and (result.get("failed") or result.get("partial"))
-                    ):
-                        print(f"Error: {result['error']}", file=sys.stderr)
-                    elif response:
-                        print(response)
-
-                    # Kanban goal-loop mode: a worker spawned for a
-                    # goal_mode card keeps working in THIS session until an
-                    # auxiliary judge agrees the card is done, the worker
-                    # terminates the task itself, or the turn budget runs
-                    # out (→ sticky block). Gated on the env vars the
-                    # dispatcher sets in `_default_spawn`; a no-op for every
-                    # normal worker and every non-kanban `-q` run.
-                    if os.environ.get("HERMES_KANBAN_GOAL_MODE") == "1":
+                        cli.agent.quiet_mode = True
+                        cli.agent.suppress_status_output = True
+                        # Suppress streaming display callbacks so stdout stays
+                        # machine-readable (no styled "Hermes" box, no tool-gen
+                        # status lines).  The response is printed once below.
+                        cli.agent.stream_delta_callback = None
+                        cli.agent.tool_gen_callback = None
                         try:
-                            _run_kanban_goal_loop_q(cli, response)
-                        except Exception as _goal_exc:
-                            logger.debug("kanban goal loop failed: %s", _goal_exc)
+                            result = cli.agent.run_conversation(
+                                user_message=effective_query,
+                                conversation_history=cli.conversation_history,
+                            )
+                        except KeyboardInterrupt:
+                            _emit_interrupted_session_end(cli, reason="keyboard_interrupt")
+                            print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
+                            sys.exit(130)
+                        # Sync session_id if mid-run compression created a
+                        # continuation session. The exit line below reports
+                        # session_id to stderr for automation wrappers; without
+                        # this sync it would point at the ended parent.
+                        if (
+                            getattr(cli.agent, "session_id", None)
+                            and cli.agent.session_id != cli.session_id
+                        ):
+                            cli.session_id = cli.agent.session_id
+                        response = result.get("final_response", "") if isinstance(result, dict) else str(result)
+                        # Surface backend errors that produced no visible output
+                        # (e.g. invalid model slug → provider 4xx). Mirrors the
+                        # interactive CLI path. Write to stderr so piped stdout
+                        # stays clean for automation wrappers.
+                        if (
+                            not response
+                            and isinstance(result, dict)
+                            and result.get("error")
+                            and (result.get("failed") or result.get("partial"))
+                        ):
+                            print(f"Error: {result['error']}", file=sys.stderr)
+                        elif response:
+                            print(response)
 
-                    # Session ID goes to stderr so piped stdout is clean.
-                    print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
-
-                    # Ensure proper exit code for automation wrappers.
-                    #
-                    # Kanban workers get a special case: when the run failed
-                    # purely because the provider rate-limited / exhausted
-                    # quota (not because the task itself is broken), exit with
-                    # the EX_TEMPFAIL sentinel instead of the generic 1. The
-                    # dispatcher's reap classifier maps that code to a
-                    # ``rate_limited`` exit and releases the task back to
-                    # ``ready`` WITHOUT incrementing the failure counter, so a
-                    # 5-hour quota window can't trip the circuit breaker and
-                    # permanently block the card. Non-kanban runs keep the
-                    # plain 0/1 contract automation wrappers expect.
-                    _exit_code = 0
-                    if isinstance(result, dict) and result.get("failed"):
-                        _exit_code = 1
-                        if os.environ.get("HERMES_KANBAN_TASK") and result.get(
-                            "failure_reason"
-                        ) in ("rate_limit", "billing"):
+                        # Kanban goal-loop mode: a worker spawned for a
+                        # goal_mode card keeps working in THIS session until an
+                        # auxiliary judge agrees the card is done, the worker
+                        # terminates the task itself, or the turn budget runs
+                        # out (→ sticky block). Gated on the env vars the
+                        # dispatcher sets in `_default_spawn`; a no-op for every
+                        # normal worker and every non-kanban `-q` run.
+                        if os.environ.get("HERMES_KANBAN_GOAL_MODE") == "1":
                             try:
-                                from hermes_cli.kanban_db import (
-                                    KANBAN_RATE_LIMIT_EXIT_CODE as _RL_CODE,
-                                )
-                                _exit_code = _RL_CODE
-                            except Exception:
-                                _exit_code = 1
-                    sys.exit(_exit_code)
-            
-            # Exit with error code if credentials or agent init fails
-            sys.exit(1)
-        else:
-            # Single-query mode (`hermes chat -q "…"`): skip the welcome
-            # banner. Building the banner takes ~420 ms on cold start —
-            # ~200 ms of that is the version-update check, the rest is
-            # toolset / skill enumeration and Rich panel rendering. None
-            # of that is useful for a one-shot query: the user already
-            # picked the prompt, doesn't need a toolset reference, and
-            # gets the session ID + resume hint from
-            # ``_print_exit_summary()`` after the response prints.
-            #
-            # The fully-quiet ``-Q`` / ``--quiet`` machine-readable path
-            # above was already banner-free; this brings the human-
-            # facing single-query path in line so all non-interactive
-            # invocations are fast.
-            _query_label = query or ("[image attached]" if single_query_images else "")
-            if _query_label:
-                cli.console.print(f"[bold blue]Query:[/] {_query_label}")
-            # Surface security advisories before the agent runs — short
-            # banner, doesn't depend on the welcome banner being shown.
-            cli._show_security_advisories()
-            cli.chat(query, images=single_query_images or None)
-            cli._print_exit_summary()
+                                _run_kanban_goal_loop_q(cli, response)
+                            except Exception as _goal_exc:
+                                logger.debug("kanban goal loop failed: %s", _goal_exc)
+
+                        # Session ID goes to stderr so piped stdout is clean.
+                        print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
+
+                        # Ensure proper exit code for automation wrappers.
+                        #
+                        # Kanban workers get a special case: when the run failed
+                        # purely because the provider rate-limited / exhausted
+                        # quota (not because the task itself is broken), exit with
+                        # the EX_TEMPFAIL sentinel instead of the generic 1. The
+                        # dispatcher's reap classifier maps that code to a
+                        # ``rate_limited`` exit and releases the task back to
+                        # ``ready`` WITHOUT incrementing the failure counter, so a
+                        # 5-hour quota window can't trip the circuit breaker and
+                        # permanently block the card. Non-kanban runs keep the
+                        # plain 0/1 contract automation wrappers expect.
+                        _exit_code = 0
+                        if isinstance(result, dict) and result.get("failed"):
+                            _exit_code = 1
+                            if os.environ.get("HERMES_KANBAN_TASK") and result.get(
+                                "failure_reason"
+                            ) in ("rate_limit", "billing"):
+                                try:
+                                    from hermes_cli.kanban_db import (
+                                        KANBAN_RATE_LIMIT_EXIT_CODE as _RL_CODE,
+                                    )
+                                    _exit_code = _RL_CODE
+                                except Exception:
+                                    _exit_code = 1
+                        sys.exit(_exit_code)
+
+                # Exit with error code if credentials or agent init fails
+                sys.exit(1)
+            else:
+                # Single-query mode (`hermes chat -q "…"`): skip the welcome
+                # banner. Building the banner takes ~420 ms on cold start —
+                # ~200 ms of that is the version-update check, the rest is
+                # toolset / skill enumeration and Rich panel rendering. None
+                # of that is useful for a one-shot query: the user already
+                # picked the prompt, doesn't need a toolset reference, and
+                # gets the session ID + resume hint from
+                # ``_print_exit_summary()`` after the response prints.
+                #
+                # The fully-quiet ``-Q`` / ``--quiet`` machine-readable path
+                # above was already banner-free; this brings the human-
+                # facing single-query path in line so all non-interactive
+                # invocations are fast.
+                _query_label = query or ("[image attached]" if single_query_images else "")
+                if _query_label:
+                    cli.console.print(f"[bold blue]Query:[/] {_query_label}")
+                # Surface security advisories before the agent runs — short
+                # banner, doesn't depend on the welcome banner being shown.
+                cli._show_security_advisories()
+                cli.chat(query, images=single_query_images or None)
+                cli._print_exit_summary()
+        finally:
+            cli._release_active_session()
         return
     
     # Run interactive mode

gateway/config.py

@@ -56,6 +56,42 @@ def _coerce_int(value: Any, default: int) -> int:
         return default
 
 
+def _coerce_optional_positive_int(value: Any, key: str) -> Optional[int]:
+    """Coerce an optional positive integer config value.
+
+    ``None``/0/negative disable the setting. Malformed values are ignored with
+    a warning so a typo never prevents the gateway from starting.
+    """
+    if value is None:
+        return None
+    if isinstance(value, bool):
+        logger.warning(
+            "Ignoring invalid %s=%r (expected a positive integer; 0/null disables)",
+            key,
+            value,
+        )
+        return None
+    try:
+        if isinstance(value, float):
+            if not value.is_integer():
+                raise ValueError(value)
+            parsed = int(value)
+        elif isinstance(value, str):
+            parsed = int(value.strip(), 10)
+        else:
+            parsed = int(value)
+    except (TypeError, ValueError):
+        logger.warning(
+            "Ignoring invalid %s=%r (expected a positive integer; 0/null disables)",
+            key,
+            value,
+        )
+        return None
+    if parsed <= 0:
+        return None
+    return parsed
+
+
 def _normalize_unauthorized_dm_behavior(value: Any, default: str = "pair") -> str:
     """Normalize unauthorized DM behavior to a supported value."""
     if isinstance(value, str):
@@ -495,6 +531,7 @@ class GatewayConfig:
     # Session isolation in shared chats
     group_sessions_per_user: bool = True  # Isolate group/channel sessions per participant when user IDs are available
     thread_sessions_per_user: bool = False  # When False (default), threads are shared across all participants
+    max_concurrent_sessions: Optional[int] = None  # Positive int caps simultaneous active chat sessions
 
     # Unauthorized DM policy
     unauthorized_dm_behavior: str = "pair"  # "pair" or "ignore"
@@ -600,6 +637,7 @@ class GatewayConfig:
             "stt_enabled": self.stt_enabled,
             "group_sessions_per_user": self.group_sessions_per_user,
             "thread_sessions_per_user": self.thread_sessions_per_user,
+            "max_concurrent_sessions": self.max_concurrent_sessions,
             "unauthorized_dm_behavior": self.unauthorized_dm_behavior,
             "streaming": self.streaming.to_dict(),
             "session_store_max_age_days": self.session_store_max_age_days,
@@ -645,6 +683,17 @@ class GatewayConfig:
 
         group_sessions_per_user = data.get("group_sessions_per_user")
         thread_sessions_per_user = data.get("thread_sessions_per_user")
+        nested_gateway = data.get("gateway") if isinstance(data.get("gateway"), dict) else {}
+        if "max_concurrent_sessions" in data:
+            max_concurrent_raw = data.get("max_concurrent_sessions")
+            max_concurrent_key = "max_concurrent_sessions"
+        else:
+            max_concurrent_raw = nested_gateway.get("max_concurrent_sessions")
+            max_concurrent_key = "gateway.max_concurrent_sessions"
+        max_concurrent_sessions = _coerce_optional_positive_int(
+            max_concurrent_raw,
+            max_concurrent_key,
+        )
         unauthorized_dm_behavior = _normalize_unauthorized_dm_behavior(
             data.get("unauthorized_dm_behavior"),
             "pair",
@@ -671,6 +720,7 @@ class GatewayConfig:
             stt_enabled=_coerce_bool(stt_enabled, True),
             group_sessions_per_user=_coerce_bool(group_sessions_per_user, True),
             thread_sessions_per_user=_coerce_bool(thread_sessions_per_user, False),
+            max_concurrent_sessions=max_concurrent_sessions,
             unauthorized_dm_behavior=unauthorized_dm_behavior,
             streaming=StreamingConfig.from_dict(data.get("streaming", {})),
             session_store_max_age_days=session_store_max_age_days,
@@ -761,6 +811,13 @@ def load_gateway_config() -> GatewayConfig:
             if "thread_sessions_per_user" in yaml_cfg:
                 gw_data["thread_sessions_per_user"] = yaml_cfg["thread_sessions_per_user"]
 
+            gateway_section = yaml_cfg.get("gateway")
+            if isinstance(gateway_section, dict) and "max_concurrent_sessions" in gateway_section:
+                gw_data["max_concurrent_sessions"] = gateway_section["max_concurrent_sessions"]
+
+            if "max_concurrent_sessions" in yaml_cfg:
+                gw_data["max_concurrent_sessions"] = yaml_cfg["max_concurrent_sessions"]
+
             streaming_cfg = yaml_cfg.get("streaming")
             if not isinstance(streaming_cfg, dict):
                 # Fall back to nested gateway.streaming written by

gateway/run.py

@@ -1934,6 +1934,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # Key: session_key, Value: AIAgent instance
         self._running_agents: Dict[str, Any] = {}
         self._running_agents_ts: Dict[str, float] = {}  # start timestamp per session
+        self._active_session_leases: Dict[str, Any] = {}
         self._pending_messages: Dict[str, str] = {}  # Queued messages during interrupt
         # Last successfully-resolved (non-empty) model, keyed by session. Used
         # as a fallback when a fresh config read transiently returns an empty
@@ -3390,6 +3391,59 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             if agent is not _AGENT_PENDING_SENTINEL
         }
 
+    def _get_max_concurrent_sessions(self) -> Optional[int]:
+        """Return the configured active chat session cap, if enabled."""
+        try:
+            from hermes_cli.active_sessions import resolve_max_concurrent_sessions
+
+            return resolve_max_concurrent_sessions(getattr(self, "config", None))
+        except Exception:
+            return None
+
+    def _active_session_limit_message(self, session_key: str) -> Optional[str]:
+        """Return a user-facing rejection when starting a new session exceeds the cap."""
+        max_sessions = self._get_max_concurrent_sessions()
+        if max_sessions is None:
+            return None
+        if session_key in getattr(self, "_running_agents", {}):
+            return None
+        active_count = len(getattr(self, "_running_agents", {}))
+        if active_count < max_sessions:
+            return None
+        return (
+            f"Hermes is at the active session limit ({active_count}/{max_sessions}). "
+            "Try again when another session finishes."
+        )
+
+    def _claim_active_session_slot(
+        self,
+        session_key: str,
+        source: SessionSource,
+    ) -> tuple[Any, Optional[str]]:
+        """Claim a cross-process active-session slot for a new gateway turn."""
+        if session_key in getattr(self, "_running_agents", {}):
+            return None, None
+        local_limit_message = self._active_session_limit_message(session_key)
+        if local_limit_message is not None:
+            return None, local_limit_message
+        try:
+            from hermes_cli.active_sessions import try_acquire_active_session
+
+            platform = source.platform.value if source and source.platform else "gateway"
+            return try_acquire_active_session(
+                session_id=session_key,
+                surface=f"gateway:{platform}",
+                config=getattr(self, "config", None),
+                metadata={
+                    "platform": platform,
+                    "chat_id": getattr(source, "chat_id", "") or "",
+                    "user_id": getattr(source, "user_id", "") or "",
+                },
+            )
+        except Exception as exc:
+            logger.warning("Failed to claim active session slot: %s", exc)
+            return None, None
+
     @staticmethod
     def _agent_has_active_subagents(running_agent: Any) -> bool:
         """Return True when *running_agent* is currently driving subagents
@@ -5751,8 +5805,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             self._background_tasks.clear()
 
             self.adapters.clear()
+            for _session_key in list(self._running_agents):
+                self._release_running_agent_state(_session_key)
             self._running_agents.clear()
             self._running_agents_ts.clear()
+            if hasattr(self, "_active_session_leases"):
+                self._active_session_leases.clear()
             self._pending_messages.clear()
             self._pending_approvals.clear()
             if hasattr(self, '_busy_ack_ts'):
@@ -7237,6 +7295,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # message arriving during any of those yields would pass the
         # "already running" guard and spin up a duplicate agent for the
         # same session — corrupting the transcript.
+        _active_session_lease, _limit_message = self._claim_active_session_slot(
+            _quick_key,
+            source,
+        )
+        if _limit_message is not None:
+            logger.info(
+                "Rejecting new active session %s: max_concurrent_sessions reached",
+                _quick_key,
+            )
+            return _limit_message
+        if _active_session_lease is not None:
+            if not hasattr(self, "_active_session_leases"):
+                self._active_session_leases = {}
+            self._active_session_leases[_quick_key] = _active_session_lease
         self._running_agents[_quick_key] = _AGENT_PENDING_SENTINEL
         self._running_agents_ts[_quick_key] = time.time()
         _run_generation = self._begin_session_run_generation(_quick_key)
@@ -11976,6 +12048,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             session_key, run_generation
         ):
             return False
+        lease = getattr(self, "_active_session_leases", {}).pop(session_key, None)
+        if lease is not None:
+            try:
+                lease.release()
+            except Exception:
+                logger.debug("Failed to release active session slot", exc_info=True)
         self._running_agents.pop(session_key, None)
         self._running_agents_ts.pop(session_key, None)
         if hasattr(self, "_busy_ack_ts"):

hermes_cli/active_sessions.py

@@ -0,0 +1,320 @@
+"""Cross-process active chat session leases.
+
+The session database records persisted conversations.  This module records
+currently open chat surfaces, including idle CLI/TUI sessions that have not
+written a transcript row yet.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Optional
+
+from hermes_constants import get_hermes_home
+
+logger = logging.getLogger(__name__)
+
+
+def coerce_max_concurrent_sessions(value: Any, key: str = "max_concurrent_sessions") -> Optional[int]:
+    """Return a positive integer cap, or None when disabled/invalid."""
+    if value is None:
+        return None
+    if isinstance(value, bool):
+        logger.warning(
+            "Ignoring invalid %s=%r (expected a positive integer; 0/null disables)",
+            key,
+            value,
+        )
+        return None
+    try:
+        if isinstance(value, float):
+            if not value.is_integer():
+                raise ValueError(value)
+            parsed = int(value)
+        elif isinstance(value, str):
+            parsed = int(value.strip(), 10)
+        else:
+            parsed = int(value)
+    except (TypeError, ValueError):
+        logger.warning(
+            "Ignoring invalid %s=%r (expected a positive integer; 0/null disables)",
+            key,
+            value,
+        )
+        return None
+    if parsed <= 0:
+        return None
+    return parsed
+
+
+def resolve_max_concurrent_sessions(config: Any) -> Optional[int]:
+    """Resolve top-level max_concurrent_sessions with gateway.* fallback."""
+    raw: Any = None
+    key = "max_concurrent_sessions"
+    if isinstance(config, dict):
+        if "max_concurrent_sessions" in config:
+            raw = config.get("max_concurrent_sessions")
+        else:
+            gateway_cfg = config.get("gateway")
+            if isinstance(gateway_cfg, dict):
+                raw = gateway_cfg.get("max_concurrent_sessions")
+                key = "gateway.max_concurrent_sessions"
+    else:
+        raw = getattr(config, "max_concurrent_sessions", None)
+    return coerce_max_concurrent_sessions(raw, key=key)
+
+
+def active_session_limit_message(active_count: int, max_sessions: int) -> str:
+    return (
+        f"Hermes is at the active session limit ({active_count}/{max_sessions}). "
+        "Try again when another session finishes."
+    )
+
+
+def _state_dir() -> Path:
+    return get_hermes_home() / "runtime"
+
+
+def _state_path() -> Path:
+    return _state_dir() / "active_sessions.json"
+
+
+def _lock_path() -> Path:
+    return _state_dir() / "active_sessions.lock"
+
+
+class _FileLock:
+    def __init__(self, path: Path):
+        self.path = path
+        self._fh = None
+
+    def __enter__(self):
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._fh = open(self.path, "a+b")
+        if os.name == "nt":
+            try:
+                import msvcrt
+
+                self._fh.seek(0)
+                msvcrt.locking(self._fh.fileno(), msvcrt.LK_LOCK, 1)
+            except Exception as exc:
+                self._fh.close()
+                self._fh = None
+                raise RuntimeError("active session file lock unavailable") from exc
+        else:
+            try:
+                import fcntl
+
+                fcntl.flock(self._fh.fileno(), fcntl.LOCK_EX)
+            except Exception as exc:
+                self._fh.close()
+                self._fh = None
+                raise RuntimeError("active session file lock unavailable") from exc
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        if self._fh is None:
+            return
+        if os.name == "nt":
+            try:
+                import msvcrt
+
+                self._fh.seek(0)
+                msvcrt.locking(self._fh.fileno(), msvcrt.LK_UNLCK, 1)
+            except Exception:
+                pass
+        else:
+            try:
+                import fcntl
+
+                fcntl.flock(self._fh.fileno(), fcntl.LOCK_UN)
+            except Exception:
+                pass
+        try:
+            self._fh.close()
+        finally:
+            self._fh = None
+
+
+def _read_entries(path: Path) -> list[dict[str, Any]]:
+    try:
+        with open(path, "r", encoding="utf-8") as fh:
+            data = json.load(fh)
+    except FileNotFoundError:
+        return []
+    except Exception:
+        logger.warning("Ignoring corrupt active session registry at %s", path)
+        return []
+    entries = data.get("entries") if isinstance(data, dict) else data
+    if not isinstance(entries, list):
+        return []
+    return [entry for entry in entries if isinstance(entry, dict)]
+
+
+def _write_entries(path: Path, entries: list[dict[str, Any]]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_name(f"{path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp")
+    with open(tmp, "w", encoding="utf-8") as fh:
+        json.dump({"entries": entries}, fh, sort_keys=True)
+    os.replace(tmp, path)
+
+
+def _process_start_time(pid: int) -> Optional[float]:
+    # Pair pid with process create_time when psutil can read it, so a recycled
+    # pid does not keep a stale lease alive indefinitely.
+    try:
+        import psutil  # type: ignore
+
+        return float(psutil.Process(pid).create_time())
+    except Exception:
+        return None
+
+
+def _optional_float(value: Any) -> Optional[float]:
+    if value is None or value == "":
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _pid_alive(pid: Any, process_start_time: Any = None) -> bool:
+    try:
+        pid_int = int(pid)
+    except (TypeError, ValueError):
+        return False
+    if pid_int <= 0:
+        return False
+    try:
+        from gateway.status import _pid_exists
+
+        exists = bool(_pid_exists(pid_int))
+    except Exception:
+        return False
+    if not exists:
+        return False
+    expected_start = _optional_float(process_start_time)
+    if expected_start is None:
+        return True
+    current_start = _process_start_time(pid_int)
+    if current_start is None:
+        return True
+    return abs(current_start - expected_start) < 0.001
+
+
+def _prune_dead(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [
+        entry
+        for entry in entries
+        if _pid_alive(entry.get("pid"), entry.get("process_start_time"))
+    ]
+
+
+@dataclass
+class ActiveSessionLease:
+    lease_id: str
+    session_id: str
+    surface: str
+    enabled: bool = True
+    released: bool = False
+
+    def release(self) -> None:
+        if self.released or not self.enabled:
+            return
+        release_active_session(self)
+
+
+def try_acquire_active_session(
+    *,
+    session_id: str,
+    surface: str,
+    config: Any,
+    metadata: Optional[dict[str, Any]] = None,
+) -> tuple[Optional[ActiveSessionLease], Optional[str]]:
+    """Acquire an active-session slot.
+
+    Returns ``(lease, None)`` on success.  When the cap is disabled, the lease is
+    a no-op object so callers can unconditionally call ``release()``.
+    """
+    max_sessions = resolve_max_concurrent_sessions(config)
+    lease_id = uuid.uuid4().hex
+    if max_sessions is None:
+        return ActiveSessionLease(
+            lease_id=lease_id,
+            session_id=session_id,
+            surface=surface,
+            enabled=False,
+        ), None
+
+    now = time.time()
+    entry = {
+        "lease_id": lease_id,
+        "session_id": str(session_id),
+        "surface": str(surface),
+        "pid": os.getpid(),
+        "process_start_time": _process_start_time(os.getpid()),
+        "started_at": now,
+        "updated_at": now,
+    }
+    if metadata:
+        entry["metadata"] = {
+            str(k): v for k, v in metadata.items() if isinstance(k, str)
+        }
+
+    state_path = _state_path()
+    with _FileLock(_lock_path()):
+        raw_entries = _read_entries(state_path)
+        entries = _prune_dead(raw_entries)
+        pruned = len(raw_entries) - len(entries)
+        if pruned:
+            logger.info("Pruned %d stale active session lease(s)", pruned)
+        active_count = len(entries)
+        if active_count >= max_sessions:
+            _write_entries(state_path, entries)
+            logger.info(
+                "Active session limit reached: active=%d max=%d surface=%s",
+                active_count,
+                max_sessions,
+                surface,
+            )
+            return None, active_session_limit_message(active_count, max_sessions)
+        entries.append(entry)
+        _write_entries(state_path, entries)
+
+    return ActiveSessionLease(
+        lease_id=lease_id,
+        session_id=str(session_id),
+        surface=str(surface),
+    ), None
+
+
+def release_active_session(lease: ActiveSessionLease) -> None:
+    state_path = _state_path()
+    try:
+        with _FileLock(_lock_path()):
+            entries = _prune_dead(_read_entries(state_path))
+            kept = [
+                entry
+                for entry in entries
+                if str(entry.get("lease_id") or "") != lease.lease_id
+            ]
+            if len(kept) != len(entries):
+                _write_entries(state_path, kept)
+    finally:
+        lease.released = True
+
+
+def active_session_registry_snapshot() -> list[dict[str, Any]]:
+    """Return the pruned active-session registry for diagnostics/tests."""
+    state_path = _state_path()
+    with _FileLock(_lock_path()):
+        entries = _prune_dead(_read_entries(state_path))
+        _write_entries(state_path, entries)
+        return entries

hermes_cli/config.py

@@ -805,6 +805,9 @@ DEFAULT_CONFIG = {
     "fallback_providers": [],
     "credential_pool_strategies": {},
     "toolsets": ["hermes-cli"],
+    # Global active chat session cap across CLI, TUI/dashboard, and messaging.
+    # None/0 = unbounded.
+    "max_concurrent_sessions": None,
     "agent": {
         "max_turns": 90,
         # Inactivity timeout for gateway agent execution (seconds).

hermes_cli/plugins_cmd.py

@@ -135,34 +135,89 @@ def _sanitize_plugin_name(
     return target
 
 
-def _resolve_git_url(identifier: str) -> str:
-    """Turn an identifier into a cloneable Git URL.
+def _resolve_git_url(identifier: str) -> tuple[str, Optional[str]]:
+    """Turn an identifier into a cloneable Git URL and optional subdirectory.
+
+    Returns ``(git_url, subdir)`` where ``subdir`` is the path within the
+    cloned repository that contains the plugin (``None`` when the plugin lives
+    at the repo root).
 
     Accepted formats:
     - Full URL: https://github.com/owner/repo.git
     - Full URL: git@github.com:owner/repo.git
     - Full URL: ssh://git@github.com/owner/repo.git
     - Shorthand: owner/repo  →  https://github.com/owner/repo.git
+    - Shorthand w/ subdir: owner/repo/path/to/plugin
+      →  (https://github.com/owner/repo.git, "path/to/plugin")
+    - Full URL w/ subdir (``.git`` boundary):
+      https://github.com/owner/repo.git/path/to/plugin
+      →  (https://github.com/owner/repo.git, "path/to/plugin")
+    - Any URL w/ explicit subdir fragment (works for every scheme, incl.
+      ``file://`` and ssh): <url>#path/to/plugin
+      →  (<url>, "path/to/plugin")
 
     NOTE: ``http://`` and ``file://`` schemes are accepted but will trigger a
     security warning at install time.
     """
-    # Already a URL
+    # Already a URL.
     if identifier.startswith(("https://", "http://", "git@", "ssh://", "file://")):
-        return identifier
+        # Explicit ``#subdir`` fragment — unambiguous for any scheme.
+        if "#" in identifier:
+            git_url, _, frag = identifier.partition("#")
+            return git_url, (frag.strip("/") or None)
+        # Natural ``.git/`` boundary (GitHub-style URLs).
+        marker = ".git/"
+        idx = identifier.find(marker)
+        if idx != -1:
+            git_url = identifier[: idx + len(".git")]
+            subdir = identifier[idx + len(marker) :].strip("/")
+            return git_url, (subdir or None)
+        return identifier, None
 
-    # owner/repo shorthand
-    parts = identifier.strip("/").split("/")
-    if len(parts) == 2:
-        owner, repo = parts
-        return f"https://github.com/{owner}/{repo}.git"
+    # owner/repo[/subdir...] shorthand
+    parts = [p for p in identifier.strip("/").split("/") if p]
+    if len(parts) >= 2:
+        owner, repo = parts[0], parts[1]
+        subdir = "/".join(parts[2:]).strip("/")
+        git_url = f"https://github.com/{owner}/{repo}.git"
+        return git_url, (subdir or None)
 
     raise ValueError(
         f"Invalid plugin identifier: '{identifier}'. "
-        "Use a Git URL or owner/repo shorthand."
+        "Use a Git URL or 'owner/repo' shorthand (optionally with a subdirectory: "
+        "'owner/repo/path/to/plugin')."
     )
 
 
+def _resolve_subdir_within(clone_root: Path, subdir: str) -> Path:
+    """Resolve ``subdir`` inside ``clone_root``, rejecting path traversal.
+
+    Guards against ``..`` segments, absolute paths, and symlinks that would
+    escape the cloned repository. Returns the resolved directory path.
+    Raises ``PluginOperationError`` if the path escapes the clone, doesn't
+    exist, or is not a directory.
+    """
+    clone_root = clone_root.resolve()
+    candidate = (clone_root / subdir).resolve()
+
+    # The resolved candidate must stay within the clone root.
+    if candidate != clone_root and clone_root not in candidate.parents:
+        raise PluginOperationError(
+            f"Plugin subdirectory '{subdir}' escapes the repository.",
+        )
+
+    if not candidate.exists():
+        raise PluginOperationError(
+            f"Plugin subdirectory '{subdir}' does not exist in the repository.",
+        )
+    if not candidate.is_dir():
+        raise PluginOperationError(
+            f"Plugin subdirectory '{subdir}' is not a directory.",
+        )
+
+    return candidate
+
+
 def _repo_name_from_url(url: str) -> str:
     """Extract the repo name from a Git URL for the plugin directory name."""
     # Strip trailing .git and slashes
@@ -372,14 +427,14 @@ def _install_plugin_core(identifier: str, *, force: bool) -> tuple[Path, dict, s
     import tempfile
 
     try:
-        git_url = _resolve_git_url(identifier)
+        git_url, subdir = _resolve_git_url(identifier)
     except ValueError as e:
         raise PluginOperationError(str(e)) from e
 
     plugins_dir = _plugins_dir()
 
     with tempfile.TemporaryDirectory() as tmp:
-        tmp_target = Path(tmp) / "plugin"
+        tmp_clone = Path(tmp) / "plugin"
 
         git_exe = _resolve_git_executable()
         if not git_exe:
@@ -387,7 +442,7 @@ def _install_plugin_core(identifier: str, *, force: bool) -> tuple[Path, dict, s
 
         try:
             result = subprocess.run(
-                [git_exe, "clone", "--depth", "1", git_url, str(tmp_target)],
+                [git_exe, "clone", "--depth", "1", git_url, str(tmp_clone)],
                 capture_output=True,
                 text=True,
                 timeout=60,
@@ -405,8 +460,16 @@ def _install_plugin_core(identifier: str, *, force: bool) -> tuple[Path, dict, s
             err = (result.stderr or result.stdout or "").strip()
             raise PluginOperationError(f"Git clone failed:\n{err}")
 
+        # Resolve the directory within the clone that holds the plugin.
+        if subdir:
+            tmp_target = _resolve_subdir_within(tmp_clone, subdir)
+        else:
+            tmp_target = tmp_clone
+
         manifest = _read_manifest(tmp_target)
-        plugin_name = manifest.get("name") or _repo_name_from_url(git_url)
+        plugin_name = manifest.get("name") or (
+            subdir.rstrip("/").rsplit("/", 1)[-1] if subdir else _repo_name_from_url(git_url)
+        )
 
         try:
             target = _sanitize_plugin_name(plugin_name, plugins_dir)
@@ -471,7 +534,7 @@ def cmd_install(
     console = Console()
 
     try:
-        git_url = _resolve_git_url(identifier)
+        git_url, _subdir = _resolve_git_url(identifier)
     except ValueError as e:
         console.print(f"[red]Error:[/red] {e}")
         sys.exit(1)
@@ -482,7 +545,10 @@ def cmd_install(
             "Consider using https:// or git@ for production installs.",
         )
 
-    console.print(f"[dim]Cloning {git_url}...[/dim]")
+    if _subdir:
+        console.print(f"[dim]Cloning {git_url} (subdir: {_subdir})...[/dim]")
+    else:
+        console.print(f"[dim]Cloning {git_url}...[/dim]")
 
     try:
         target, installed_manifest, installed_name = _install_plugin_core(
@@ -1473,7 +1539,7 @@ def dashboard_install_plugin(
     """Non-interactive install for the web dashboard. Returns a JSON-serializable dict."""
     warnings: list[str] = []
     try:
-        git_url = _resolve_git_url(identifier)
+        git_url, _subdir = _resolve_git_url(identifier)
         if git_url.startswith(("http://", "file://")):
             warnings.append(
                 "Insecure URL scheme; prefer https:// or git@ for production installs.",

plugins/observability/langfuse/__init__.py

@@ -837,8 +837,16 @@ def on_post_llm_call(*, task_id: str = "", session_id: str = "", provider: str =
     if output.get("tool_calls"):
         state.turn_tool_calls.extend(output["tool_calls"])
 
-    # Extract usage: prefer response object, fall back to usage dict from post_api_request
-    if response is not None:
+    # Extract usage: prefer a real response object that carries usage, else
+    # fall back to the usage summary dict from post_api_request.
+    #
+    # post_api_request passes `response` as a SANITIZED dict (no ``.usage``
+    # attribute) alongside a separate `usage` summary dict. Gating on
+    # ``response is not None`` here took the response-object path on that dict,
+    # where ``getattr(response, "usage", None)`` is always None — so usage and
+    # cost were silently dropped for every gateway turn. Gate on a real
+    # ``.usage`` attribute instead so the usage-dict fallback below is reached.
+    if getattr(response, "usage", None) is not None:
         usage_details, cost_details = _usage_and_cost(
             response,
             provider=provider,

plugins/observability/nemo_relay/README.md

@@ -177,8 +177,8 @@ include an adaptive component in the same `plugins.toml`:
 kind = "adaptive"
 enabled = true
 
-[components.config]
-mode = "route"
+[components.config.tool_parallelism]
+mode = "observe_only"
 ```
 
 When the adaptive component is enabled and the installed NeMo Relay runtime
@@ -186,15 +186,16 @@ exposes `llm.execute(...)` / `tools.execute(...)`, Hermes routes LLM and tool
 execution through those middleware boundaries. The observer hooks still emit
 session, turn, approval, and subagent marks; the plugin skips its manual
 `llm.call` and `tools.call` spans for executions that are already managed by
-NeMo Relay.
+NeMo Relay. `tool_parallelism.mode = "observe_only"` keeps tool scheduling
+observational while still wrapping the real execution boundary.
 
 For the full generic Hermes middleware contract, see
 [`docs/middleware/README.md`](../../../docs/middleware/README.md).
 
 ## Canonical Local Examples
 
-The examples below use the official `nemo-relay==0.3` distribution and a local
-Ollama model served through the OpenAI-compatible API.
+The observe-only examples in this section use the official `nemo-relay==0.3`
+distribution and a local Ollama model served through the OpenAI-compatible API.
 
 ```bash
 pip install "nemo-relay==0.3"
@@ -408,8 +409,8 @@ version = 1
 kind = "adaptive"
 enabled = true
 
-[components.config]
-mode = "route"
+[components.config.tool_parallelism]
+mode = "observe_only"
 ```
 
 Enable it for Hermes:
@@ -442,11 +443,12 @@ for the same execution.
 ### Local Adaptive E2E
 
 This example enables both NeMo Relay observability export and adaptive execution
-middleware for a local Hermes run.
+middleware for a local Hermes run. This path requires a NeMo Relay runtime that
+supports `[components.config.tool_parallelism]`; the `nemo-relay==0.3`
+install used by the earlier observability-only examples does not support this
+adaptive config.
 
 ```bash
-pip install "nemo-relay==0.3"
-
 export HERMES_HOME=/tmp/hermes-middleware-test/hermes-home
 mkdir -p "$HERMES_HOME" /tmp/hermes-middleware-test/nemo-relay
 
@@ -488,8 +490,8 @@ agent_version = "local"
 kind = "adaptive"
 enabled = true
 
-[components.config]
-mode = "route"
+[components.config.tool_parallelism]
+mode = "observe_only"
 TOML
 
 export HERMES_NEMO_RELAY_PLUGINS_TOML=/tmp/hermes-middleware-test/nemo-relay/plugins.toml
@@ -514,8 +516,8 @@ middleware_execution_ok
 Expected ATOF shape:
 
 ```jsonl
-{"kind":"scope","category":"llm","name":"custom","scope_category":"start","metadata":{"session_id":"middleware-demo-session"},"data":{"mode":"route"}}
-{"kind":"scope","category":"tool","name":"terminal","scope_category":"start","metadata":{"session_id":"middleware-demo-session","tool_call_id":"call_terminal"},"data":{"mode":"route"}}
+{"kind":"scope","category":"llm","name":"custom","scope_category":"start","metadata":{"session_id":"middleware-demo-session"},"data":{"mode":"observe_only"}}
+{"kind":"scope","category":"tool","name":"terminal","scope_category":"start","metadata":{"session_id":"middleware-demo-session","tool_call_id":"call_terminal"},"data":{"mode":"observe_only"}}
 {"kind":"scope","category":"tool","name":"terminal","scope_category":"end","metadata":{"session_id":"middleware-demo-session","tool_call_id":"call_terminal","status":"ok"},"data":"{\"output\":\"middleware_execution_ok\",\"exit_code\":0,\"error\":null}"}
 ```
 

plugins/observability/nemo_relay/__init__.py

@@ -44,7 +44,7 @@ class _Settings:
     plugins_toml_path: str = ""
     plugins_config: dict[str, Any] | None = None
     adaptive_enabled: bool = False
-    adaptive_mode: str = "observe"
+    adaptive_mode: str = "observe_only"
     atof_enabled: bool = False
     atof_output_directory: str = ""
     atof_filename: str = "hermes-atof.jsonl"
@@ -660,11 +660,16 @@ def _enabled_component_config(
 
 def _adaptive_mode(config: dict[str, Any] | None) -> str:
     if not isinstance(config, dict):
-        return "observe"
+        return "observe_only"
+    tool_parallelism = config.get("tool_parallelism")
+    if isinstance(tool_parallelism, dict):
+        mode = tool_parallelism.get("mode")
+        if isinstance(mode, str) and mode.strip():
+            return mode.strip()
     mode = config.get("mode")
     if isinstance(mode, str) and mode.strip():
         return mode.strip()
-    return "observe"
+    return "observe_only"
 
 
 def _observability_exporter_enabled(

plugins/platforms/photon/README.md

@@ -106,14 +106,16 @@ All env vars are documented in `plugin.yaml`. The most important are:
 
 ## Limitations (current Photon API)
 
-- **Attachments are metadata only.** Inbound webhooks include the
+- **Inbound attachments are metadata only.** Inbound webhooks include the
   filename + MIME type but no download URL. The plugin surfaces a
   text marker (`[Photon attachment received: …]`) so the agent knows
   something arrived, but cannot read the bytes.  Photon's docs note
   an attachment retrieval endpoint is on the roadmap.
-- **Outbound attachments are not supported yet.** Adding them is
-  straightforward once the sidecar wires up `attachment(...)` /
-  `space.send(attachment(...))` from `spectrum-ts`.
+- **Outbound attachments are supported.** Images, voice notes, video,
+  and documents are sent via `space.send(attachment(...))` /
+  `space.send(voice(...))` through the sidecar's `/send-attachment`
+  endpoint. A caption is delivered as a separate text bubble after the
+  media.
 - **Reactions, message effects, polls** — not exposed yet; the
   `spectrum-ts` SDK supports them, and the sidecar is the natural
   place to add them when the agent has reason to use them.

plugins/platforms/photon/adapter.py

@@ -14,8 +14,10 @@ Outbound:
     Photon does not currently expose a public HTTP send-message
     endpoint, so the adapter spawns a small Node sidecar (see
     ``sidecar/index.mjs``) that runs the ``spectrum-ts`` SDK.  Each
-    ``send`` / ``send_typing`` call from Hermes is a loopback POST to
-    the sidecar with a shared bearer token.
+    ``send`` / ``send_typing`` / attachment call from Hermes is a
+    loopback POST to the sidecar with a shared bearer token.  Outbound
+    media (images, voice notes, video, documents) goes through
+    spectrum-ts' ``attachment()`` / ``voice()`` content builders.
 
 When Photon ships an HTTP send endpoint we can collapse the sidecar
 into ``_send_via_http`` and drop the Node dependency entirely.
@@ -670,6 +672,99 @@ class PhotonAdapter(BasePlatformAdapter):
     ) -> SendResult:
         return await self._sidecar_send(chat_id, content, reply_to=reply_to)
 
+    # -- Outbound media (parity with the BlueBubbles iMessage channel) -----
+    #
+    # Photon ships outbound attachments via spectrum-ts' `attachment()` /
+    # `voice()` content builders. The sidecar's `/send-attachment` endpoint
+    # wraps `space.send(attachment(path, {...}))`. These overrides mirror
+    # BlueBubbles: URL-based helpers cache to a local path first, file-based
+    # helpers pass the path straight through.
+
+    async def send_image(
+        self,
+        chat_id: str,
+        image_url: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        try:
+            from gateway.platforms.base import cache_image_from_url
+
+            local_path = await cache_image_from_url(image_url)
+        except Exception:
+            # Couldn't fetch the URL — fall back to sending it as text.
+            return await super().send_image(chat_id, image_url, caption, reply_to)
+        return await self._sidecar_send_attachment(
+            chat_id, local_path, caption=caption, reply_to=reply_to,
+        )
+
+    async def send_image_file(
+        self,
+        chat_id: str,
+        image_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> SendResult:
+        return await self._sidecar_send_attachment(
+            chat_id, image_path, caption=caption, reply_to=reply_to,
+        )
+
+    async def send_voice(
+        self,
+        chat_id: str,
+        audio_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> SendResult:
+        return await self._sidecar_send_attachment(
+            chat_id, audio_path, caption=caption, reply_to=reply_to, kind="voice",
+        )
+
+    async def send_video(
+        self,
+        chat_id: str,
+        video_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> SendResult:
+        return await self._sidecar_send_attachment(
+            chat_id, video_path, caption=caption, reply_to=reply_to,
+        )
+
+    async def send_document(
+        self,
+        chat_id: str,
+        file_path: str,
+        caption: Optional[str] = None,
+        file_name: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> SendResult:
+        return await self._sidecar_send_attachment(
+            chat_id, file_path, name=file_name, caption=caption, reply_to=reply_to,
+        )
+
+    async def send_animation(
+        self,
+        chat_id: str,
+        animation_url: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        # iMessage renders GIFs inline as ordinary image attachments.
+        return await self.send_image(
+            chat_id, animation_url, caption, reply_to, metadata,
+        )
+
     async def send_typing(self, chat_id: str, metadata=None) -> None:
         try:
             await self._sidecar_call("/typing", {"spaceId": chat_id})
@@ -704,6 +799,57 @@ class PhotonAdapter(BasePlatformAdapter):
             return SendResult(success=False, error=str(e))
         return SendResult(success=True, message_id=data.get("messageId"))
 
+    async def _sidecar_send_attachment(
+        self,
+        space_id: str,
+        path: str,
+        *,
+        name: Optional[str] = None,
+        mime_type: Optional[str] = None,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        kind: str = "attachment",
+    ) -> SendResult:
+        """POST a local file to the sidecar's ``/send-attachment`` endpoint.
+
+        ``kind`` is ``"voice"`` for audio sent as a voice note (downgrades
+        to a plain audio attachment on platforms without voice notes),
+        otherwise ``"attachment"``. spectrum-ts infers ``name`` and
+        ``mimeType`` from the file extension; we only pass overrides when
+        Hermes supplied them.
+        """
+        # Defense-in-depth: re-validate the path before handing it to the
+        # Node sidecar. The gateway already filters MEDIA paths, but
+        # send_*_file / cron callers may pass arbitrary strings.
+        safe_path = self.validate_media_delivery_path(str(path))
+        if not safe_path:
+            return SendResult(
+                success=False, error=f"unsafe or missing attachment path: {path}"
+            )
+        if not mime_type:
+            import mimetypes
+
+            guessed, _ = mimetypes.guess_type(safe_path)
+            mime_type = guessed or None
+        body: Dict[str, Any] = {
+            "spaceId": space_id,
+            "path": safe_path,
+            "kind": "voice" if kind == "voice" else "attachment",
+        }
+        if name:
+            body["name"] = name
+        if mime_type:
+            body["mimeType"] = mime_type
+        if caption:
+            body["caption"] = caption
+        if reply_to:
+            body["replyTo"] = reply_to
+        try:
+            data = await self._sidecar_call("/send-attachment", body)
+        except Exception as e:
+            return SendResult(success=False, error=str(e))
+        return SendResult(success=True, message_id=data.get("messageId"))
+
     async def _sidecar_call(self, path: str, body: Dict[str, Any]) -> Dict[str, Any]:
         if self._http_client is None:
             raise RuntimeError("Photon adapter not connected")
@@ -753,8 +899,8 @@ async def _standalone_send(
     message: str,
     *,
     thread_id: Optional[str] = None,  # noqa: ARG001 — Spectrum has no threads yet
-    media_files: Optional[list] = None,  # noqa: ARG001 — attachment send not supported yet
-    force_document: bool = False,  # noqa: ARG001
+    media_files: Optional[list] = None,
+    force_document: bool = False,  # noqa: ARG001 — iMessage auto-detects file kind
 ) -> Dict[str, Any]:
     if not HTTPX_AVAILABLE:
         return {"error": "httpx not installed"}
@@ -771,20 +917,54 @@ async def _standalone_send(
                 "cannot spawn the sidecar themselves."
             )
         }
-    body: Dict[str, Any] = {"spaceId": chat_id, "text": message[:_MAX_MESSAGE_LENGTH]}
+    base = f"http://{_DEFAULT_SIDECAR_BIND}:{port}"
+    headers = {"X-Hermes-Sidecar-Token": token}
+    last_message_id: Optional[str] = None
     try:
         async with httpx.AsyncClient(timeout=30.0) as client:
-            resp = await client.post(
-                f"http://{_DEFAULT_SIDECAR_BIND}:{port}/send",
-                json=body,
-                headers={"X-Hermes-Sidecar-Token": token},
-            )
-        if resp.status_code != 200:
-            return {"error": f"sidecar returned {resp.status_code}: {resp.text[:200]}"}
-        data = resp.json() or {}
-        if not data.get("ok"):
-            return {"error": data.get("error") or "sidecar reported failure"}
-        return {"success": True, "message_id": data.get("messageId")}
+            # 1. Text body first (if any), so it leads the conversation.
+            if message:
+                resp = await client.post(
+                    f"{base}/send",
+                    json={"spaceId": chat_id, "text": message[:_MAX_MESSAGE_LENGTH]},
+                    headers=headers,
+                )
+                if resp.status_code != 200:
+                    return {"error": f"sidecar returned {resp.status_code}: {resp.text[:200]}"}
+                data = resp.json() or {}
+                if not data.get("ok"):
+                    return {"error": data.get("error") or "sidecar reported failure"}
+                last_message_id = data.get("messageId")
+
+            # 2. Each attachment as a separate /send-attachment call.
+            #    media_files is List[Tuple[path, is_voice]] (see
+            #    BasePlatformAdapter.filter_media_delivery_paths).
+            import mimetypes
+
+            for media_path, is_voice in media_files or []:
+                safe_path = BasePlatformAdapter.validate_media_delivery_path(str(media_path))
+                if not safe_path:
+                    logger.warning("[photon] standalone send skipping unsafe path")
+                    continue
+                guessed, _ = mimetypes.guess_type(safe_path)
+                att_body: Dict[str, Any] = {
+                    "spaceId": chat_id,
+                    "path": safe_path,
+                    "kind": "voice" if is_voice else "attachment",
+                }
+                if guessed:
+                    att_body["mimeType"] = guessed
+                resp = await client.post(
+                    f"{base}/send-attachment", json=att_body, headers=headers,
+                )
+                if resp.status_code != 200:
+                    return {"error": f"sidecar returned {resp.status_code}: {resp.text[:200]}"}
+                data = resp.json() or {}
+                if not data.get("ok"):
+                    return {"error": data.get("error") or "sidecar reported failure"}
+                last_message_id = data.get("messageId") or last_message_id
+
+        return {"success": True, "message_id": last_message_id}
     except Exception as e:
         return {"error": f"Photon standalone send failed: {e}"}
 

plugins/platforms/photon/sidecar/index.mjs

@@ -12,6 +12,10 @@
 //   - POST /healthz                     -> {"ok": true}
 //   - POST /send                        -> {"ok": true, "messageId": "..."}
 //       body: {"spaceId": "...", "text": "...", "replyTo": "..." | null}
+//   - POST /send-attachment             -> {"ok": true, "messageId": "..."}
+//       body: {"spaceId": "...", "path": "...", "name": "..." | null,
+//              "mimeType": "..." | null, "caption": "..." | null,
+//              "kind": "attachment" | "voice", "replyTo": "..." | null}
 //   - POST /typing                      -> {"ok": true}
 //       body: {"spaceId": "..."}
 //   - POST /shutdown                    -> {"ok": true}; then process exits
@@ -48,9 +52,9 @@ if (!projectId || !projectSecret || !sharedToken) {
 
 // Lazy-load spectrum-ts so a missing install fails with a clear message
 // instead of a cryptic module-resolution error during import.
-let Spectrum, imessage;
+let Spectrum, imessage, attachment, voice;
 try {
-  ({ Spectrum } = await import("spectrum-ts"));
+  ({ Spectrum, attachment, voice } = await import("spectrum-ts"));
   ({ imessage } = await import("spectrum-ts/providers/imessage"));
 } catch (e) {
   console.error(
@@ -179,6 +183,44 @@ const server = http.createServer(async (req, res) => {
         : await space.send(text);
       return ok(res, { messageId: result?.id || result?.messageId || null });
     }
+    if (req.url === "/send-attachment") {
+      const { spaceId, path, name, mimeType, caption, kind, replyTo } =
+        body || {};
+      if (!spaceId || typeof path !== "string" || !path) {
+        return badRequest(res, "spaceId and path are required");
+      }
+      const space = await resolveSpace(spaceId);
+
+      // spectrum-ts infers name + MIME from the file extension; pass
+      // overrides only when Hermes supplied them so a known-good
+      // inference isn't clobbered with an empty string.
+      const opts = {};
+      if (name) opts.name = name;
+      if (mimeType) opts.mimeType = mimeType;
+      const builder =
+        kind === "voice"
+          ? voice(path, Object.keys(opts).length ? opts : undefined)
+          : attachment(path, Object.keys(opts).length ? opts : undefined);
+
+      const sendOpts = replyTo ? { replyTo } : undefined;
+      const result = sendOpts
+        ? await space.send(builder, sendOpts)
+        : await space.send(builder);
+
+      // iMessage delivers the caption as a separate bubble; send it
+      // after the media so the attachment renders first.
+      if (caption && typeof caption === "string") {
+        try {
+          await space.send(caption);
+        } catch (e) {
+          console.error(
+            "photon-sidecar: attachment sent but caption failed: " +
+              (e && e.stack ? e.stack : String(e))
+          );
+        }
+      }
+      return ok(res, { messageId: result?.id || result?.messageId || null });
+    }
     if (req.url === "/typing") {
       const { spaceId } = body || {};
       if (!spaceId) return badRequest(res, "spaceId is required");

scripts/build_skills_index.py

@@ -297,6 +297,21 @@ def main():
     # Batch resolve GitHub paths for skills.sh entries
     all_skills = batch_resolve_paths(all_skills, auth)
 
+    # Collect which sources hit a GitHub API rate limit during the crawl.
+    # github / claude-marketplace / well-known all read api.github.com, so a
+    # rate-limited token zeroes all three at once — surfaced below so the
+    # failure message names the real cause instead of "source returned 0".
+    rate_limited_sources = {
+        name for name, source in sources.items()
+        if getattr(source, "is_rate_limited", False)
+    }
+    if rate_limited_sources:
+        print(
+            "  WARNING: GitHub API rate limit hit for: "
+            + ", ".join(sorted(rate_limited_sources)),
+            file=sys.stderr,
+        )
+
     # Deduplicate by identifier
     seen: dict[str, dict] = {}
     for skill in all_skills:
@@ -311,25 +326,9 @@ def main():
                     "browse-sh": 5, "claude-marketplace": 6, "lobehub": 7}
     deduped.sort(key=lambda s: (source_order.get(s["source"], 99), s["name"]))
 
-    # Build index
-    index = {
-        "version": INDEX_VERSION,
-        "generated_at": datetime.now(timezone.utc).isoformat(),
-        "skill_count": len(deduped),
-        "skills": deduped,
-    }
-
-    os.makedirs(os.path.dirname(OUTPUT_PATH), exist_ok=True)
-    with open(OUTPUT_PATH, "w", encoding="utf-8") as f:
-        json.dump(index, f, separators=(",", ":"), ensure_ascii=False)
-
-    elapsed = time.time() - overall_start
-    file_size = os.path.getsize(OUTPUT_PATH)
-    print(f"\nDone! {len(deduped)} skills indexed in {elapsed:.0f}s")
-    print(f"Output: {OUTPUT_PATH} ({file_size / 1024:.0f} KB)")
-
     from collections import Counter
     by_source = Counter(s["source"] for s in deduped)
+    print(f"\nCrawled {len(deduped)} skills in {time.time() - overall_start:.0f}s")
     for src, count in sorted(by_source.items(), key=lambda x: -x[1]):
         resolved = sum(1 for s in deduped
                        if s["source"] == src and s.get("resolved_github_id"))
@@ -380,14 +379,46 @@ def main():
         )
         for line in health_errors:
             print(line, file=sys.stderr)
+        if rate_limited_sources:
+            print(
+                "\nGitHub API rate limit was hit during this crawl for: "
+                + ", ".join(sorted(rate_limited_sources))
+                + ". This is the usual cause of an all-GitHub-tap collapse "
+                "(github / claude-marketplace / well-known dropping to zero "
+                "together). Re-run with a higher-quota GITHUB_TOKEN.",
+                file=sys.stderr,
+            )
         print(
             "\nIf the drop is expected (e.g. a hub is genuinely shutting "
             "down), lower the floor in scripts/build_skills_index.py "
             "EXPECTED_FLOORS in the same PR.",
             file=sys.stderr,
         )
+        # IMPORTANT: do NOT write OUTPUT_PATH on failure. The index file is
+        # gitignored, so a fresh deploy checkout has no copy on disk — leaving
+        # it absent lets website/scripts/extract-skills.py fall back to the
+        # legacy snapshot cache (or skip the unified index) instead of reading
+        # a degenerate file. Writing-then-exiting-2 was the bug that shipped an
+        # index with every GitHub-API source dropped to zero: deploy-site.yml
+        # swallows the exit code with `|| echo non-fatal`, and the partial file
+        # was already on disk for extract-skills to pick up.
         sys.exit(2)
 
+    # Healthy — only now write the index out for the docs build to consume.
+    index = {
+        "version": INDEX_VERSION,
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "skill_count": len(deduped),
+        "skills": deduped,
+    }
+    os.makedirs(os.path.dirname(OUTPUT_PATH), exist_ok=True)
+    with open(OUTPUT_PATH, "w", encoding="utf-8") as f:
+        json.dump(index, f, separators=(",", ":"), ensure_ascii=False)
+    file_size = os.path.getsize(OUTPUT_PATH)
+    print(f"\nDone! {len(deduped)} skills indexed in "
+          f"{time.time() - overall_start:.0f}s")
+    print(f"Output: {OUTPUT_PATH} ({file_size / 1024:.0f} KB)")
+
 
 if __name__ == "__main__":
     main()

tests/gateway/test_config.py

@@ -1,5 +1,6 @@
 """Tests for gateway configuration management."""
 
+import logging
 import os
 from unittest.mock import patch
 
@@ -213,6 +214,43 @@ class TestGatewayConfigRoundtrip:
         assert restored.group_sessions_per_user is False
         assert restored.thread_sessions_per_user is True
 
+    def test_max_concurrent_sessions_from_dict_normalizes_disabled_values(self):
+        assert GatewayConfig.from_dict({}).max_concurrent_sessions is None
+        assert GatewayConfig.from_dict({"max_concurrent_sessions": None}).max_concurrent_sessions is None
+        assert GatewayConfig.from_dict({"max_concurrent_sessions": 0}).max_concurrent_sessions is None
+        assert GatewayConfig.from_dict({"max_concurrent_sessions": -1}).max_concurrent_sessions is None
+
+    def test_max_concurrent_sessions_from_dict_accepts_positive_integer(self):
+        config = GatewayConfig.from_dict({"max_concurrent_sessions": "3"})
+
+        assert config.max_concurrent_sessions == 3
+
+    def test_max_concurrent_sessions_from_dict_ignores_invalid_values(self, caplog):
+        caplog.set_level(logging.WARNING, logger="gateway.config")
+
+        config = GatewayConfig.from_dict({"max_concurrent_sessions": "many"})
+
+        assert config.max_concurrent_sessions is None
+        assert any(
+            "Ignoring invalid max_concurrent_sessions='many'" in record.message
+            for record in caplog.records
+        )
+
+    def test_max_concurrent_sessions_from_dict_accepts_nested_fallback(self):
+        config = GatewayConfig.from_dict({"gateway": {"max_concurrent_sessions": 4}})
+
+        assert config.max_concurrent_sessions == 4
+
+    def test_max_concurrent_sessions_top_level_overrides_nested(self):
+        config = GatewayConfig.from_dict(
+            {
+                "gateway": {"max_concurrent_sessions": 4},
+                "max_concurrent_sessions": 2,
+            }
+        )
+
+        assert config.max_concurrent_sessions == 2
+
     def test_roundtrip_preserves_unauthorized_dm_behavior(self):
         config = GatewayConfig(
             unauthorized_dm_behavior="ignore",
@@ -309,6 +347,51 @@ class TestLoadGatewayConfig:
 
         assert config.thread_sessions_per_user is False
 
+    def test_bridges_top_level_max_concurrent_sessions_from_config_yaml(self, tmp_path, monkeypatch):
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        config_path = hermes_home / "config.yaml"
+        config_path.write_text("max_concurrent_sessions: 2\n", encoding="utf-8")
+
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        config = load_gateway_config()
+
+        assert config.max_concurrent_sessions == 2
+
+    def test_bridges_nested_max_concurrent_sessions_from_config_yaml(self, tmp_path, monkeypatch):
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        config_path = hermes_home / "config.yaml"
+        config_path.write_text(
+            "gateway:\n"
+            "  max_concurrent_sessions: 3\n",
+            encoding="utf-8",
+        )
+
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        config = load_gateway_config()
+
+        assert config.max_concurrent_sessions == 3
+
+    def test_top_level_max_concurrent_sessions_overrides_nested_config_yaml(self, tmp_path, monkeypatch):
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        config_path = hermes_home / "config.yaml"
+        config_path.write_text(
+            "max_concurrent_sessions: 2\n"
+            "gateway:\n"
+            "  max_concurrent_sessions: 3\n",
+            encoding="utf-8",
+        )
+
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        config = load_gateway_config()
+
+        assert config.max_concurrent_sessions == 2
+
     def test_bridges_discord_thread_require_mention_from_config_yaml(self, tmp_path, monkeypatch):
         """discord.thread_require_mention in config.yaml should reach the runtime env var."""
         hermes_home = tmp_path / ".hermes"

tests/gateway/test_max_concurrent_sessions.py

@@ -0,0 +1,208 @@
+"""Tests for the gateway max_concurrent_sessions active-session cap."""
+
+import asyncio
+import time
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform, PlatformConfig
+from gateway.platforms.base import MessageEvent, MessageType
+from gateway.run import GatewayRunner, _AGENT_PENDING_SENTINEL
+from gateway.session import SessionSource, build_session_key
+
+
+@pytest.fixture(autouse=True)
+def _isolated_active_session_registry(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+
+
+class _FakeAdapter:
+    def __init__(self):
+        self._pending_messages = {}
+        self._active_sessions = {}
+
+    async def send(self, chat_id, text, **kwargs):
+        return None
+
+    async def interrupt_session_activity(self, session_key, chat_id):
+        event = self._active_sessions.get(session_key)
+        if event is not None:
+            event.set()
+
+
+def _make_source(chat_id: str = "chat-1") -> SessionSource:
+    return SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id=chat_id,
+        chat_type="dm",
+        user_id=f"user-{chat_id}",
+    )
+
+
+def _make_event(text: str = "hello", chat_id: str = "chat-1") -> MessageEvent:
+    return MessageEvent(
+        text=text,
+        message_type=MessageType.TEXT,
+        source=_make_source(chat_id),
+    )
+
+
+def _make_runner(max_concurrent_sessions: int | None = None) -> GatewayRunner:
+    runner = object.__new__(GatewayRunner)
+    runner.config = GatewayConfig(
+        platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")},
+        max_concurrent_sessions=max_concurrent_sessions,
+    )
+    runner.adapters = {Platform.TELEGRAM: _FakeAdapter()}
+    runner._running_agents = {}
+    runner._running_agents_ts = {}
+    runner._active_session_leases = {}
+    runner._session_run_generation = {}
+    runner._pending_messages = {}
+    runner._pending_approvals = {}
+    runner._voice_mode = {}
+    runner._background_tasks = set()
+    runner._draining = False
+    runner._restart_requested = False
+    runner._restart_task_started = False
+    runner._restart_detached = False
+    runner._restart_via_service = False
+    runner._restart_drain_timeout = 0.0
+    runner._stop_task = None
+    runner._exit_code = None
+    runner._busy_ack_ts = {}
+    runner._busy_input_mode = "interrupt"
+    runner._busy_text_mode = "interrupt"
+    runner._queued_events = {}
+    runner._update_runtime_status = MagicMock()
+    runner._is_user_authorized = lambda _source: True
+    runner.hooks = MagicMock()
+    runner.hooks.emit = AsyncMock()
+    runner.session_store = MagicMock()
+    runner.delivery_router = MagicMock()
+    return runner
+
+
+def _occupy_session(runner: GatewayRunner, chat_id: str = "busy"):
+    source = _make_source(chat_id)
+    session_key = build_session_key(source)
+    runner._running_agents[session_key] = MagicMock()
+    runner._running_agents_ts[session_key] = time.time()
+    return session_key
+
+
+def _silence_global_gateway_hooks(monkeypatch):
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *args, **kwargs: [])
+    monkeypatch.setattr("tools.slash_confirm.get_pending", lambda *args, **kwargs: None)
+    monkeypatch.setattr("tools.slash_confirm.clear_if_stale", lambda *args, **kwargs: None)
+    monkeypatch.setattr("tools.approval.has_blocking_approval", lambda *args, **kwargs: False)
+
+
+def test_new_session_gets_clean_error_at_active_session_limit(monkeypatch):
+    _silence_global_gateway_hooks(monkeypatch)
+    runner = _make_runner(max_concurrent_sessions=1)
+    _occupy_session(runner, "busy")
+    event = _make_event(chat_id="new")
+    new_key = build_session_key(event.source)
+
+    async def fail_if_agent_runs(self_inner, ev, src, qk, generation):
+        raise AssertionError("_handle_message_with_agent should not run at capacity")
+
+    with patch.object(GatewayRunner, "_handle_message_with_agent", fail_if_agent_runs):
+        result = asyncio.run(runner._handle_message(event))
+
+    assert result == (
+        "Hermes is at the active session limit (1/1). "
+        "Try again when another session finishes."
+    )
+    assert new_key not in runner._running_agents
+    runner.session_store.get_or_create_session.assert_not_called()
+
+
+def test_existing_active_session_uses_busy_handling_at_limit(monkeypatch):
+    _silence_global_gateway_hooks(monkeypatch)
+    runner = _make_runner(max_concurrent_sessions=1)
+    runner._busy_input_mode = "queue"
+    event = _make_event(chat_id="busy")
+    session_key = build_session_key(event.source)
+    runner._running_agents[session_key] = MagicMock()
+    runner._running_agents_ts[session_key] = 0
+
+    async def fail_if_agent_runs(self_inner, ev, src, qk, generation):
+        raise AssertionError("_handle_message_with_agent should not run for busy follow-up")
+
+    with patch.object(GatewayRunner, "_handle_message_with_agent", fail_if_agent_runs):
+        result = asyncio.run(runner._handle_message(event))
+
+    assert result is None
+    assert runner.adapters[Platform.TELEGRAM]._pending_messages[session_key] is event
+
+
+def test_new_session_can_start_after_active_session_released(monkeypatch):
+    _silence_global_gateway_hooks(monkeypatch)
+    runner = _make_runner(max_concurrent_sessions=1)
+    busy_key = _occupy_session(runner, "busy")
+    runner._release_running_agent_state(busy_key)
+    event = _make_event(chat_id="new")
+
+    sentinel_seen = False
+
+    async def mock_agent_run(self_inner, ev, src, qk, generation):
+        nonlocal sentinel_seen
+        sentinel_seen = runner._running_agents.get(qk) is _AGENT_PENDING_SENTINEL
+        return "ok"
+
+    with patch.object(GatewayRunner, "_handle_message_with_agent", mock_agent_run):
+        result = asyncio.run(runner._handle_message(event))
+
+    assert result == "ok"
+    assert sentinel_seen is True
+
+
+def test_status_command_bypasses_active_session_limit(monkeypatch):
+    _silence_global_gateway_hooks(monkeypatch)
+    runner = _make_runner(max_concurrent_sessions=1)
+    _occupy_session(runner, "busy")
+    runner._handle_status_command = AsyncMock(return_value="status ok")
+
+    result = asyncio.run(runner._handle_message(_make_event("/status", chat_id="new")))
+
+    assert result == "status ok"
+    runner._handle_status_command.assert_awaited_once()
+
+
+def test_skill_command_that_would_start_agent_is_blocked_at_limit(monkeypatch):
+    _silence_global_gateway_hooks(monkeypatch)
+    runner = _make_runner(max_concurrent_sessions=1)
+    _occupy_session(runner, "busy")
+
+    monkeypatch.setattr(
+        "agent.skill_commands.get_skill_commands",
+        lambda: {"demo": {"name": "demo-skill"}},
+    )
+    monkeypatch.setattr(
+        "agent.skill_commands.resolve_skill_command_key",
+        lambda command: "demo" if command == "demo" else None,
+    )
+    monkeypatch.setattr(
+        "agent.skill_commands.build_skill_invocation_message",
+        lambda *args, **kwargs: "invoke demo skill",
+    )
+    monkeypatch.setattr(
+        "agent.skill_utils.get_disabled_skill_names",
+        lambda *args, **kwargs: [],
+    )
+
+    async def fail_if_agent_runs(self_inner, ev, src, qk, generation):
+        raise AssertionError("_handle_message_with_agent should not run at capacity")
+
+    with patch.object(GatewayRunner, "_handle_message_with_agent", fail_if_agent_runs):
+        result = asyncio.run(
+            runner._handle_message(_make_event("/demo please", chat_id="new"))
+        )
+
+    assert result == (
+        "Hermes is at the active session limit (1/1). "
+        "Try again when another session finishes."
+    )

tests/gateway/test_shutdown_cache_cleanup.py

@@ -84,6 +84,12 @@ class _FakeGateway:
     def _evict_cached_agent(self, key):
         pass
 
+    def _release_running_agent_state(self, session_key, **_kwargs):
+        agent = self._running_agents.pop(session_key, None)
+        self._running_agents_ts.pop(session_key, None)
+        self._cleanup_agent_resources(agent)
+        return agent is not None
+
 
 def _make_mock_agent():
     a = MagicMock()

tests/hermes_cli/test_active_sessions.py

@@ -0,0 +1,313 @@
+import logging
+import os
+import subprocess
+import sys
+import time
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+from hermes_cli import active_sessions
+
+
+def test_resolve_max_concurrent_sessions_values(caplog):
+    assert active_sessions.resolve_max_concurrent_sessions({}) is None
+    assert active_sessions.resolve_max_concurrent_sessions({"max_concurrent_sessions": None}) is None
+    assert active_sessions.resolve_max_concurrent_sessions({"max_concurrent_sessions": 0}) is None
+    assert active_sessions.resolve_max_concurrent_sessions({"max_concurrent_sessions": -1}) is None
+    assert active_sessions.resolve_max_concurrent_sessions({"max_concurrent_sessions": "3"}) == 3
+    assert (
+        active_sessions.resolve_max_concurrent_sessions(
+            {"gateway": {"max_concurrent_sessions": 4}}
+        )
+        == 4
+    )
+    assert (
+        active_sessions.resolve_max_concurrent_sessions(
+            {"max_concurrent_sessions": 2, "gateway": {"max_concurrent_sessions": 4}}
+        )
+        == 2
+    )
+
+    caplog.set_level(logging.WARNING)
+    assert active_sessions.resolve_max_concurrent_sessions({"max_concurrent_sessions": "many"}) is None
+    assert any(
+        "Ignoring invalid max_concurrent_sessions='many'" in record.message
+        for record in caplog.records
+    )
+
+
+def test_active_session_lease_blocks_until_release(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    cfg = {"max_concurrent_sessions": 1}
+
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="session-1",
+        surface="cli",
+        config=cfg,
+    )
+
+    assert message is None
+    assert lease is not None
+
+    blocked_lease, blocked_message = active_sessions.try_acquire_active_session(
+        session_id="session-2",
+        surface="tui",
+        config=cfg,
+    )
+
+    assert blocked_lease is None
+    assert blocked_message == (
+        "Hermes is at the active session limit (1/1). "
+        "Try again when another session finishes."
+    )
+
+    lease.release()
+
+    next_lease, next_message = active_sessions.try_acquire_active_session(
+        session_id="session-3",
+        surface="gateway:telegram",
+        config=cfg,
+    )
+
+    assert next_message is None
+    assert next_lease is not None
+    next_lease.release()
+    assert active_sessions.active_session_registry_snapshot() == []
+
+
+def test_active_session_registry_prunes_dead_pids(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(
+        "gateway.status._pid_exists",
+        lambda pid: int(pid) != 99999999,
+    )
+    runtime = home / "runtime"
+    runtime.mkdir(parents=True)
+    active_sessions._write_entries(
+        runtime / "active_sessions.json",
+        [
+            {
+                "lease_id": "stale",
+                "session_id": "stale-session",
+                "surface": "cli",
+                "pid": 99999999,
+                "started_at": 1,
+                "updated_at": 1,
+            }
+        ],
+    )
+
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="session-1",
+        surface="cli",
+        config={"max_concurrent_sessions": 1},
+    )
+
+    assert message is None
+    assert lease is not None
+    assert [entry["session_id"] for entry in active_sessions.active_session_registry_snapshot()] == [
+        "session-1"
+    ]
+    lease.release()
+
+
+def test_pid_alive_uses_safe_pid_exists_without_signalling(monkeypatch):
+    checked: list[int] = []
+
+    monkeypatch.setattr(
+        active_sessions.os,
+        "kill",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("os.kill used")),
+    )
+    monkeypatch.setattr(
+        "gateway.status._pid_exists",
+        lambda pid: checked.append(int(pid)) or True,
+    )
+
+    assert active_sessions._pid_alive(12345) is True
+    assert checked == [12345]
+
+
+def test_active_session_hard_exit_is_reclaimed(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    repo_root = Path(__file__).resolve().parents[2]
+    env = os.environ.copy()
+    env["HERMES_HOME"] = str(home)
+    env["PYTHONPATH"] = str(repo_root)
+    child = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            (
+                "import os\n"
+                "from hermes_cli.active_sessions import try_acquire_active_session\n"
+                "lease, message = try_acquire_active_session("
+                "session_id='crash-session', surface='cli', "
+                "config={'max_concurrent_sessions': 1})\n"
+                "assert message is None, message\n"
+                "print(os.getpid(), flush=True)\n"
+                "os._exit(0)\n"
+            ),
+        ],
+        env=env,
+        text=True,
+        capture_output=True,
+        timeout=10,
+        check=True,
+    )
+    child_pid = int(child.stdout.strip())
+
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="next-session",
+        surface="cli",
+        config={"max_concurrent_sessions": 1},
+    )
+
+    assert child_pid > 0
+    assert message is None
+    assert lease is not None
+    assert [entry["session_id"] for entry in active_sessions.active_session_registry_snapshot()] == [
+        "next-session"
+    ]
+    lease.release()
+
+
+def test_concurrent_acquire_claims_only_one_last_slot(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    cfg = {"max_concurrent_sessions": 1}
+
+    def _claim(index: int):
+        return active_sessions.try_acquire_active_session(
+            session_id=f"session-{index}",
+            surface="cli",
+            config=cfg,
+        )
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        results = list(pool.map(_claim, range(8)))
+
+    leases = [lease for lease, message in results if lease is not None and message is None]
+    blocked = [message for lease, message in results if lease is None and message]
+
+    try:
+        assert len(leases) == 1
+        assert len(blocked) == 7
+        assert active_sessions.active_session_registry_snapshot()[0]["session_id"].startswith("session-")
+    finally:
+        for lease in leases:
+            lease.release()
+
+
+def test_cross_process_acquire_claims_only_one_last_slot(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    repo_root = Path(__file__).resolve().parents[2]
+    ready_dir = tmp_path / "ready"
+    ready_dir.mkdir()
+    go_file = tmp_path / "go"
+    env = os.environ.copy()
+    env["HERMES_HOME"] = str(home)
+    env["PYTHONPATH"] = str(repo_root)
+    script = (
+        "import os, time\n"
+        "from pathlib import Path\n"
+        "from hermes_cli.active_sessions import try_acquire_active_session\n"
+        "idx = os.environ['WORKER_INDEX']\n"
+        "ready_dir = Path(os.environ['READY_DIR'])\n"
+        "go_file = Path(os.environ['GO_FILE'])\n"
+        "(ready_dir / idx).write_text('ready', encoding='utf-8')\n"
+        "deadline = time.time() + 10\n"
+        "while not go_file.exists():\n"
+        "    if time.time() > deadline:\n"
+        "        raise RuntimeError('timed out waiting for go file')\n"
+        "    time.sleep(0.01)\n"
+        "lease, message = try_acquire_active_session(\n"
+        "    session_id=f'process-{idx}',\n"
+        "    surface='cli',\n"
+        "    config={'max_concurrent_sessions': 1},\n"
+        ")\n"
+        "if lease is None:\n"
+        "    print('BLOCK', flush=True)\n"
+        "else:\n"
+        "    print('OK', flush=True)\n"
+        "    time.sleep(2.0)\n"
+        "    lease.release()\n"
+    )
+    workers: list[subprocess.Popen[str]] = []
+    try:
+        for index in range(6):
+            worker_env = env.copy()
+            worker_env["WORKER_INDEX"] = str(index)
+            worker_env["READY_DIR"] = str(ready_dir)
+            worker_env["GO_FILE"] = str(go_file)
+            workers.append(
+                subprocess.Popen(
+                    [sys.executable, "-c", script],
+                    env=worker_env,
+                    text=True,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                )
+            )
+
+        deadline = time.time() + 10
+        while len(list(ready_dir.iterdir())) < len(workers):
+            if time.time() > deadline:
+                raise AssertionError("workers did not become ready")
+            time.sleep(0.01)
+        go_file.write_text("go", encoding="utf-8")
+
+        outputs = []
+        for worker in workers:
+            stdout, stderr = worker.communicate(timeout=10)
+            assert worker.returncode == 0, stderr
+            outputs.append(stdout.strip())
+    finally:
+        for worker in workers:
+            if worker.poll() is None:
+                worker.kill()
+                worker.communicate()
+
+    assert outputs.count("OK") == 1
+    assert outputs.count("BLOCK") == len(workers) - 1
+    assert active_sessions.active_session_registry_snapshot() == []
+
+
+def test_pid_start_time_mismatch_prunes_reused_pid(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr("gateway.status._pid_exists", lambda _pid: True)
+    monkeypatch.setattr(active_sessions, "_process_start_time", lambda _pid: 200.0)
+    runtime = home / "runtime"
+    runtime.mkdir(parents=True)
+    active_sessions._write_entries(
+        runtime / "active_sessions.json",
+        [
+            {
+                "lease_id": "stale-reused-pid",
+                "session_id": "stale-session",
+                "surface": "cli",
+                "pid": os.getpid(),
+                "process_start_time": 100.0,
+                "started_at": 1,
+                "updated_at": 1,
+            }
+        ],
+    )
+
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="new-session",
+        surface="cli",
+        config={"max_concurrent_sessions": 1},
+    )
+
+    assert message is None
+    assert lease is not None
+    assert [entry["session_id"] for entry in active_sessions.active_session_registry_snapshot()] == [
+        "new-session"
+    ]
+    lease.release()

tests/hermes_cli/test_cli_active_session_limit.py

@@ -0,0 +1,41 @@
+from cli import HermesCLI
+from hermes_cli.active_sessions import (
+    active_session_registry_snapshot,
+    try_acquire_active_session,
+)
+
+
+def test_cli_claim_active_session_respects_global_limit(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    cfg = {"max_concurrent_sessions": 1}
+    held, message = try_acquire_active_session(
+        session_id="held-session",
+        surface="tui",
+        config=cfg,
+    )
+    assert message is None
+    assert held is not None
+
+    cli = object.__new__(HermesCLI)
+    cli.session_id = "new-cli-session"
+    cli.config = cfg
+    cli._active_session_lease = None
+    printed: list[str] = []
+    cli._console_print = lambda text: printed.append(text)
+
+    try:
+        assert cli._claim_active_session("cli") is False
+        assert printed == [
+            "[bold red]Hermes is at the active session limit (1/1). "
+            "Try again when another session finishes.[/]"
+        ]
+
+        held.release()
+
+        assert cli._claim_active_session("cli") is True
+        assert [entry["session_id"] for entry in active_session_registry_snapshot()] == [
+            "new-cli-session"
+        ]
+    finally:
+        held.release()
+        cli._release_active_session()

tests/hermes_cli/test_plugins_cmd.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import logging
+import os
+import shutil
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -16,6 +18,7 @@ from hermes_cli.plugins_cmd import (
     _repo_name_from_url,
     _resolve_git_executable,
     _resolve_git_url,
+    _resolve_subdir_within,
     _sanitize_plugin_name,
 )
 
@@ -97,35 +100,127 @@ class TestSanitizePluginName:
 
 
 class TestResolveGitUrl:
-    """Shorthand and full-URL resolution."""
+    """Shorthand and full-URL resolution, with optional subdirectory."""
 
     def test_owner_repo_shorthand(self):
-        url = _resolve_git_url("owner/repo")
+        url, subdir = _resolve_git_url("owner/repo")
         assert url == "https://github.com/owner/repo.git"
+        assert subdir is None
 
     def test_https_url_passthrough(self):
-        url = _resolve_git_url("https://github.com/x/y.git")
+        url, subdir = _resolve_git_url("https://github.com/x/y.git")
         assert url == "https://github.com/x/y.git"
+        assert subdir is None
 
     def test_ssh_url_passthrough(self):
-        url = _resolve_git_url("git@github.com:x/y.git")
+        url, subdir = _resolve_git_url("git@github.com:x/y.git")
         assert url == "git@github.com:x/y.git"
+        assert subdir is None
 
     def test_http_url_passthrough(self):
-        url = _resolve_git_url("http://example.com/repo.git")
+        url, subdir = _resolve_git_url("http://example.com/repo.git")
         assert url == "http://example.com/repo.git"
+        assert subdir is None
 
     def test_file_url_passthrough(self):
-        url = _resolve_git_url("file:///tmp/repo")
+        url, subdir = _resolve_git_url("file:///tmp/repo")
         assert url == "file:///tmp/repo"
+        assert subdir is None
 
     def test_invalid_single_word_raises(self):
         with pytest.raises(ValueError, match="Invalid plugin identifier"):
             _resolve_git_url("justoneword")
 
-    def test_invalid_three_parts_raises(self):
-        with pytest.raises(ValueError, match="Invalid plugin identifier"):
-            _resolve_git_url("a/b/c")
+    def test_shorthand_with_subdir(self):
+        url, subdir = _resolve_git_url("owner/repo/my-plugin")
+        assert url == "https://github.com/owner/repo.git"
+        assert subdir == "my-plugin"
+
+    def test_shorthand_with_nested_subdir(self):
+        url, subdir = _resolve_git_url("owner/repo/path/to/plugin")
+        assert url == "https://github.com/owner/repo.git"
+        assert subdir == "path/to/plugin"
+
+    def test_shorthand_with_subdir_trailing_slash(self):
+        url, subdir = _resolve_git_url("owner/repo/my-plugin/")
+        assert url == "https://github.com/owner/repo.git"
+        assert subdir == "my-plugin"
+
+    def test_https_url_with_subdir(self):
+        url, subdir = _resolve_git_url("https://github.com/owner/repo.git/my-plugin")
+        assert url == "https://github.com/owner/repo.git"
+        assert subdir == "my-plugin"
+
+    def test_https_url_with_nested_subdir(self):
+        url, subdir = _resolve_git_url(
+            "https://github.com/owner/repo.git/path/to/plugin"
+        )
+        assert url == "https://github.com/owner/repo.git"
+        assert subdir == "path/to/plugin"
+
+    def test_url_with_fragment_subdir(self):
+        url, subdir = _resolve_git_url("https://github.com/owner/repo.git#my-plugin")
+        assert url == "https://github.com/owner/repo.git"
+        assert subdir == "my-plugin"
+
+    def test_file_url_with_fragment_subdir(self):
+        url, subdir = _resolve_git_url("file:///tmp/repo#path/to/plugin")
+        assert url == "file:///tmp/repo"
+        assert subdir == "path/to/plugin"
+
+    def test_ssh_url_with_fragment_subdir(self):
+        url, subdir = _resolve_git_url("git@github.com:owner/repo.git#sub")
+        assert url == "git@github.com:owner/repo.git"
+        assert subdir == "sub"
+
+
+# ── _resolve_subdir_within ──────────────────────────────────────────────────
+
+
+class TestResolveSubdirWithin:
+    """Subdirectory resolution stays within the clone and rejects traversal."""
+
+    def test_valid_subdir(self, tmp_path):
+        (tmp_path / "my-plugin").mkdir()
+        result = _resolve_subdir_within(tmp_path, "my-plugin")
+        assert result == (tmp_path / "my-plugin").resolve()
+
+    def test_valid_nested_subdir(self, tmp_path):
+        (tmp_path / "a" / "b" / "c").mkdir(parents=True)
+        result = _resolve_subdir_within(tmp_path, "a/b/c")
+        assert result == (tmp_path / "a" / "b" / "c").resolve()
+
+    def test_rejects_dot_dot_escape(self, tmp_path):
+        clone = tmp_path / "clone"
+        clone.mkdir()
+        (tmp_path / "secret").mkdir()
+        with pytest.raises(PluginOperationError, match="escapes the repository"):
+            _resolve_subdir_within(clone, "../secret")
+
+    def test_rejects_absolute_path_escape(self, tmp_path):
+        clone = tmp_path / "clone"
+        clone.mkdir()
+        # An absolute path resolves outside the clone root.
+        with pytest.raises(PluginOperationError, match="escapes the repository"):
+            _resolve_subdir_within(clone, "/etc")
+
+    def test_rejects_symlink_escape(self, tmp_path):
+        clone = tmp_path / "clone"
+        clone.mkdir()
+        outside = tmp_path / "outside"
+        outside.mkdir()
+        (clone / "link").symlink_to(outside)
+        with pytest.raises(PluginOperationError, match="escapes the repository"):
+            _resolve_subdir_within(clone, "link")
+
+    def test_rejects_missing_subdir(self, tmp_path):
+        with pytest.raises(PluginOperationError, match="does not exist"):
+            _resolve_subdir_within(tmp_path, "nope")
+
+    def test_rejects_file_not_dir(self, tmp_path):
+        (tmp_path / "afile").write_text("x")
+        with pytest.raises(PluginOperationError, match="not a directory"):
+            _resolve_subdir_within(tmp_path, "afile")
 
 
 # ── _resolve_git_executable ─────────────────────────────────────────────────
@@ -698,3 +793,90 @@ class TestNoAutoActivation:
         # The old code had: "Even with default config, check if a plugin registered one"
         # The fix removes this. Verify it's gone.
         assert "Even with default config, check if a plugin registered one" not in source
+
+
+# ── End-to-end subdirectory install ──────────────────────────────────────────
+
+
+class TestSubdirInstallE2E:
+    """Install a plugin that lives in a subdirectory of a real local git repo."""
+
+    @staticmethod
+    def _make_repo_with_subdir_plugin(repo_root: Path) -> None:
+        """Create a git repo where the plugin lives in ``./my-plugin/`` and the
+        repo root holds unrelated docs/tests."""
+        import subprocess as sp
+
+        repo_root.mkdir(parents=True, exist_ok=True)
+        # Root-level noise: docs + tests that should NOT be installed.
+        (repo_root / "README.md").write_text("# Monorepo docs\n")
+        (repo_root / "tests").mkdir()
+        (repo_root / "tests" / "test_x.py").write_text("def test_x():\n    pass\n")
+        # The actual plugin in a subdirectory.
+        plugin_dir = repo_root / "my-plugin"
+        plugin_dir.mkdir()
+        (plugin_dir / "plugin.yaml").write_text(
+            "name: my-plugin\nmanifest_version: 1\ndescription: A subdir plugin\n"
+        )
+        (plugin_dir / "__init__.py").write_text("# plugin entry\n")
+
+        env = {
+            **os.environ,
+            "GIT_AUTHOR_NAME": "t",
+            "GIT_AUTHOR_EMAIL": "t@t",
+            "GIT_COMMITTER_NAME": "t",
+            "GIT_COMMITTER_EMAIL": "t@t",
+        }
+        sp.run(["git", "init", "-q"], cwd=repo_root, check=True, env=env)
+        sp.run(["git", "add", "-A"], cwd=repo_root, check=True, env=env)
+        sp.run(
+            ["git", "commit", "-q", "-m", "init"],
+            cwd=repo_root,
+            check=True,
+            env=env,
+        )
+
+    def test_installs_only_the_subdir_plugin(self, tmp_path, monkeypatch):
+        if shutil.which("git") is None:
+            pytest.skip("git not available")
+
+        from hermes_cli import plugins_cmd as pc
+
+        repo_root = tmp_path / "monorepo"
+        self._make_repo_with_subdir_plugin(repo_root)
+
+        plugins_dir = tmp_path / "installed"
+        plugins_dir.mkdir()
+        monkeypatch.setattr(pc, "_plugins_dir", lambda: plugins_dir)
+
+        identifier = f"file://{repo_root}#my-plugin"
+        target, manifest, name = pc._install_plugin_core(identifier, force=False)
+
+        # Installed under the plugin's own name, not the repo name.
+        assert name == "my-plugin"
+        assert manifest.get("name") == "my-plugin"
+        assert target == (plugins_dir / "my-plugin").resolve()
+
+        # The plugin's files are present...
+        assert (target / "plugin.yaml").exists()
+        assert (target / "__init__.py").exists()
+        # ...and the repo-root noise is NOT.
+        assert not (target / "README.md").exists()
+        assert not (target / "tests").exists()
+
+    def test_missing_subdir_raises(self, tmp_path, monkeypatch):
+        if shutil.which("git") is None:
+            pytest.skip("git not available")
+
+        from hermes_cli import plugins_cmd as pc
+
+        repo_root = tmp_path / "monorepo"
+        self._make_repo_with_subdir_plugin(repo_root)
+
+        plugins_dir = tmp_path / "installed"
+        plugins_dir.mkdir()
+        monkeypatch.setattr(pc, "_plugins_dir", lambda: plugins_dir)
+
+        identifier = f"file://{repo_root}#does-not-exist"
+        with pytest.raises(PluginOperationError, match="does not exist"):
+            pc._install_plugin_core(identifier, force=False)

tests/plugins/platforms/photon/test_outbound_media.py

@@ -0,0 +1,255 @@
+"""Outbound-media tests for PhotonAdapter.
+
+Photon ships outbound attachments via spectrum-ts' ``attachment()`` /
+``voice()`` content builders, reached through the Node sidecar's
+``/send-attachment`` endpoint. These tests stub ``_sidecar_call`` so we
+can assert the endpoint + body shape each ``send_*`` override produces
+without spawning Node or binding ports.
+"""
+from __future__ import annotations
+
+import os
+from typing import Any, Dict, List, Tuple
+
+import pytest
+
+from gateway.config import PlatformConfig
+from plugins.platforms.photon import adapter as photon_adapter
+from plugins.platforms.photon.adapter import PhotonAdapter
+
+
+def _make_adapter(monkeypatch: pytest.MonkeyPatch) -> PhotonAdapter:
+    monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id")
+    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret")
+    monkeypatch.delenv("PHOTON_WEBHOOK_SECRET", raising=False)
+    cfg = PlatformConfig(enabled=True, token="", extra={})
+    return PhotonAdapter(cfg)
+
+
+def _capture_sidecar(adapter: PhotonAdapter) -> List[Tuple[str, Dict[str, Any]]]:
+    """Replace ``_sidecar_call`` with a recorder that returns a fixed id."""
+    calls: List[Tuple[str, Dict[str, Any]]] = []
+
+    async def _fake_call(path: str, body: Dict[str, Any]) -> Dict[str, Any]:
+        calls.append((path, body))
+        return {"ok": True, "messageId": "msg-123"}
+
+    adapter._sidecar_call = _fake_call  # type: ignore[assignment]
+    return calls
+
+
+@pytest.fixture()
+def real_file(tmp_path) -> str:
+    p = tmp_path / "photo.jpg"
+    p.write_bytes(b"\xff\xd8\xff\xe0fake-jpeg")
+    return str(p)
+
+
+def _patch_safe_path(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Make path validation a passthrough so tmp files outside the cache pass."""
+    monkeypatch.setattr(
+        PhotonAdapter,
+        "validate_media_delivery_path",
+        staticmethod(lambda p: p if os.path.exists(p) else None),
+    )
+
+
+@pytest.mark.asyncio
+async def test_send_image_file_hits_attachment_endpoint(
+    monkeypatch: pytest.MonkeyPatch, real_file: str
+) -> None:
+    _patch_safe_path(monkeypatch)
+    adapter = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(adapter)
+
+    result = await adapter.send_image_file(
+        "any;-;+15551234567", real_file, caption="look"
+    )
+
+    assert result.success is True
+    assert result.message_id == "msg-123"
+    assert len(calls) == 1
+    path, body = calls[0]
+    assert path == "/send-attachment"
+    assert body["spaceId"] == "any;-;+15551234567"
+    assert body["path"] == real_file
+    assert body["kind"] == "attachment"
+    assert body["caption"] == "look"
+    assert body["mimeType"] == "image/jpeg"  # inferred from .jpg
+
+
+@pytest.mark.asyncio
+async def test_send_voice_marks_kind_voice(
+    monkeypatch: pytest.MonkeyPatch, tmp_path
+) -> None:
+    _patch_safe_path(monkeypatch)
+    audio = tmp_path / "note.m4a"
+    audio.write_bytes(b"fake-audio")
+    adapter = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(adapter)
+
+    result = await adapter.send_voice("any;-;+1", str(audio))
+
+    assert result.success is True
+    path, body = calls[0]
+    assert path == "/send-attachment"
+    assert body["kind"] == "voice"
+
+
+@pytest.mark.asyncio
+async def test_send_document_passes_filename(
+    monkeypatch: pytest.MonkeyPatch, tmp_path
+) -> None:
+    _patch_safe_path(monkeypatch)
+    doc = tmp_path / "report.pdf"
+    doc.write_bytes(b"%PDF-1.4 fake")
+    adapter = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(adapter)
+
+    await adapter.send_document("any;-;+1", str(doc), file_name="Q3.pdf")
+
+    _, body = calls[0]
+    assert body["kind"] == "attachment"
+    assert body["name"] == "Q3.pdf"
+    assert body["mimeType"] == "application/pdf"
+
+
+@pytest.mark.asyncio
+async def test_send_video_passes_through(
+    monkeypatch: pytest.MonkeyPatch, tmp_path
+) -> None:
+    _patch_safe_path(monkeypatch)
+    vid = tmp_path / "clip.mp4"
+    vid.write_bytes(b"fake-mp4")
+    adapter = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(adapter)
+
+    await adapter.send_video("any;+;groupguid", str(vid), caption="watch")
+
+    _, body = calls[0]
+    assert body["kind"] == "attachment"
+    assert body["caption"] == "watch"
+
+
+@pytest.mark.asyncio
+async def test_send_image_url_caches_then_sends_attachment(
+    monkeypatch: pytest.MonkeyPatch, real_file: str
+) -> None:
+    _patch_safe_path(monkeypatch)
+    adapter = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(adapter)
+
+    async def _fake_cache(url: str, *a, **k) -> str:
+        assert url == "https://example.com/cat.jpg"
+        return real_file
+
+    import gateway.platforms.base as base_mod
+
+    monkeypatch.setattr(base_mod, "cache_image_from_url", _fake_cache)
+
+    result = await adapter.send_image(
+        "any;-;+1", "https://example.com/cat.jpg", caption="cat"
+    )
+
+    assert result.success is True
+    path, body = calls[0]
+    assert path == "/send-attachment"
+    assert body["path"] == real_file
+    assert body["caption"] == "cat"
+
+
+@pytest.mark.asyncio
+async def test_send_image_url_fetch_failure_falls_back_to_text(
+    monkeypatch: pytest.MonkeyPatch
+) -> None:
+    adapter = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(adapter)
+
+    async def _boom(url: str, *a, **k) -> str:
+        raise RuntimeError("network down")
+
+    import gateway.platforms.base as base_mod
+
+    monkeypatch.setattr(base_mod, "cache_image_from_url", _boom)
+
+    result = await adapter.send_image(
+        "any;-;+1", "https://example.com/cat.jpg", caption="cat"
+    )
+
+    # Fallback path: base send_image() routes to send() → /send (text).
+    assert result.success is True
+    assert calls[0][0] == "/send"
+    assert "https://example.com/cat.jpg" in calls[0][1]["text"]
+
+
+@pytest.mark.asyncio
+async def test_send_attachment_rejects_unsafe_path(
+    monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Default validation (no passthrough patch) should reject a nonexistent /
+    # traversal path, returning a failed SendResult without calling the sidecar.
+    monkeypatch.setattr(
+        PhotonAdapter,
+        "validate_media_delivery_path",
+        staticmethod(lambda p: None),
+    )
+    adapter = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(adapter)
+
+    result = await adapter.send_image_file("any;-;+1", "/etc/passwd")
+
+    assert result.success is False
+    assert "unsafe" in (result.error or "")
+    assert calls == []  # never reached the sidecar
+
+
+@pytest.mark.asyncio
+async def test_standalone_send_text_then_attachments(
+    monkeypatch: pytest.MonkeyPatch, tmp_path
+) -> None:
+    _patch_safe_path(monkeypatch)
+    img = tmp_path / "a.png"
+    img.write_bytes(b"\x89PNG fake")
+    monkeypatch.setenv("PHOTON_SIDECAR_TOKEN", "tok")
+
+    posted: List[Tuple[str, Dict[str, Any]]] = []
+
+    class _Resp:
+        status_code = 200
+
+        @staticmethod
+        def json() -> Dict[str, Any]:
+            return {"ok": True, "messageId": "m-9"}
+
+    class _FakeClient:
+        def __init__(self, *a, **k):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+        async def post(self, url: str, json: Dict[str, Any], headers=None):
+            posted.append((url, json))
+            return _Resp()
+
+    monkeypatch.setattr(photon_adapter.httpx, "AsyncClient", _FakeClient)
+
+    cfg = PlatformConfig(enabled=True, token="", extra={})
+    result = await photon_adapter._standalone_send(
+        cfg,
+        "any;-;+1",
+        "hello",
+        media_files=[(str(img), False)],
+    )
+
+    assert result.get("success") is True
+    # First call is the text /send, second is /send-attachment.
+    assert posted[0][0].endswith("/send")
+    assert posted[0][1]["text"] == "hello"
+    assert posted[1][0].endswith("/send-attachment")
+    assert posted[1][1]["path"] == str(img)
+    assert posted[1][1]["kind"] == "attachment"
+    assert posted[1][1]["mimeType"] == "image/png"

tests/plugins/test_langfuse_plugin.py

@@ -704,3 +704,76 @@ class TestToolObservationKeying:
         assert ended["output"] == {"status": "done"}
         assert not state.tools
 
+
+class TestUsageFromSanitizedResponse:
+    """Regression: ``post_api_request`` delivers ``response`` as a sanitized
+    dict (no ``.usage`` attribute) plus a separate ``usage`` summary dict. The
+    post-call handler must read the ``usage`` dict instead of treating the dict
+    response as a usage-bearing object and dropping all token/cost data."""
+
+    def _setup(self, mod, monkeypatch):
+        # Active client so on_post_llm_call does not early-return.
+        monkeypatch.setattr(mod, "_get_langfuse", lambda: object())
+        observation = object()
+        state = mod.TraceState(trace_id="trace-1", root_ctx=None, root_span=None)
+        state.generations[mod._request_key(1)] = observation
+        monkeypatch.setitem(mod._TRACE_STATE, mod._trace_key("task-1", "session-1"), state)
+        captured = {}
+
+        def fake_end_observation(obs, *, output=None, metadata=None, usage_details=None, cost_details=None):
+            captured["usage_details"] = usage_details
+
+        monkeypatch.setattr(mod, "_end_observation", fake_end_observation)
+        return captured
+
+    def test_sanitized_dict_response_uses_usage_dict(self, monkeypatch):
+        sys.modules.pop("plugins.observability.langfuse", None)
+        mod = importlib.import_module("plugins.observability.langfuse")
+        captured = self._setup(mod, monkeypatch)
+
+        # A plain dict has no ``.usage`` attribute — mirrors post_api_request.
+        mod.on_post_llm_call(
+            task_id="task-1",
+            session_id="session-1",
+            api_call_count=1,
+            model="gemini-3-flash-preview",
+            response={"model": "gemini-3-flash-preview", "usage": {"input_tokens": 100, "output_tokens": 20}},
+            usage={"input_tokens": 100, "output_tokens": 20},
+            assistant_content_chars=42,
+        )
+
+        # Before the fix the dict response shadowed the usage dict and tokens
+        # were lost (usage_details == {}).
+        assert captured["usage_details"] == {"input": 100, "output": 20}
+
+    def test_real_response_object_with_usage_still_used(self, monkeypatch):
+        sys.modules.pop("plugins.observability.langfuse", None)
+        mod = importlib.import_module("plugins.observability.langfuse")
+        captured = self._setup(mod, monkeypatch)
+
+        # A response object that genuinely carries usage must still take the
+        # response-object path (post_llm_call / legacy behavior).
+        seen = {}
+
+        def fake_usage_and_cost(resp, **_):
+            seen["resp"] = resp
+            return {"input": 7, "output": 3}, {}
+
+        monkeypatch.setattr(mod, "_usage_and_cost", fake_usage_and_cost)
+
+        class _Resp:
+            usage = {"prompt_tokens": 7, "completion_tokens": 3}
+
+        resp = _Resp()
+        mod.on_post_llm_call(
+            task_id="task-1",
+            session_id="session-1",
+            api_call_count=1,
+            model="gemini-3-flash-preview",
+            response=resp,
+            usage={"input_tokens": 999, "output_tokens": 999},
+            assistant_content_chars=42,
+        )
+
+        assert seen["resp"] is resp
+        assert captured["usage_details"] == {"input": 7, "output": 3}

tests/plugins/test_nemo_relay_plugin.py

@@ -713,8 +713,8 @@ version = 1
 kind = "adaptive"
 enabled = true
 
-[components.config]
-mode = "route"
+[components.config.tool_parallelism]
+mode = "observe_only"
 """,
         encoding="utf-8",
     )
@@ -762,7 +762,7 @@ mode = "route"
     assert response.choices == [raw_choice]
     assert seen_request["intercepted"] is True
     execute_start = next(event for event in fake.events if event[0] == "llm.execute.start")
-    assert execute_start[3]["data"]["mode"] == "route"
+    assert execute_start[3]["data"]["mode"] == "observe_only"
     execute_end = next(event for event in fake.events if event[0] == "llm.execute.end")
     assert execute_end[2] == {
         "model": "demo-model",
@@ -783,6 +783,84 @@ mode = "route"
     }
 
 
+def _adaptive_llm_execute_mode(tmp_path, monkeypatch, plugins_toml_text: str) -> str:
+    fake = _FakeNemoRelay()
+    plugin = _fresh_plugin(monkeypatch, fake)
+    plugins_toml = tmp_path / "plugins.toml"
+    plugins_toml.write_text(plugins_toml_text, encoding="utf-8")
+    monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml))
+
+    plugin.on_llm_execution_middleware(
+        session_id="s1",
+        provider="anthropic",
+        model="demo-model",
+        request={"messages": [{"role": "user", "content": "hi"}]},
+        next_call=lambda request: {"raw": request},
+    )
+
+    execute_start = next(event for event in fake.events if event[0] == "llm.execute.start")
+    return execute_start[3]["data"]["mode"]
+
+
+def test_nemo_relay_adaptive_llm_execution_middleware_defaults_to_observe_only_when_mode_is_unset(
+    tmp_path, monkeypatch
+):
+    mode = _adaptive_llm_execute_mode(
+        tmp_path,
+        monkeypatch,
+        """
+version = 1
+
+[[components]]
+kind = "adaptive"
+enabled = true
+
+[components.config]
+version = 1
+""",
+    )
+    assert mode == "observe_only"
+
+
+def test_nemo_relay_adaptive_llm_execution_middleware_accepts_legacy_top_level_mode(tmp_path, monkeypatch):
+    mode = _adaptive_llm_execute_mode(
+        tmp_path,
+        monkeypatch,
+        """
+version = 1
+
+[[components]]
+kind = "adaptive"
+enabled = true
+
+[components.config]
+mode = "route"
+""",
+    )
+    assert mode == "route"
+
+
+def test_nemo_relay_adaptive_llm_execution_middleware_prefers_tool_parallelism_mode(tmp_path, monkeypatch):
+    mode = _adaptive_llm_execute_mode(
+        tmp_path,
+        monkeypatch,
+        """
+version = 1
+
+[[components]]
+kind = "adaptive"
+enabled = true
+
+[components.config]
+mode = "route"
+
+[components.config.tool_parallelism]
+mode = "schedule"
+""",
+    )
+    assert mode == "schedule"
+
+
 def test_nemo_relay_llm_execution_middleware_calls_through_without_adaptive(monkeypatch):
     fake = _FakeNemoRelay()
     plugin = _fresh_plugin(monkeypatch, fake)
@@ -811,8 +889,8 @@ version = 1
 kind = "adaptive"
 enabled = true
 
-[components.config]
-mode = "route"
+[components.config.tool_parallelism]
+mode = "observe_only"
 """,
         encoding="utf-8",
     )
@@ -838,7 +916,7 @@ mode = "route"
     assert response == {"raw": True, "args": {"command": "pwd", "intercepted": True}}
     assert seen_args["intercepted"] is True
     execute_start = next(event for event in fake.events if event[0] == "tool.execute.start")
-    assert execute_start[3]["data"]["mode"] == "route"
+    assert execute_start[3]["data"]["mode"] == "observe_only"
     assert execute_start[3]["data"]["tool_call_id"] == "tool-1"
 
 
@@ -869,8 +947,8 @@ version = 1
 kind = "adaptive"
 enabled = true
 
-[components.config]
-mode = "route"
+[components.config.tool_parallelism]
+mode = "observe_only"
 """,
         encoding="utf-8",
     )

tests/scripts/test_build_skills_index_health.py

@@ -0,0 +1,99 @@
+"""Invariants for scripts/build_skills_index.py's health-check guard.
+
+Regression context (June 2026): a GitHub API rate limit zeroed every
+api.github.com-backed source (github / claude-marketplace / well-known) at
+once during the docs deploy crawl. The build's health check fired and exited
+non-zero — but it had ALREADY written the degenerate index to disk, and
+deploy-site.yml swallowed the exit code with ``|| echo non-fatal``. The
+partial index (missing the OpenAI/Anthropic/HuggingFace/NVIDIA tabs) shipped
+to the live Skills Hub.
+
+These tests pin the two contracts that prevent a recurrence:
+  1. A degenerate crawl exits non-zero AND does NOT write the output file
+     (so extract-skills.py falls back instead of reading a broken index).
+  2. A healthy crawl exits zero AND writes the file with every source present.
+"""
+
+import os
+import sys
+import types
+
+import pytest
+
+import scripts.build_skills_index as build_mod
+
+
+def _meta(name, src):
+    return build_mod.SkillMeta(
+        name=name, description="d", source=src,
+        identifier=f"{src}/{name}", trust_level="community",
+    )
+
+
+class _FakeSource:
+    def __init__(self, src, n, rate_limited=False):
+        self._src = src
+        self._n = n
+        self.is_rate_limited = rate_limited
+
+    def search(self, query, limit=10):
+        return [_meta(f"{self._src}-{i}", self._src) for i in range(self._n)]
+
+
+def _install_fake_sources(monkeypatch, *, github_count, claude_count=40,
+                          well_known_count=10, github_rate_limited=False):
+    monkeypatch.setattr(build_mod, "SkillsShSource", lambda auth: _FakeSource("skills.sh", 15000))
+    monkeypatch.setattr(build_mod, "OptionalSkillSource", lambda: _FakeSource("official", 95))
+    monkeypatch.setattr(build_mod, "WellKnownSkillSource", lambda: _FakeSource("well-known", well_known_count))
+    monkeypatch.setattr(
+        build_mod, "GitHubSource",
+        lambda auth: _FakeSource("github", github_count, rate_limited=github_rate_limited),
+    )
+    monkeypatch.setattr(build_mod, "ClawHubSource", lambda: _FakeSource("clawhub", 69000))
+    monkeypatch.setattr(
+        build_mod, "ClaudeMarketplaceSource",
+        lambda auth: _FakeSource("claude-marketplace", claude_count, rate_limited=github_rate_limited),
+    )
+    monkeypatch.setattr(build_mod, "LobeHubSource", lambda: _FakeSource("lobehub", 500))
+    monkeypatch.setattr(build_mod, "BrowseShSource", lambda: _FakeSource("browse-sh", 380))
+    monkeypatch.setattr(
+        build_mod, "crawl_skills_sh",
+        lambda source: [build_mod._meta_to_dict(m) for m in source.search("", 0)],
+    )
+    monkeypatch.setattr(build_mod, "batch_resolve_paths", lambda skills, auth: skills)
+    monkeypatch.setattr(
+        build_mod, "GitHubAuth",
+        lambda: types.SimpleNamespace(auth_method=lambda: "token"),
+    )
+
+
+def test_degenerate_crawl_exits_nonzero_and_writes_no_file(tmp_path, monkeypatch):
+    """A collapsed GitHub crawl must fail loud and leave OUTPUT_PATH unwritten."""
+    out = tmp_path / "skills-index.json"
+    monkeypatch.setattr(build_mod, "OUTPUT_PATH", str(out))
+    _install_fake_sources(monkeypatch, github_count=0, claude_count=0,
+                          well_known_count=0, github_rate_limited=True)
+
+    with pytest.raises(SystemExit) as exc:
+        build_mod.main()
+
+    assert exc.value.code != 0
+    # The degenerate index must NOT have been written — extract-skills.py
+    # relies on the file's absence to fall back instead of reading garbage.
+    assert not out.exists()
+
+
+def test_healthy_crawl_writes_index_with_all_sources(tmp_path, monkeypatch):
+    out = tmp_path / "skills-index.json"
+    monkeypatch.setattr(build_mod, "OUTPUT_PATH", str(out))
+    _install_fake_sources(monkeypatch, github_count=200)
+
+    build_mod.main()  # exit 0 (no SystemExit)
+
+    assert out.exists()
+    import json
+    data = json.loads(out.read_text())
+    sources = {s["source"] for s in data["skills"]}
+    # Every GitHub-API-backed source that vanished in the regression is present.
+    assert {"github", "claude-marketplace", "well-known"} <= sources
+    assert data["skill_count"] == len(data["skills"])

tests/test_tui_gateway_server.py

@@ -9,9 +9,55 @@ from datetime import datetime
 from pathlib import Path
 from unittest.mock import patch
 
+from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+from hermes_cli.active_sessions import active_session_registry_snapshot
 from tui_gateway import server
 
 
+def test_session_create_rejects_at_active_session_limit(monkeypatch, tmp_path):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    (home / "config.yaml").write_text("max_concurrent_sessions: 1\n", encoding="utf-8")
+    token = set_hermes_home_override(home)
+
+    def _clear_server_sessions():
+        for session in list(server._sessions.values()):
+            server._teardown_session(session)
+        server._sessions.clear()
+
+    try:
+        server._cfg_cache = None
+        server._cfg_mtime = None
+        server._cfg_path = None
+        _clear_server_sessions()
+        monkeypatch.setattr(server, "_start_agent_build", lambda *args, **kwargs: None)
+        monkeypatch.setattr(server, "_completion_cwd", lambda params=None: str(tmp_path))
+
+        first = server._methods["session.create"]("r1", {"cols": 80})
+        assert "result" in first
+        sid = first["result"]["session_id"]
+
+        second = server._methods["session.create"]("r2", {"cols": 80})
+        assert second["error"]["message"] == (
+            "Hermes is at the active session limit (1/1). "
+            "Try again when another session finishes."
+        )
+        assert list(server._sessions) == [sid]
+
+        closed = server._methods["session.close"]("r3", {"session_id": sid})
+        assert closed["result"]["closed"] is True
+        assert active_session_registry_snapshot() == []
+
+        third = server._methods["session.create"]("r4", {"cols": 80})
+        assert "result" in third
+    finally:
+        _clear_server_sessions()
+        server._cfg_cache = None
+        server._cfg_mtime = None
+        server._cfg_path = None
+        reset_hermes_home_override(token)
+
+
 def test_session_context_uses_session_cwd(monkeypatch, tmp_path):
     """Desktop/TUI sessions must pin the agent cwd per session.
 

tools/skills_hub.py

@@ -550,11 +550,8 @@ class GitHubSource(SkillSource):
             return [SkillMeta(**s) for s in cached]
 
         url = f"https://api.github.com/repos/{repo}/contents/{path.rstrip('/')}"
-        try:
-            resp = httpx.get(url, headers=self.auth.get_headers(), timeout=15, follow_redirects=True)
-            if resp.status_code != 200:
-                return []
-        except httpx.HTTPError:
+        resp = self._github_get(url)
+        if resp is None or resp.status_code != 200:
             return []
 
         entries = resp.json()
@@ -639,15 +636,98 @@ class GitHubSource(SkillSource):
 
     def _check_rate_limit_response(self, resp: "httpx.Response") -> None:
         """Flag the instance as rate-limited when GitHub returns 403 + exhausted quota."""
-        if resp.status_code == 403:
+        if resp.status_code in (403, 429):
             remaining = resp.headers.get("X-RateLimit-Remaining", "")
-            if remaining == "0":
+            if remaining == "0" or resp.status_code == 429:
                 self._rate_limited = True
                 logger.warning(
                     "GitHub API rate limit exhausted (unauthenticated: 60 req/hr). "
                     "Set GITHUB_TOKEN or install the gh CLI to raise the limit to 5,000/hr."
                 )
 
+    def _github_get(
+        self,
+        url: str,
+        *,
+        params: Optional[Dict] = None,
+        headers: Optional[Dict] = None,
+        timeout: float = 15.0,
+        max_retries: int = 3,
+    ) -> Optional["httpx.Response"]:
+        """GET against the GitHub API with retry/backoff on transient failures.
+
+        Returns the final ``httpx.Response`` (caller inspects status) or
+        ``None`` when every attempt raised a transport error.
+
+        Retries on:
+          - 403/429 with ``X-RateLimit-Remaining: 0`` — waits until the
+            reset time (capped) when the header is present, else exponential
+            backoff. This is the all-GitHub-tap-collapse case: a single
+            shared rate limit zeroes github + claude-marketplace + well-known
+            at once during the index build.
+          - 5xx and connection/timeout errors — exponential backoff.
+
+        On terminal rate-limit exhaustion the instance is flagged via
+        ``_check_rate_limit_response`` so the build can fail loud instead of
+        silently shipping an index with the GitHub sources dropped to zero.
+        """
+        hdrs = headers if headers is not None else self.auth.get_headers()
+        backoff = 1.0
+        last_resp: Optional["httpx.Response"] = None
+        for attempt in range(max_retries):
+            try:
+                resp = httpx.get(
+                    url, params=params, headers=hdrs,
+                    timeout=timeout, follow_redirects=True,
+                )
+            except httpx.HTTPError as e:
+                logger.debug("GitHub GET %s failed (attempt %d/%d): %s",
+                             url, attempt + 1, max_retries, e)
+                if attempt < max_retries - 1:
+                    time.sleep(backoff)
+                    backoff = min(backoff * 2, 30.0)
+                    continue
+                return None
+
+            last_resp = resp
+            if resp.status_code == 200:
+                return resp
+
+            # Rate-limited: honor the reset header when present, else back off.
+            if resp.status_code in (403, 429):
+                remaining = resp.headers.get("X-RateLimit-Remaining", "")
+                is_rl = remaining == "0" or resp.status_code == 429
+                if is_rl and attempt < max_retries - 1:
+                    wait = backoff
+                    reset = resp.headers.get("X-RateLimit-Reset", "")
+                    retry_after = resp.headers.get("Retry-After", "")
+                    if retry_after.isdigit():
+                        wait = min(float(retry_after), 60.0)
+                    elif reset.isdigit():
+                        delta = float(reset) - time.time()
+                        if 0 < delta <= 60.0:
+                            wait = delta
+                    logger.debug(
+                        "GitHub rate limited on %s, waiting %.1fs (attempt %d/%d)",
+                        url, wait, attempt + 1, max_retries,
+                    )
+                    time.sleep(wait)
+                    backoff = min(backoff * 2, 30.0)
+                    continue
+                # Out of retries (or not a rate-limit 403) — flag and return.
+                self._check_rate_limit_response(resp)
+                return resp
+
+            # 5xx — retry; 4xx (other than rate limit) — return immediately.
+            if 500 <= resp.status_code < 600 and attempt < max_retries - 1:
+                time.sleep(backoff)
+                backoff = min(backoff * 2, 30.0)
+                continue
+            return resp
+
+        return last_resp
+
+
     def _download_directory(self, repo: str, path: str) -> Dict[str, str]:
         """Recursively download all text files from a GitHub directory.
 
@@ -768,17 +848,12 @@ class GitHubSource(SkillSource):
     def _fetch_file_content(self, repo: str, path: str) -> Optional[str]:
         """Fetch a single file's content from GitHub."""
         url = f"https://api.github.com/repos/{repo}/contents/{path}"
-        try:
-            resp = httpx.get(
-                url,
-                headers={**self.auth.get_headers(), "Accept": "application/vnd.github.v3.raw"},
-                timeout=15, follow_redirects=True,
-            )
-            if resp.status_code == 200:
-                return resp.text
-            self._check_rate_limit_response(resp)
-        except httpx.HTTPError as e:
-            logger.debug("GitHub contents API fetch failed: %s", e)
+        resp = self._github_get(
+            url,
+            headers={**self.auth.get_headers(), "Accept": "application/vnd.github.v3.raw"},
+        )
+        if resp is not None and resp.status_code == 200:
+            return resp.text
         return None
 
     def _get_skillsh_groupings(self, repo: str) -> Optional[Dict[str, str]]:
@@ -2373,10 +2448,19 @@ class ClaudeMarketplaceSource(SkillSource):
 
     def __init__(self, auth: GitHubAuth):
         self.auth = auth
+        # Persistent GitHubSource so rate-limit state survives across the
+        # marketplace-index fetch + per-skill inspect calls and can be
+        # surfaced to the index builder (see is_rate_limited).
+        self.github = GitHubSource(auth=auth)
 
     def source_id(self) -> str:
         return "claude-marketplace"
 
+    @property
+    def is_rate_limited(self) -> bool:
+        """Whether the underlying GitHub API hit a rate limit during the crawl."""
+        return self.github.is_rate_limited
+
     def trust_level_for(self, identifier: str) -> str:
         parts = identifier.split("/", 2)
         if len(parts) >= 2:
@@ -2415,15 +2499,13 @@ class ClaudeMarketplaceSource(SkillSource):
 
     def fetch(self, identifier: str) -> Optional[SkillBundle]:
         # Delegate to GitHub Contents API since marketplace skills live in GitHub repos
-        gh = GitHubSource(auth=self.auth)
-        bundle = gh.fetch(identifier)
+        bundle = self.github.fetch(identifier)
         if bundle:
             bundle.source = "claude-marketplace"
         return bundle
 
     def inspect(self, identifier: str) -> Optional[SkillMeta]:
-        gh = GitHubSource(auth=self.auth)
-        meta = gh.inspect(identifier)
+        meta = self.github.inspect(identifier)
         if meta:
             meta.source = "claude-marketplace"
             meta.trust_level = self.trust_level_for(identifier)
@@ -2437,16 +2519,15 @@ class ClaudeMarketplaceSource(SkillSource):
             return cached
 
         url = f"https://api.github.com/repos/{repo}/contents/.claude-plugin/marketplace.json"
+        resp = self.github._github_get(
+            url,
+            headers={**self.auth.get_headers(), "Accept": "application/vnd.github.v3.raw"},
+        )
+        if resp is None or resp.status_code != 200:
+            return []
         try:
-            resp = httpx.get(
-                url,
-                headers={**self.auth.get_headers(), "Accept": "application/vnd.github.v3.raw"},
-                timeout=15,
-            )
-            if resp.status_code != 200:
-                return []
             data = json.loads(resp.text)
-        except (httpx.HTTPError, json.JSONDecodeError):
+        except json.JSONDecodeError:
             return []
 
         plugins = data.get("plugins", [])

tui_gateway/server.py

@@ -345,11 +345,44 @@ def _notify_session_boundary(event_type: str, session_id: str | None) -> None:
         pass
 
 
+def _claim_active_session_slot(
+    session_key: str,
+    *,
+    live_session_id: str,
+    surface: str = "tui",
+) -> tuple[Any, str | None]:
+    try:
+        from hermes_cli.active_sessions import try_acquire_active_session
+
+        return try_acquire_active_session(
+            session_id=session_key,
+            surface=surface,
+            config=_load_cfg(),
+            metadata={"live_session_id": live_session_id},
+        )
+    except Exception as exc:
+        logger.warning("Failed to claim active session slot: %s", exc)
+        return None, None
+
+
+def _release_active_session_slot(session: dict | None) -> None:
+    if not session:
+        return
+    lease = session.pop("active_session_lease", None)
+    if lease is None:
+        return
+    try:
+        lease.release()
+    except Exception:
+        logger.debug("Failed to release active session slot", exc_info=True)
+
+
 def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> None:
     """Best-effort finalize hook + memory commit for a session."""
     if not session or session.get("_finalized"):
         return
     session["_finalized"] = True
+    _release_active_session_slot(session)
     stop_event = session.get("_notif_stop")
     if stop_event is not None:
         stop_event.set()
@@ -3284,6 +3317,9 @@ def _(rid, params: dict) -> dict:
 
     ready = threading.Event()
     now = time.time()
+    lease, limit_message = _claim_active_session_slot(key, live_session_id=sid)
+    if limit_message is not None:
+        return _err(rid, 4090, limit_message)
 
     with _sessions_lock:
         _sessions[sid] = {
@@ -3292,6 +3328,7 @@ def _(rid, params: dict) -> dict:
             "agent_ready": ready,
             "attached_images": [],
             "close_on_disconnect": is_truthy_value(params.get("close_on_disconnect", False)),
+            "active_session_lease": lease,
             "cols": cols,
             "created_at": now,
             "edit_snapshots": {},
@@ -3497,6 +3534,9 @@ def _(rid, params: dict) -> dict:
     # _session_resume_lock across it would stall session.close on the main
     # dispatch thread (it's not a _LONG_HANDLER), blocking fast-path RPCs.
     sid = uuid.uuid4().hex[:8]
+    lease, limit_message = _claim_active_session_slot(target, live_session_id=sid)
+    if limit_message is not None:
+        return _err(rid, 4090, limit_message)
     _enable_gateway_prompts()
     home_token = (
         set_hermes_home_override(str(profile_home)) if profile_home is not None else None
@@ -3520,6 +3560,8 @@ def _(rid, params: dict) -> dict:
         finally:
             _clear_session_context(tokens)
     except Exception as e:
+        if lease is not None:
+            lease.release()
         return _err(rid, 5000, f"resume failed: {e}")
     finally:
         if home_token is not None:
@@ -3536,6 +3578,8 @@ def _(rid, params: dict) -> dict:
                     agent.close()
             except Exception:
                 pass
+            if lease is not None:
+                lease.release()
             other_sid, other_session = live
             payload = _live_session_payload(
                 other_sid,
@@ -3555,7 +3599,10 @@ def _(rid, params: dict) -> dict:
                 # skills — must resolve to the resumed profile too).
                 if profile_home is not None:
                     _sessions[sid]["profile_home"] = str(profile_home)
+                _sessions[sid]["active_session_lease"] = lease
         except Exception as e:
+            if lease is not None:
+                lease.release()
             return _err(rid, 5000, f"resume failed: {e}")
         session = _sessions.get(sid) or {}
     return _ok(
@@ -4192,6 +4239,10 @@ def _(rid, params: dict) -> dict:
     if not history:
         return _err(rid, 4008, "nothing to branch — send a message first")
     new_key = _new_session_key()
+    new_sid = uuid.uuid4().hex[:8]
+    lease, limit_message = _claim_active_session_slot(new_key, live_session_id=new_sid)
+    if limit_message is not None:
+        return _err(rid, 4090, limit_message)
     branch_name = params.get("name", "")
     try:
         if branch_name:
@@ -4224,8 +4275,9 @@ def _(rid, params: dict) -> dict:
             )
         db.set_session_title(new_key, title)
     except Exception as e:
+        if lease is not None:
+            lease.release()
         return _err(rid, 5008, f"branch failed: {e}")
-    new_sid = uuid.uuid4().hex[:8]
     try:
         tokens = _set_session_context(new_key)
         try:
@@ -4235,7 +4287,11 @@ def _(rid, params: dict) -> dict:
         _init_session(
             new_sid, new_key, agent, list(history), cols=session.get("cols", 80)
         )
+        if new_sid in _sessions:
+            _sessions[new_sid]["active_session_lease"] = lease
     except Exception as e:
+        if lease is not None:
+            lease.release()
         return _err(rid, 5000, f"agent init failed on branch: {e}")
     return _ok(rid, {"session_id": new_sid, "title": title, "parent": old_key})
 

web/src/i18n/en.ts

@@ -362,7 +362,7 @@ export const en: Translations = {
     inactive: "inactive",
     installBtn: "Install",
     installHeading: "Install from GitHub / Git URL",
-    installHint: "Use owner/repo shorthand or a full https:// or git@ clone URL.",
+    installHint: "Use owner/repo shorthand or a full https:// or git@ clone URL. For a plugin in a subdirectory, append the path: owner/repo/path/to/plugin (or <url>#path/to/plugin).",
     memoryProviderLabel: "Memory provider",
     missingEnvWarn: "Set these in Keys before the plugin can run:",
     noDashboardTab: "No dashboard tab",

web/src/pages/PluginsPage.tsx

@@ -240,7 +240,7 @@ export default function PluginsPage() {
               <Input
                 className="font-mono-ui lowercase"
                 id="install-url"
-                placeholder="owner/repo or https://..."
+                placeholder="owner/repo, owner/repo/subdir, or https://..."
                 spellCheck={false}
                 value={installId}
                 onChange={(e) => setInstallId(e.target.value)}

website/docs/user-guide/configuration.md

@@ -1417,6 +1417,25 @@ The master `streaming.enabled` switch is `false` by default — nothing streams
 
 ## Group Chat Session Isolation
 
+Limit how many chat sessions can actively be open across CLI, TUI/dashboard,
+and messaging gateway:
+
+```yaml
+max_concurrent_sessions: null  # null/0 = unlimited; positive integer = active session cap
+```
+
+When the cap is reached, Hermes returns a direct limit message for new sessions.
+Existing active sessions keep their normal behavior.
+
+The canonical key is top-level `max_concurrent_sessions`. Hermes also accepts
+`gateway.max_concurrent_sessions` as a fallback, but the top-level key wins when
+both are set.
+
+The cap is enforced with a local runtime lease file and is best-effort: Hermes
+fails open if the registry cannot be read or locked so users are not stranded.
+It is intended for a single host/profile runtime, not a shared `$HERMES_HOME`
+mounted across multiple machines.
+
 Control whether shared chats keep one conversation per room or one conversation per participant:
 
 ```yaml

website/docs/user-guide/messaging/photon.md

@@ -206,11 +206,14 @@ hermes photon webhook delete <webhook-id>   # remove one
 
 ## Limits today
 
-- **Attachments are metadata-only.** Inbound webhooks carry the
+- **Inbound attachments are metadata-only.** Inbound webhooks carry the
   filename + MIME type but no download URL — Photon documents an
   attachment retrieval endpoint as roadmap.
-- **Outbound attachments not wired yet.** Easy to add in the sidecar
-  once the agent has reason to send them.
+- **Outbound attachments are supported.** Hermes sends images, voice
+  notes, video, and documents through spectrum-ts' `attachment()` /
+  `voice()` content builders via the sidecar's `/send-attachment`
+  endpoint. Captions arrive as a separate iMessage bubble after the
+  media.
 - **Photon's free quotas:** 5,000 messages per server per day,
   50 new-conversation initiations per shared line per day. Increases
   available — email `help@photon.codes`.