feat(tui): interactive Plugins Hub overlay for enable/disable

The TUI had no way to toggle plugins — `/plugins` only printed a static
list, and the classic `hermes plugins` picker is curses-based and can't
run inside the Ink UI. Users had to drop to a separate shell and run
`hermes plugins enable/disable`.

Add a PluginsHub overlay modeled on the existing SkillsHub:

- New gateway RPC `plugins.manage` (list + toggle) backed by the same
  disk-discovery + dashboard_set_agent_plugin_enabled primitives the CLI
  and dashboard already use, so all three surfaces agree on state. The
  toggle path also wires the plugin's toolset into platform_toolsets.
- `/plugins` with no arg opens the hub; any subcommand still falls
  through to the text slash worker for CLI parity.
- pluginsHub overlay state threaded through overlayStore / interfaces /
  useInputHandlers (Esc closes) / appOverlays (renders the FloatBox);
  preserved across turn teardown like other user-toggled overlays.
- Hub UI: arrow/number select, Enter/Space toggles live, Tab switches
  user-only vs all (bundled) scope, shows ✓/✗/○ activation glyphs.

plugins.manage added to _LONG_HANDLERS (disk + config I/O).

apps/desktop/src/components/assistant-ui/thread.tsx

@@ -77,6 +77,7 @@ import type { HermesGateway } from '@/hermes'
 import { useResizeObserver } from '@/hooks/use-resize-observer'
 import { useI18n } from '@/i18n'
 import { DATA_IMAGE_URL_RE } from '@/lib/embedded-images'
+import { LinkifiedText } from '@/lib/external-link'
 import { triggerHaptic } from '@/lib/haptics'
 import { GitBranchIcon, Loader2Icon, Volume2Icon, VolumeXIcon } from '@/lib/icons'
 import { extractPreviewTargets } from '@/lib/preview-targets'
@@ -919,7 +920,7 @@ const SystemMessage: FC = () => {
       >
         <span className="font-mono text-muted-foreground/55">{slashStatus.groups.command}</span>
         <span className="mx-1.5 text-muted-foreground/35">·</span>
-        <span className="whitespace-pre-wrap">{slashStatus.groups.output.trim()}</span>
+        <LinkifiedText className="whitespace-pre-wrap" explicitOnly pretty={false} text={slashStatus.groups.output.trim()} />
       </MessagePrimitive.Root>
     )
   }
@@ -930,7 +931,7 @@ const SystemMessage: FC = () => {
       data-role="system"
       data-slot="aui_system-message-root"
     >
-      <span className="whitespace-pre-wrap">{text}</span>
+      <LinkifiedText className="whitespace-pre-wrap" explicitOnly pretty={false} text={text} />
     </MessagePrimitive.Root>
   )
 }

apps/desktop/src/lib/external-link.test.tsx

@@ -165,4 +165,31 @@ describe('external link helpers', () => {
       'https://expedia.com/things-to-do/puerto-rico-el-yunque-rainforest-adventure'
     )
   })
+
+  it('explicitOnly skips bare filename/domain tokens and only links explicit URLs', () => {
+    installDesktopBridge()
+
+    render(
+      <LinkifiedText
+        explicitOnly
+        pretty={false}
+        text={'Report  https://paste.rs/abc\nagent.log  https://paste.rs/def\nerrors.log'}
+      />
+    )
+
+    const links = screen.getAllByRole('link')
+    expect(links.map(a => a.getAttribute('href'))).toEqual(['https://paste.rs/abc', 'https://paste.rs/def'])
+    // Bare filename-shaped tokens stay as plain text, not links.
+    expect(screen.queryByText(content => content.includes('agent.log'))).toBeTruthy()
+    expect(links.some(a => (a.textContent ?? '').includes('.log'))).toBe(false)
+  })
+
+  it('without explicitOnly, bare filename tokens are still linkified (default behavior)', () => {
+    installDesktopBridge()
+
+    render(<LinkifiedText pretty={false} text="open agent.log please" />)
+
+    const link = screen.getByRole('link', { name: 'agent.log' })
+    expect(link.getAttribute('href')).toBe('https://agent.log')
+  })
 })

apps/desktop/src/lib/external-link.tsx

@@ -12,6 +12,12 @@ const titleSubs = new Map<string, Set<(value: string) => void>>()
 const URL_RE =
   /(?:https?:\/\/|www\.)[^\s<>"'`]+[^\s<>"'`.,;:!?)]|[a-z0-9](?:[a-z0-9-]*\.)+[a-z]{2,}(?:\/[^\s<>"'`.,;:!?)]*)?/gi
 
+// Explicit-scheme / www. URLs only — no bare-domain matching. Used where the
+// surrounding text is full of filename-shaped tokens (e.g. `agent.log`,
+// `errors.log` in a /debug report) that the bare-domain branch of URL_RE would
+// otherwise mistake for domains and linkify.
+const EXPLICIT_URL_RE = /(?:https?:\/\/|www\.)[^\s<>"'`]+[^\s<>"'`.,;:!?)]/gi
+
 const DOMAIN_RE = /^(?:www\.)?[a-z0-9](?:[a-z0-9-]*\.)+[a-z]{2,}(?::\d+)?(?:[/?#][^\s]*)?$/i
 const SKIP_PROTO_RE = /^(?:file|data|mailto|javascript|blob|chrome|about|hermes):/i
 const LOCAL_HOST_RE = /^(?:localhost|127\.0\.0\.1|0\.0\.0\.0|\[::1\])(?::\d+)?$/i
@@ -261,13 +267,14 @@ interface LinkifiedTextProps {
   className?: string
   text: string
   pretty?: boolean
+  explicitOnly?: boolean
 }
 
-export function LinkifiedText({ className, pretty = true, text }: LinkifiedTextProps) {
+export function LinkifiedText({ className, explicitOnly = false, pretty = true, text }: LinkifiedTextProps) {
   const nodes: ReactNode[] = []
   let cursor = 0
 
-  for (const match of text.matchAll(URL_RE)) {
+  for (const match of text.matchAll(explicitOnly ? EXPLICIT_URL_RE : URL_RE)) {
     const raw = match[0]
     const url = normalizeExternalUrl(raw)
     const index = match.index ?? 0

cli.py

@@ -6218,27 +6218,20 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         choices visible and lets the normal Enter key binding submit the typed
         or highlighted choice.
 
-        **Platform note (Windows dead-lock — issue #30768):**
-        The queue-based modal relies on prompt_toolkit key bindings receiving
-        keyboard events and calling ``_submit_slash_confirm_response``.  On
-        Windows (PowerShell / Windows Terminal) the prompt_toolkit input
-        channel can become unresponsive when the modal is entered from the
-        ``process_loop`` daemon thread, causing a dead-lock: the user sees the
-        confirmation panel but keystrokes never reach the key bindings and the
-        ``response_queue.get()`` blocks until the 120-second timeout expires.
+        **Platform note (Windows — issue #33961):**
+        Earlier code bypassed the modal on ``sys.platform == "win32"`` and fell
+        back to a raw ``input()`` prompt.  When the confirm was triggered from the
+        ``process_loop`` daemon thread (the normal case) that ``input()`` ran off
+        the main thread and deadlocked against prompt_toolkit's stdin ownership —
+        the user saw a frozen cursor and Ctrl-C was swallowed (bare ``/reset``
+        froze; ``/reset now`` worked only because it skips the prompt entirely).
 
-        To avoid this, we fall back to ``_prompt_text_input`` (a simple
-        ``input()``-based prompt) when any of these conditions hold:
-
-        * ``sys.platform == "win32"`` — native Windows console (ConPTY /
-          win32_input) does not support the modal reliably.
-        * ``self._app`` is not set — unit tests / non-interactive contexts.
-
-        On non-Windows platforms the modal itself is still safe from the
-        ``process_loop`` daemon thread as long as the main-thread event loop
-        owns the prompt_toolkit buffer mutations.  When we are off the main
-        thread, schedule the modal snapshot / restore work on ``self._app.loop``
-        via ``call_soon_threadsafe`` and keep the queue-based response path.
+        Native Windows now uses the same path as Linux/macOS: the modal is set up
+        on ``self._app.loop`` via ``call_soon_threadsafe`` and answered by the
+        normal prompt_toolkit key bindings (the same input channel that already
+        handles ordinary typing on Windows).  The raw ``input()`` fallback is kept
+        only for the genuinely safe cases: no running app (unit tests /
+        non-interactive), no resolvable event loop, or a scheduling failure.
         """
         import threading
         import time as _time
@@ -6251,23 +6244,26 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         if not getattr(self, "_app", None):
             return self._prompt_text_input("Choice [1/2/3]: ")
 
-        # On Windows the prompt_toolkit input channel can deadlock when the
-        # modal is entered from the process_loop daemon thread — keystrokes
-        # never reach the key bindings, so response_queue.get() blocks for
-        # the full timeout (issue #30768).  Fall back to the simpler
-        # stdin-based prompt which works reliably on Windows.
-        if sys.platform == "win32":
-            return self._prompt_text_input("Choice [1/2/3]: ")
-
         try:
             app_loop = self._app.loop
         except Exception:
             app_loop = None
 
         in_main_thread = threading.current_thread() is threading.main_thread()
-        if not in_main_thread and app_loop is None:
+
+        def _stdin_fallback() -> str | None:
+            # On native Windows a raw input() from a non-main thread deadlocks
+            # against prompt_toolkit's stdin ownership (#33961).  With an app
+            # running we cannot safely prompt off the main thread, so cancel
+            # cleanly (None) rather than hang the terminal.
+            if sys.platform == "win32" and not in_main_thread:
+                self._invalidate()
+                return None
             return self._prompt_text_input("Choice [1/2/3]: ")
 
+        if not in_main_thread and app_loop is None:
+            return _stdin_fallback()
+
         response_queue = queue.Queue()
 
         def _setup_modal() -> None:
@@ -6307,7 +6303,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
             return ready.wait(timeout=5)
 
         if not _run_on_app_loop(_setup_modal):
-            return self._prompt_text_input("Choice [1/2/3]: ")
+            return _stdin_fallback()
 
         _last_countdown_refresh = _time.monotonic()
         try:
@@ -8223,9 +8219,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
             print("  ⚠️  MCP reload timed out (30s). Some servers may not have reconnected.")
 
     # Inline-skip tokens that bypass the destructive-slash confirmation modal.
-    # Matches the escape-hatch pattern users on broken modal platforms
-    # (currently native Windows PowerShell — issue #30768) need to self-serve
-    # without having to flip approvals.destructive_slash_confirm in config.
+    # A general escape hatch for non-interactive use (scripting/automation) and
+    # for the degraded path where the modal can't be marshaled onto the app loop
+    # — lets users self-serve without flipping approvals.destructive_slash_confirm
+    # in config. (Native Windows now drives the modal normally — see #33961.)
     _DESTRUCTIVE_SKIP_TOKENS = frozenset({"now", "--yes", "-y"})
 
     @classmethod
@@ -8283,8 +8280,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         Inline-skip: if ``cmd_original`` contains ``now``, ``--yes``, or
         ``-y`` as an argument (e.g. ``/reset now``, ``/new --yes My title``),
         the modal is bypassed and ``"once"`` is returned immediately. This is
-        an escape hatch for platforms where the prompt_toolkit modal hangs
-        (issue #30768 — native Windows PowerShell). Callers are responsible
+        an escape hatch for non-interactive use and for the degraded path where
+        the modal can't be marshaled onto the app loop (native Windows itself now
+        drives the modal normally — see #33961). Callers are responsible
         for stripping the skip tokens from any remaining argument parsing
         (see :meth:`_split_destructive_skip`).
 

gateway/platforms/api_server.py

@@ -3510,35 +3510,46 @@ class APIServerAdapter(BasePlatformAdapter):
         loop = asyncio.get_running_loop()
 
         def _run():
-            agent = self._create_agent(
-                ephemeral_system_prompt=ephemeral_system_prompt,
-                session_id=session_id,
-                stream_delta_callback=stream_delta_callback,
-                tool_progress_callback=tool_progress_callback,
-                tool_start_callback=tool_start_callback,
-                tool_complete_callback=tool_complete_callback,
-                gateway_session_key=gateway_session_key,
+            from gateway.session_context import clear_session_vars, set_session_vars
+
+            tokens = set_session_vars(
+                platform="api_server",
+                chat_id=session_id or "",
+                session_key=gateway_session_key or session_id or "",
+                session_id=session_id or "",
             )
-            if agent_ref is not None:
-                agent_ref[0] = agent
-            effective_task_id = session_id or str(uuid.uuid4())
-            result = agent.run_conversation(
-                user_message=user_message,
-                conversation_history=conversation_history,
-                task_id=effective_task_id,
-            )
-            usage = {
-                "input_tokens": getattr(agent, "session_prompt_tokens", 0) or 0,
-                "output_tokens": getattr(agent, "session_completion_tokens", 0) or 0,
-                "total_tokens": getattr(agent, "session_total_tokens", 0) or 0,
-            }
-            # Include the effective session ID in the result so callers
-            # (e.g. X-Hermes-Session-Id header) can track compression-
-            # triggered session rotations. (#16938)
-            _eff_sid = getattr(agent, "session_id", session_id)
-            if isinstance(_eff_sid, str) and _eff_sid:
-                result["session_id"] = _eff_sid
-            return result, usage
+            try:
+                agent = self._create_agent(
+                    ephemeral_system_prompt=ephemeral_system_prompt,
+                    session_id=session_id,
+                    stream_delta_callback=stream_delta_callback,
+                    tool_progress_callback=tool_progress_callback,
+                    tool_start_callback=tool_start_callback,
+                    tool_complete_callback=tool_complete_callback,
+                    gateway_session_key=gateway_session_key,
+                )
+                if agent_ref is not None:
+                    agent_ref[0] = agent
+                effective_task_id = session_id or str(uuid.uuid4())
+                result = agent.run_conversation(
+                    user_message=user_message,
+                    conversation_history=conversation_history,
+                    task_id=effective_task_id,
+                )
+                usage = {
+                    "input_tokens": getattr(agent, "session_prompt_tokens", 0) or 0,
+                    "output_tokens": getattr(agent, "session_completion_tokens", 0) or 0,
+                    "total_tokens": getattr(agent, "session_total_tokens", 0) or 0,
+                }
+                # Include the effective session ID in the result so callers
+                # (e.g. X-Hermes-Session-Id header) can track compression-
+                # triggered session rotations. (#16938)
+                _eff_sid = getattr(agent, "session_id", session_id)
+                if isinstance(_eff_sid, str) and _eff_sid:
+                    result["session_id"] = _eff_sid
+                return result, usage
+            finally:
+                clear_session_vars(tokens)
 
         return await loop.run_in_executor(None, _run)
 

gateway/platforms/base.py

@@ -1795,9 +1795,11 @@ class BasePlatformAdapter(ABC):
 
     # Whether this platform renders triple-backtick fenced code blocks (i.e.
     # ``format_message`` translates/preserves markdown fences into a real code
-    # block).  Drives presentation choices like rendering a ``terminal`` tool
-    # call's command as a ```bash block instead of a flat preview line.
+    # block).  Capability flag for markdown-aware presentation choices.
     # Default False (plain-text platforms); markdown-rendering adapters set True.
+    # Note: tool-progress deliberately does NOT use this to render a terminal
+    # command as a ```bash block — that exposed full commands in chat. Progress
+    # shows a short truncated preview only (see gateway/run.py progress_callback).
     supports_code_blocks: bool = False
 
     def __init__(self, config: PlatformConfig, platform: Platform):

gateway/platforms/telegram.py

@@ -181,6 +181,8 @@ def _strip_mdv2(text: str) -> str:
     """
     # Remove escape backslashes before special characters
     cleaned = re.sub(r'\\([_*\[\]()~`>#\+\-=|{}.!\\])', r'\1', text)
+    # Remove standard markdown bold (**text** → text) BEFORE MarkdownV2 bold
+    cleaned = re.sub(r'\*\*([^*]+)\*\*', r'\1', cleaned)
     # Remove MarkdownV2 bold markers that format_message converted from **bold**
     cleaned = re.sub(r'\*([^*]+)\*', r'\1', cleaned)
     # Remove MarkdownV2 italic markers that format_message converted from *italic*
@@ -2208,11 +2210,17 @@ class TelegramAdapter(BasePlatformAdapter):
                 # "Message is not modified" is a no-op, not an error
                 if "not modified" in str(fmt_err).lower():
                     return SendResult(success=True, message_id=message_id)
-                # Fallback: retry without markdown formatting
+                # Fallback: strip MarkdownV2 escapes and retry as clean plain text
+                logger.warning(
+                    "[%s] MarkdownV2 edit failed, falling back to plain text: %s",
+                    self.name,
+                    fmt_err,
+                )
+                _plain = _strip_mdv2(content) if content else content
                 await self._bot.edit_message_text(
                     chat_id=int(chat_id),
                     message_id=int(message_id),
-                    text=content,
+                    text=_plain,
                 )
             return SendResult(success=True, message_id=message_id)
         except Exception as e:

gateway/run.py

@@ -12971,32 +12971,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             # Build progress message with primary argument preview
             from agent.display import get_tool_emoji
             emoji = get_tool_emoji(tool_name, default="⚙️")
-
-            # Markdown-capable platforms render a terminal command as a native
-            # ```bash fenced block (full command, no quotes, no label, no
-            # truncation) instead of the noisy `terminal: "cmd…"` line.  Gated
-            # on the adapter's ``supports_code_blocks`` capability so every
-            # markdown-rendering platform (and plugin adapters that opt in) gets
-            # it, while plain-text platforms keep the compact line.
-            _bash_block = None
-            try:
-                _progress_adapter = self.adapters.get(source.platform)
-            except Exception:
-                _progress_adapter = None
-            if (
-                getattr(_progress_adapter, "supports_code_blocks", False)
-                and tool_name == "terminal"
-                and isinstance(args, dict)
-                and isinstance(args.get("command"), str)
-                and args["command"].strip()
-            ):
-                _bash_block = f"```bash\n{args['command'].rstrip()}\n```"
             
             # Verbose mode: show detailed arguments, respects tool_preview_length
             if progress_mode == "verbose":
-                if _bash_block is not None:
-                    msg = _bash_block
-                elif args:
+                if args:
                     from agent.display import get_tool_preview_max_len
                     _pl = get_tool_preview_max_len()
                     args_str = json.dumps(args, ensure_ascii=False, default=str)
@@ -13016,9 +12994,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             # "all" / "new" modes: short preview, respects tool_preview_length
             # config (defaults to 40 chars when unset to keep gateway messages
             # compact — unlike CLI spinners, these persist as permanent messages).
-            if _bash_block is not None:
-                msg = _bash_block
-            elif preview:
+            if preview:
                 from agent.display import get_tool_preview_max_len
                 _pl = get_tool_preview_max_len()
                 _cap = _pl if _pl > 0 else 40
@@ -13130,6 +13106,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     "message_id": message_id,
                     "content": content,
                 }
+                if getattr(adapter, "REQUIRES_EDIT_FINALIZE", False):
+                    kwargs["finalize"] = True
                 if _edit_accepts_metadata:
                     kwargs["metadata"] = _progress_metadata
                 return await adapter.edit_message(**kwargs)

gateway/session_context.py

@@ -106,6 +106,7 @@ def set_session_vars(
     user_id: str = "",
     user_name: str = "",
     session_key: str = "",
+    session_id: str = "",
     message_id: str = "",
     cwd: str = "",
 ) -> list:
@@ -127,6 +128,7 @@ def set_session_vars(
         _SESSION_USER_ID.set(user_id),
         _SESSION_USER_NAME.set(user_name),
         _SESSION_KEY.set(session_key),
+        _SESSION_ID.set(session_id),
         _SESSION_MESSAGE_ID.set(message_id),
     ]
     try:
@@ -157,6 +159,7 @@ def clear_session_vars(tokens: list) -> None:
         _SESSION_USER_ID,
         _SESSION_USER_NAME,
         _SESSION_KEY,
+        _SESSION_ID,
         _SESSION_MESSAGE_ID,
     ):
         var.set("")

hermes_cli/dashboard_register.py

@@ -94,19 +94,36 @@ def _register_self_hosted_client(
     *,
     access_token: str,
     portal_base_url: str,
-    name: str,
+    name: Optional[str],
     custom_redirect_uri: Optional[str],
+    existing_client_id: Optional[str] = None,
     timeout: float = 15.0,
 ) -> dict:
     """POST to the portal's self-hosted-client endpoint and return the JSON body.
 
+    When ``existing_client_id`` is provided (the client_id this install
+    persisted on a prior run), it is sent so the portal updates that existing
+    dashboard record in place instead of minting a duplicate — this is what
+    makes re-running ``hermes dashboard register`` idempotent. The portal
+    falls back to creating a fresh client if the id no longer resolves to a row
+    in the caller's org (stale/deleted), so passing it is always safe.
+
+    ``name`` may be ``None`` on the idempotent update path (re-run without an
+    explicit ``--name``): omitting it tells the portal to keep the name it
+    already stored rather than overwriting it. It is required on the create
+    path; the caller guarantees a value there.
+
     Raises RuntimeError with a user-facing message on any non-2xx response or
     transport failure.
     """
     url = f"{portal_base_url.rstrip('/')}/api/oauth/self-hosted-client"
-    body: dict[str, str] = {"name": name}
+    body: dict[str, str] = {}
+    if name:
+        body["name"] = name
     if custom_redirect_uri:
         body["custom_redirect_uri"] = custom_redirect_uri
+    if existing_client_id:
+        body["client_id"] = existing_client_id
 
     data = json.dumps(body).encode("utf-8")
     req = urllib.request.Request(
@@ -165,16 +182,20 @@ def _print_post_register_hint(
     portal_base_url: str,
     custom_redirect_uri: Optional[str],
     wrote_portal_url: bool,
+    public_url: str = "",
 ) -> None:
     """Print the success summary + the gate-engagement caveat."""
     from hermes_cli.config import get_env_path
 
     env_path = get_env_path()
+    _cid = client_id
     print()
     print(f"  Wrote to {env_path}:")
-    print(f"    HERMES_DASHBOARD_OAUTH_CLIENT_ID={client_id}")
+    print("    HERMES_DASHBOARD_OAUTH_CLIENT_ID=" + str(_cid))
     if wrote_portal_url:
-        print(f"    HERMES_DASHBOARD_PORTAL_URL={portal_base_url}")
+        print("    HERMES_DASHBOARD_PORTAL_URL=" + str(portal_base_url))
+    if public_url:
+        print("    HERMES_DASHBOARD_PUBLIC_URL=" + str(public_url))
     print()
     print(
         "  Heads up — Nous login only *engages* on a non-loopback bind. A plain\n"
@@ -240,12 +261,48 @@ def cmd_dashboard_register(args) -> None:
 
     # Portal override: explicit --portal-url flag wins, else the
     # HERMES_DASHBOARD_PORTAL_URL env var, else the stored login's portal.
+    #
+    # We track whether a custom URL was *explicitly supplied* (flag or env)
+    # separately from the resolved value. An explicit custom URL is an
+    # intentional choice the user wants to persist (and update in place if it
+    # already exists in .env); a portal merely inferred from the stored login
+    # keeps the older, more conservative write-only-if-absent behaviour so we
+    # don't clutter .env for the common production case.
     portal_override = getattr(args, "portal_url", None) or os.environ.get(
         "HERMES_DASHBOARD_PORTAL_URL"
     )
+    custom_portal_supplied = bool(
+        isinstance(portal_override, str) and portal_override.strip()
+    )
     portal_base_url = _resolve_portal_base_url(portal_override)
 
-    name = getattr(args, "name", None) or _generate_dashboard_name()
+    # Idempotency: if this install already registered a dashboard, we hold its
+    # client_id locally (HERMES_DASHBOARD_OAUTH_CLIENT_ID). Re-send it so the
+    # portal UPDATES that existing record instead of creating a duplicate. No
+    # stored client_id -> this is a first registration -> create a fresh one
+    # (the original behavior). This mirrors the portal's rule: no client id =
+    # new dashboard; client id present = the stable key of the row to modify.
+    existing_client_id = None
+    try:
+        existing_client_id = get_env_value("HERMES_DASHBOARD_OAUTH_CLIENT_ID")
+    except Exception:
+        existing_client_id = None
+    if isinstance(existing_client_id, str):
+        existing_client_id = existing_client_id.strip() or None
+    else:
+        existing_client_id = None
+
+    explicit_name = getattr(args, "name", None)
+    # Auto-generate a random name ONLY for a first registration. On a re-run
+    # (we hold a client_id) without an explicit --name, keep the name the
+    # portal already stored rather than churning it to a new random value
+    # every time — so leave `name` unset and let the portal preserve it.
+    if explicit_name:
+        name = explicit_name
+    elif existing_client_id:
+        name = None
+    else:
+        name = _generate_dashboard_name()
     custom_redirect_uri = getattr(args, "redirect_uri", None)
 
     # 2. Register with the portal.
@@ -255,20 +312,26 @@ def cmd_dashboard_register(args) -> None:
             portal_base_url=portal_base_url,
             name=name,
             custom_redirect_uri=custom_redirect_uri,
+            existing_client_id=existing_client_id,
         )
     except RuntimeError as exc:
         print(f"✗ Registration failed: {exc}")
         sys.exit(1)
 
     client_id = str(result["client_id"])
-    registered_name = str(result.get("name") or name)
+    registered_name = str(result.get("name") or name or "")
 
-    print(f'✓ Registered dashboard "{registered_name}"')
+    # Distinguish create vs update for the user: the portal echoes back the
+    # same client_id we sent when it updated in place.
+    updated_existing = bool(
+        existing_client_id and client_id == existing_client_id
+    )
+    if updated_existing:
+        print(f'✓ Updated dashboard "{registered_name}"')
+    else:
+        print(f'✓ Registered dashboard "{registered_name}"')
 
-    # 3. Write env vars idempotently. Always set the client_id. Only set the
-    #    portal URL when it isn't already configured (env or config) AND differs
-    #    from the production default, so we don't clutter .env for the common case
-    #    but DO persist a non-default portal (e.g. a preview deploy used in dev).
+    # 3. Write env vars idempotently. Always set the client_id.
     try:
         save_env_value("HERMES_DASHBOARD_OAUTH_CLIENT_ID", client_id)
     except Exception as exc:
@@ -276,6 +339,18 @@ def cmd_dashboard_register(args) -> None:
         print(f"  Set it manually:  HERMES_DASHBOARD_OAUTH_CLIENT_ID={client_id}")
         sys.exit(1)
 
+    # Persist the portal URL. Two cases:
+    #   a) The user explicitly supplied a custom portal (--portal-url flag or
+    #      HERMES_DASHBOARD_PORTAL_URL env). That's an intentional choice we
+    #      always persist so it survives across sessions — overwriting any
+    #      existing entry in place (save_env_value updates a matching key
+    #      rather than appending a duplicate). This is true even when it equals
+    #      the production default: the user asked for it explicitly.
+    #   b) No custom portal was supplied. Keep the older conservative behaviour:
+    #      only write a portal inferred from the stored login when it isn't
+    #      already configured AND differs from the production default, so we
+    #      don't clutter .env for the common production case and don't alter an
+    #      existing entry unexpectedly.
     wrote_portal_url = False
     default_portal = "https://portal.nousresearch.com"
     existing_portal = None
@@ -283,7 +358,15 @@ def cmd_dashboard_register(args) -> None:
         existing_portal = get_env_value("HERMES_DASHBOARD_PORTAL_URL")
     except Exception:
         existing_portal = None
-    if not existing_portal and portal_base_url.rstrip("/") != default_portal:
+
+    if custom_portal_supplied:
+        should_write_portal = existing_portal != portal_base_url
+    else:
+        should_write_portal = (
+            not existing_portal and portal_base_url.rstrip("/") != default_portal
+        )
+
+    if should_write_portal:
         try:
             save_env_value("HERMES_DASHBOARD_PORTAL_URL", portal_base_url)
             wrote_portal_url = True
@@ -291,10 +374,54 @@ def cmd_dashboard_register(args) -> None:
             # Non-fatal: the client_id is the load-bearing value.
             pass
 
+    # Persist the dashboard public URL derived from the OAuth redirect URI.
+    #
+    # --redirect-uri is the full public HTTPS callback the user registered with
+    # the portal, e.g. https://hermes.example.com/auth/callback. At serve time
+    # the dashboard auth layer (dashboard_auth/routes._redirect_uri) reconstructs
+    # that same callback by taking HERMES_DASHBOARD_PUBLIC_URL and appending
+    # "/auth/callback" verbatim. So the value the runtime actually consumes is
+    # the ORIGIN (scheme://host[:port]), not the full callback path — persisting
+    # the raw redirect URI would double up the path. We derive the origin from
+    # the supplied redirect URI and persist it as HERMES_DASHBOARD_PUBLIC_URL so
+    # the operator doesn't have to re-supply it and the public-URL override is
+    # actually wired (the gate engages and the callback round-trips correctly).
+    #
+    # Like the portal URL, an explicitly supplied value is always written
+    # (updating an existing entry in place rather than appending a duplicate),
+    # a no-op when it already matches, and never written on a localhost-only
+    # install (no --redirect-uri).
+    wrote_public_url = False
+    public_url = ""
+    if custom_redirect_uri:
+        try:
+            from urllib.parse import urlparse
+
+            parsed = urlparse(custom_redirect_uri)
+            if parsed.scheme in ("http", "https") and parsed.netloc:
+                public_url = f"{parsed.scheme}://{parsed.netloc}"
+        except Exception:
+            public_url = ""
+
+    if public_url:
+        existing_public_url = None
+        try:
+            existing_public_url = get_env_value("HERMES_DASHBOARD_PUBLIC_URL")
+        except Exception:
+            existing_public_url = None
+        if existing_public_url != public_url:
+            try:
+                save_env_value("HERMES_DASHBOARD_PUBLIC_URL", public_url)
+                wrote_public_url = True
+            except Exception:
+                # Non-fatal: the client_id is the load-bearing value.
+                pass
+
     # 4. Hint.
     _print_post_register_hint(
         client_id=client_id,
         portal_base_url=portal_base_url,
         custom_redirect_uri=custom_redirect_uri,
         wrote_portal_url=wrote_portal_url,
+        public_url=public_url if wrote_public_url else "",
     )

hermes_cli/doctor.py

@@ -738,11 +738,14 @@ def run_doctor(args):
                         issues,
                     )
 
-            # Warn if model is set to a provider-prefixed name on a provider that doesn't use them
+            # Warn if model is set to a provider-prefixed name on a provider that doesn't use them.
+            # Vendor/model slugs are valid on aggregator-style providers and on any custom
+            # provider — bare "custom" or a named "custom:<name>" that fronts an OpenAI-compatible
+            # aggregator (e.g. custom:hpc-ai serving deepseek/deepseek-v4-flash) requires the prefix.
             provider_for_policy = runtime_provider or catalog_provider
+            provider_policy_id = str(provider_for_policy or "").strip().lower()
             providers_accepting_vendor_slugs = {
                 "openrouter",
-                "custom",
                 "auto",
                 "kilocode",
                 "opencode-zen",
@@ -750,11 +753,16 @@ def run_doctor(args):
                 "lmstudio",
                 "nous",
             }
+            provider_accepts_vendor_slug = (
+                provider_policy_id in providers_accepting_vendor_slugs
+                or provider_policy_id == "custom"
+                or provider_policy_id.startswith("custom:")
+            )
             if (
                 default_model
                 and "/" in default_model
-                and provider_for_policy
-                and provider_for_policy not in providers_accepting_vendor_slugs
+                and provider_policy_id
+                and not provider_accepts_vendor_slug
             ):
                 check_warn(
                     f"model.default '{default_model}' uses a vendor/model slug but provider is '{provider_raw}'",

hermes_cli/providers.py

@@ -492,7 +492,10 @@ def get_label(provider_id: str) -> str:
 
 def is_aggregator(provider: str) -> bool:
     """Return True when the provider is a multi-model aggregator."""
-    pdef = get_provider(provider)
+    provider_norm = normalize_provider(provider or "")
+    if provider_norm.startswith("custom:"):
+        return True
+    pdef = get_provider(provider_norm)
     return pdef.is_aggregator if pdef else False
 
 

plugins/platforms/photon/README.md

@@ -1,123 +1,136 @@
 # Photon iMessage platform plugin
 
-This plugin connects Hermes Agent to iMessage (and WhatsApp Business +
-future Spectrum interfaces) through [Photon][photon] — a managed
-service that handles the iMessage line allocation, delivery, and
-abuse-prevention layer so users don't have to run their own Mac
-relay.
+This plugin connects Hermes Agent to iMessage (and other Spectrum
+interfaces) through [Photon][photon] — a managed service that handles
+iMessage line allocation, delivery, and abuse-prevention so users don't
+have to run their own Mac relay.
 
-The free tier uses Photon's shared iMessage line pool (`type: shared`)
-and is the path we recommend for everyone who doesn't already pay for a
-dedicated number.
+The free tier uses Photon's shared iMessage line pool and is the path we
+recommend for everyone who doesn't already pay for a dedicated number.
 
 ## Architecture
 
+Like Discord and Slack, Photon is a **persistent-connection** channel — no
+public URL, no webhook, no signing secret. The `spectrum-ts` SDK holds a
+long-lived **gRPC stream** to Photon for both directions. Because the SDK is
+TypeScript-only, Hermes runs it inside a small supervised Node sidecar and
+talks to it over loopback.
+
 ```
-┌─────────────────────────┐    HMAC-signed POSTs      ┌──────────────────┐
-│  Photon Spectrum cloud  │ ──────────────────────►   │  Hermes Agent    │
-│  (iMessage line owner)  │                           │  (Python)        │
-└─────────────────────────┘    JSON over loopback     │                  │
-        ▲                  ◄──────────────────────    │  PhotonAdapter   │
-        │                                             │  + aiohttp recv  │
-        │  spectrum-ts                                │                  │
-        │  SDK (Node)                                 │  spawns + super- │
-        ▼                                             │  vises ▼         │
-┌─────────────────────────┐                           ├──────────────────┤
-│  Node sidecar           │   ◄────  X-Hermes-      ─ │  Node sidecar    │
-│  (plugins/.../sidecar)  │       Sidecar-Token       │  child process   │
-└─────────────────────────┘                           └──────────────────┘
+                         gRPC (spectrum-ts)
+┌─────────────────────────┐ ◄───────────────► ┌──────────────────────┐
+│  Photon Spectrum cloud  │   app.messages    │  Node sidecar        │
+│  (iMessage line owner)  │   space.send()    │  (plugins/…/sidecar) │
+└─────────────────────────┘                   └──────────┬───────────┘
+                                       GET /inbound (NDJSON) │  ▲ POST /send
+                                       inbound events        ▼  │ /typing
+                                              ┌──────────────────────┐
+                                              │  PhotonAdapter        │
+                                              │  (Python, in gateway) │
+                                              └──────────────────────┘
 ```
 
-Inbound traffic is webhook-only — Hermes runs an aiohttp listener
-that verifies `X-Spectrum-Signature` and dedupes on `message.id`.
-
-Outbound traffic goes through a tiny Node sidecar that runs the
-`spectrum-ts` SDK. Photon does not currently expose an HTTP
-send-message endpoint; their own docs say:
-
-> Pass `space.id` to `Space.send(...)` from a separate `spectrum-ts`
-> SDK instance to reply. **No public HTTP send endpoint exists today.**
-> — https://photon.codes/docs/webhooks/events
-
-When Photon ships an HTTP send endpoint, `_sidecar_send` is the one
-function that swaps and the sidecar disappears. The rest of the
-plugin stays the same.
+- **Inbound**: the sidecar consumes the SDK's `app.messages` gRPC stream,
+  normalizes each message, and streams it to the adapter over a loopback
+  `GET /inbound` (NDJSON). The adapter dedupes on `messageId` and dispatches
+  a `MessageEvent` to the gateway. It reconnects automatically if the stream
+  drops; the sidecar owns the gRPC reconnect to Photon.
+- **Outbound**: `send` / `send_typing` are loopback POSTs to the sidecar,
+  authenticated with a shared `X-Hermes-Sidecar-Token`.
 
 ## First-time setup
 
 ```bash
-# 1. One-shot setup: device login (opens browser) + project + user + sidecar deps
+# One-shot setup: device login (opens browser) + project + user + sidecar deps
 hermes photon setup --phone +15551234567
 
-# 2. Expose your webhook URL to the public internet
-#    (cloudflared, ngrok, your gateway's public hostname, etc.)
-#    Then register it with Photon:
-hermes photon webhook register https://your-host.example.com/photon/webhook
-
-# 3. Save the signing secret it prints to ~/.hermes/.env
-#    as PHOTON_WEBHOOK_SECRET=...
-#    Photon only returns it ONCE.
-
-# 4. Start the gateway
+# Start the gateway
 hermes gateway start --platform photon
 ```
 
-`hermes photon setup` runs the RFC 8628 device-code login as its first
-step — it opens `https://app.photon.codes/` for approval, then
-provisions the Spectrum project + iMessage line. There is no separate
-`login` command; like every other Hermes channel, onboarding goes
-through one setup surface. Re-running `setup` reuses an existing token
-and project, so it's safe to run again to finish a partial setup.
+`hermes photon setup` does, in order:
+
+1. **Device login** (RFC 8628, `client_id=photon-cli`) — opens
+   `https://app.photon.codes/` for approval and stores the bearer token.
+2. **Find or create** the `Hermes Agent` project on the Photon dashboard.
+3. **Enable Spectrum**, read the project's `spectrumProjectId`, rotate the
+   project secret, and persist both.
+4. **Register your phone number** as a Spectrum user (idempotent — skipped if
+   a user with that number already exists).
+5. **Print the assigned iMessage line** — the number you text to reach your
+   agent.
+6. **Install the sidecar deps** (`spectrum-ts`).
+
+There is no separate `login` command; like every other Hermes channel,
+onboarding goes through one setup surface. Re-running `setup` reuses an
+existing token/project, so it's safe to run again to finish a partial setup.
+Run `hermes photon status` to see what's configured.
 
 ## Credentials
 
-Stored in `~/.hermes/auth.json` under `credential_pool`:
+Runtime SDK credentials live in `~/.hermes/.env` (the same place every other
+channel keeps its token), and the adapter reads them from the environment:
+
+```bash
+PHOTON_PROJECT_ID=<spectrumProjectId>   # the SDK's projectId
+PHOTON_PROJECT_SECRET=<projectSecret>
+```
+
+Management metadata lives in `~/.hermes/auth.json` under `credential_pool`:
 
 ```jsonc
 {
   "credential_pool": {
     "photon": [
-      { "access_token": "<dashboard-bearer>", "issued_at": ... }
+      { "access_token": "<device-bearer>", "issued_at": ... }
     ],
     "photon_project": [
-      { "project_id": "...", "project_secret": "...", "name": "Hermes Agent" }
+      {
+        "dashboard_project_id": "<dashboard id>",
+        "spectrum_project_id": "<spectrumProjectId>",
+        "project_secret": "<projectSecret>",
+        "name": "Hermes Agent"
+      }
     ]
   }
 }
 ```
 
-The per-URL webhook signing secret is treated like an API key and
-lives in `~/.hermes/.env` as `PHOTON_WEBHOOK_SECRET`.
+> **Note on ids.** A Photon project has two identifiers: the dashboard `id`
+> (used for management API calls) and the `spectrumProjectId` (what the SDK
+> authenticates with). `PHOTON_PROJECT_ID` is the **spectrum** id.
 
 ## Configuration knobs
 
-All env vars are documented in `plugin.yaml`. The most important are:
+All env vars are documented in `plugin.yaml`. The most important:
 
-| Env var                  | Default            | Meaning                                 |
-|--------------------------|--------------------|-----------------------------------------|
-| `PHOTON_PROJECT_ID`      | from auth.json     | Spectrum project ID                     |
-| `PHOTON_PROJECT_SECRET`  | from auth.json     | Spectrum project secret (HTTP Basic)    |
-| `PHOTON_WEBHOOK_SECRET`  | (unset)            | Signing secret returned at register     |
-| `PHOTON_WEBHOOK_PORT`    | 8788               | Local port for the aiohttp listener     |
-| `PHOTON_WEBHOOK_PATH`    | /photon/webhook    | Path under which the listener mounts    |
-| `PHOTON_SIDECAR_PORT`    | 8789               | Loopback port for sidecar control      |
-| `PHOTON_HOME_CHANNEL`    | (unset)            | Default space ID for cron delivery     |
-| `PHOTON_ALLOWED_USERS`   | (unset)            | Comma-separated E.164 allowlist        |
+| Env var                   | Default                    | Meaning                              |
+|---------------------------|----------------------------|--------------------------------------|
+| `PHOTON_PROJECT_ID`       | from .env / auth.json      | Spectrum project id (SDK `projectId`)|
+| `PHOTON_PROJECT_SECRET`   | from .env / auth.json      | Project secret                       |
+| `PHOTON_SIDECAR_PORT`     | 8789                       | Loopback port for the sidecar        |
+| `PHOTON_SIDECAR_AUTOSTART`| true                       | Spawn the sidecar on connect         |
+| `PHOTON_DASHBOARD_HOST`   | https://app.photon.codes   | Dashboard API host                   |
+| `PHOTON_HOME_CHANNEL`     | your number (set by setup) | Default space for cron delivery — a space id, or a bare E.164 number (resolved to a DM) |
+| `PHOTON_ALLOWED_USERS`    | your number (set by setup) | Comma-separated E.164 allowlist      |
+| `PHOTON_REQUIRE_MENTION`  | false                      | Gate group chats on a wake word      |
+| `PHOTON_MAX_INLINE_ATTACHMENT_BYTES` | 20 MB           | Max inbound attachment size the sidecar reads & inlines |
 
-## Limitations (current Photon API)
+## Attachments & limitations
 
-- **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 supported.** Images, voice notes, video,
-  and documents are sent via `space.send(attachment(...))` /
+- **Inbound attachments are downloaded.** The sidecar reads the bytes
+  (`content.read()`) and base64-inlines them on the NDJSON event; the adapter
+  caches them to the shared media cache and populates `media_urls` /
+  `media_types`, so the agent sees the real image/file (vision included) —
+  parity with the BlueBubbles iMessage channel. Attachments larger than
+  `PHOTON_MAX_INLINE_ATTACHMENT_BYTES` (default 20 MB), or any byte read that
+  fails, fall back to a text marker (`[Photon attachment received: …]`) so the
+  agent still knows something arrived.
+- **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.
+  endpoint; a caption is delivered as a separate text bubble after the media.
+- **Reactions, message effects, polls** — supported by `spectrum-ts` but not
+  yet exposed; the sidecar is the natural place to add them.
 
 [photon]: https://photon.codes/

plugins/platforms/photon/adapter.py

@@ -1,32 +1,30 @@
 """
 Photon Spectrum (iMessage) platform adapter for Hermes Agent.
 
+Both directions of traffic flow through a small supervised Node sidecar
+(see ``sidecar/index.mjs``) that runs the ``spectrum-ts`` SDK — the SDK is
+TypeScript-only and there is no public HTTP message API, so a sidecar is
+unavoidable.
+
 Inbound:
-    Photon delivers signed JSON ``POST``s to a URL we register.  The
-    adapter spins up an aiohttp server on ``PHOTON_WEBHOOK_PORT``,
-    verifies ``X-Spectrum-Signature`` (HMAC-SHA256 of
-    ``v0:{timestamp}:{body}`` keyed by the per-URL signing secret),
-    rejects deliveries with a timestamp drift > 5 minutes, dedupes on
-    ``message.id``, and dispatches a normalized ``MessageEvent`` to the
-    gateway runner via ``BasePlatformAdapter.handle_message``.
+    The SDK's ``app.messages`` is a long-lived **gRPC** stream. The sidecar
+    serializes each message to a normalized JSON event and streams it to this
+    adapter over a loopback ``GET /inbound`` (NDJSON). A background task here
+    consumes that stream, dedupes on ``messageId``, and dispatches a
+    ``MessageEvent`` to the gateway via ``BasePlatformAdapter.handle_message``.
+    No webhook, no public URL, no signing secret.
 
 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`` / 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.
+    ``send`` / ``send_typing`` are loopback POSTs to the sidecar's control
+    endpoints, authenticated with a shared bearer token.  Outbound media
+    (images, voice notes, video, documents) goes through spectrum-ts'
+    ``attachment()`` / ``voice()`` content builders via the sidecar's
+    ``/send-attachment`` endpoint.
 """
 from __future__ import annotations
 
 import asyncio
-import hashlib
-import hmac
+import base64
 import json
 import logging
 import os
@@ -39,21 +37,21 @@ import sys
 import time
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
-try:
+if TYPE_CHECKING:
+    # Type checkers see ``httpx`` as the always-imported module, so every use
+    # site type-checks cleanly. The runtime fallback below keeps the optional
+    # dependency truly optional (each use site is guarded by HTTPX_AVAILABLE).
     import httpx
     HTTPX_AVAILABLE = True
-except ImportError:  # pragma: no cover - httpx is already a Hermes dep
-    HTTPX_AVAILABLE = False
-    httpx = None  # type: ignore[assignment]
-
-try:
-    from aiohttp import web
-    AIOHTTP_AVAILABLE = True
-except ImportError:
-    AIOHTTP_AVAILABLE = False
-    web = None  # type: ignore[assignment]
+else:
+    try:
+        import httpx
+        HTTPX_AVAILABLE = True
+    except ImportError:  # pragma: no cover - httpx is already a Hermes dep
+        HTTPX_AVAILABLE = False
+        httpx = None
 
 from gateway.config import Platform, PlatformConfig
 from gateway.platforms.base import (
@@ -63,21 +61,13 @@ from gateway.platforms.base import (
     SendResult,
 )
 
-from .auth import (
-    DEFAULT_SPECTRUM_HOST,
-    load_project_credentials,
-    _spectrum_host,
-)
+from .auth import load_project_credentials
 
 logger = logging.getLogger(__name__)
 
 # ---------------------------------------------------------------------------
 # Constants
 
-_DEFAULT_WEBHOOK_PORT = 8788
-_DEFAULT_WEBHOOK_PATH = "/photon/webhook"
-_DEFAULT_WEBHOOK_BIND = "0.0.0.0"
-
 _DEFAULT_SIDECAR_PORT = 8789
 _DEFAULT_SIDECAR_BIND = "127.0.0.1"
 
@@ -86,11 +76,8 @@ _DEFAULT_SIDECAR_BIND = "127.0.0.1"
 # size to ~16 KB.  Keep a conservative cap that matches BlueBubbles.
 _MAX_MESSAGE_LENGTH = 8000
 
-# Spec says reject deliveries older than ~5 minutes for replay protection.
-_TIMESTAMP_DRIFT_SECONDS = 300
-
-# Dedup parameters — keep at least 1k IDs for ~48h per Photon's
-# at-least-once guidance.
+# Dedup parameters — the gRPC stream is at-least-once, and a sidecar
+# reconnect can replay, so keep at least 1k ids for ~48h.
 _DEDUP_MAX_SIZE = 4000
 _DEDUP_WINDOW_SECONDS = 48 * 3600
 
@@ -118,7 +105,7 @@ def _coerce_port(value: Any, default: int) -> int:
 
 def check_requirements() -> bool:
     """Return True when both Python deps and the Node sidecar are available."""
-    if not HTTPX_AVAILABLE or not AIOHTTP_AVAILABLE:
+    if not HTTPX_AVAILABLE:
         return False
     if not shutil.which(os.getenv("PHOTON_NODE_BIN") or "node"):
         return False
@@ -146,61 +133,33 @@ def is_connected(cfg: PlatformConfig) -> bool:
 
 
 def _env_enablement() -> Optional[dict]:
-    """Seed PlatformConfig.extra from env so env-only setups appear in status."""
+    """Seed PlatformConfig.extra from env so env-only setups appear in status.
+
+    The special ``home_channel`` key is handled by the core plugin hook and
+    becomes a proper ``HomeChannel`` on ``PlatformConfig``.
+    """
     project_id, project_secret = load_project_credentials()
     if not (project_id and project_secret):
         return None
-    return {
-        "project_id": project_id,
-        "project_secret": project_secret,
-        "webhook_port": _coerce_port(os.getenv("PHOTON_WEBHOOK_PORT"), _DEFAULT_WEBHOOK_PORT),
-        "webhook_path": os.getenv("PHOTON_WEBHOOK_PATH") or _DEFAULT_WEBHOOK_PATH,
-    }
-
-
-# ---------------------------------------------------------------------------
-# Signature verification
-
-def verify_signature(
-    *,
-    body: bytes,
-    timestamp_header: str,
-    signature_header: str,
-    signing_secret: str,
-    now: Optional[float] = None,
-    drift: int = _TIMESTAMP_DRIFT_SECONDS,
-) -> bool:
-    """Constant-time verify a Photon webhook signature.
-
-    Returns True iff the timestamp is within ``drift`` of *now* AND
-    ``signature_header == "v0=" + hmac_sha256(secret, "v0:{ts}:{body}")``.
-
-    Exposed at module scope so tests can exercise it without an adapter
-    instance.
-    """
-    if not timestamp_header or not signature_header or not signing_secret:
-        return False
-    try:
-        ts = int(timestamp_header)
-    except ValueError:
-        return False
-    if abs((now or time.time()) - ts) > drift:
-        return False
-    if not signature_header.startswith("v0="):
-        return False
-    expected = hmac.new(
-        signing_secret.encode("utf-8"),
-        f"v0:{ts}:".encode("utf-8") + body,
-        hashlib.sha256,
-    ).hexdigest()
-    return hmac.compare_digest(expected, signature_header[3:])
+    seed = {"project_id": project_id, "project_secret": project_secret}
+    home = os.getenv("PHOTON_HOME_CHANNEL", "").strip()
+    if home:
+        seed["home_channel"] = {
+            "chat_id": home,
+            "name": os.getenv("PHOTON_HOME_CHANNEL_NAME", "Home"),
+        }
+    return seed
 
 
 # ---------------------------------------------------------------------------
 # Adapter
 
 class PhotonAdapter(BasePlatformAdapter):
-    """Inbound: signed webhook on aiohttp. Outbound: Node sidecar via loopback HTTP."""
+    """Bidirectional bridge to Photon Spectrum via the Node spectrum-ts sidecar.
+
+    Inbound: consume the sidecar's ``/inbound`` gRPC stream.
+    Outbound: loopback POSTs to the sidecar's control channel.
+    """
 
     MAX_MESSAGE_LENGTH = _MAX_MESSAGE_LENGTH
 
@@ -209,6 +168,8 @@ class PhotonAdapter(BasePlatformAdapter):
         extra = config.extra or {}
 
         # Project credentials (env wins, then config.extra, then auth.json).
+        # ``project_id`` here is the project's spectrumProjectId — the value
+        # the spectrum-ts SDK authenticates with.
         stored_id, stored_sec = load_project_credentials()
         self._project_id: str = (
             os.getenv("PHOTON_PROJECT_ID")
@@ -223,27 +184,6 @@ class PhotonAdapter(BasePlatformAdapter):
             or ""
         )
 
-        # Webhook receiver
-        self._webhook_port = _coerce_port(
-            extra.get("webhook_port") or os.getenv("PHOTON_WEBHOOK_PORT"),
-            _DEFAULT_WEBHOOK_PORT,
-        )
-        self._webhook_path = (
-            extra.get("webhook_path")
-            or os.getenv("PHOTON_WEBHOOK_PATH")
-            or _DEFAULT_WEBHOOK_PATH
-        )
-        self._webhook_bind = (
-            extra.get("webhook_bind")
-            or os.getenv("PHOTON_WEBHOOK_BIND")
-            or _DEFAULT_WEBHOOK_BIND
-        )
-        self._webhook_secret: str = (
-            os.getenv("PHOTON_WEBHOOK_SECRET")
-            or extra.get("webhook_secret")
-            or ""
-        )
-
         # Sidecar
         self._sidecar_port = _coerce_port(
             extra.get("sidecar_port") or os.getenv("PHOTON_SIDECAR_PORT"),
@@ -259,12 +199,13 @@ class PhotonAdapter(BasePlatformAdapter):
         self._node_bin = os.getenv("PHOTON_NODE_BIN") or shutil.which("node") or "node"
 
         # Runtime state
-        self._runner: Optional["web.AppRunner"] = None
         self._sidecar_proc: Optional[subprocess.Popen] = None
         self._sidecar_supervisor_task: Optional[asyncio.Task] = None
+        self._inbound_task: Optional[asyncio.Task] = None
+        self._inbound_running = False
         self._http_client: Optional["httpx.AsyncClient"] = None
-        # Lightweight in-memory dedup. Photon's at-least-once guarantee
-        # means we WILL see the same message.id more than once.
+        # Lightweight in-memory dedup. The gRPC stream is at-least-once, so we
+        # may see the same messageId more than once (e.g. after a reconnect).
         self._seen_messages: Dict[str, float] = {}
 
         # Group-chat mention gating (parity with BlueBubbles). When enabled,
@@ -345,13 +286,6 @@ class PhotonAdapter(BasePlatformAdapter):
     # -- Connection lifecycle ---------------------------------------------
 
     async def connect(self) -> bool:
-        if not AIOHTTP_AVAILABLE:
-            self._set_fatal_error(
-                "MISSING_DEP",
-                "aiohttp not installed. Run: pip install aiohttp",
-                retryable=False,
-            )
-            return False
         if not HTTPX_AVAILABLE:
             self._set_fatal_error(
                 "MISSING_DEP", "httpx not installed", retryable=False
@@ -366,19 +300,11 @@ class PhotonAdapter(BasePlatformAdapter):
             )
             return False
 
-        # Start the aiohttp receiver first; without it the sidecar would
-        # be able to forward inbound traffic to a closed port.
-        try:
-            await self._start_webhook_server()
-        except OSError as e:
-            self._set_fatal_error(
-                "PORT_IN_USE",
-                f"webhook port {self._webhook_port} unavailable: {e}",
-                retryable=True,
-            )
-            return False
+        client = httpx.AsyncClient(timeout=30.0)
+        self._http_client = client
 
-        # Spin up the Node sidecar (required for outbound).
+        # The sidecar holds the gRPC stream for BOTH directions, so it is
+        # required now (not just for outbound).
         if self._autostart_sidecar:
             try:
                 await self._start_sidecar()
@@ -388,23 +314,39 @@ class PhotonAdapter(BasePlatformAdapter):
                     f"failed to start Photon sidecar: {e}",
                     retryable=True,
                 )
-                await self._stop_webhook_server()
+                await client.aclose()
+                self._http_client = None
                 return False
         else:
-            logger.info("[photon] sidecar autostart disabled — outbound will fail")
+            logger.warning(
+                "[photon] sidecar autostart disabled — inbound + outbound will fail"
+            )
+
+        # Start consuming the inbound gRPC stream from the sidecar.
+        self._inbound_running = True
+        self._inbound_task = asyncio.get_event_loop().create_task(
+            self._inbound_loop()
+        )
 
-        self._http_client = httpx.AsyncClient(timeout=30.0)
         self._mark_connected()
         logger.info(
-            "[photon] connected — webhook at %s:%d%s, sidecar on %s:%d",
-            self._webhook_bind, self._webhook_port, self._webhook_path,
+            "[photon] connected — sidecar on %s:%d, streaming inbound over gRPC",
             self._sidecar_bind, self._sidecar_port,
         )
         return True
 
     async def disconnect(self) -> None:
+        self._inbound_running = False
+        if self._inbound_task is not None:
+            self._inbound_task.cancel()
+            try:
+                await self._inbound_task
+            except asyncio.CancelledError:
+                pass
+            except Exception:
+                pass
+            self._inbound_task = None
         await self._stop_sidecar()
-        await self._stop_webhook_server()
         if self._http_client is not None:
             try:
                 await self._http_client.aclose()
@@ -413,68 +355,61 @@ class PhotonAdapter(BasePlatformAdapter):
             self._http_client = None
         self._mark_disconnected()
 
-    # -- Webhook server ----------------------------------------------------
+    # -- Inbound stream consumer ------------------------------------------
 
-    async def _start_webhook_server(self) -> None:
-        app = web.Application()
-        app.router.add_post(self._webhook_path, self._handle_webhook)
-        app.router.add_get("/healthz", lambda _: web.Response(text="ok"))
-        self._runner = web.AppRunner(app)
-        await self._runner.setup()
-        site = web.TCPSite(self._runner, self._webhook_bind, self._webhook_port)
-        await site.start()
+    async def _inbound_loop(self) -> None:
+        """Consume the sidecar's ``/inbound`` NDJSON stream, with reconnect.
 
-    async def _stop_webhook_server(self) -> None:
-        if self._runner is not None:
+        The sidecar owns the gRPC reconnect/heartbeat to Photon; this loop
+        only has to re-open the loopback HTTP stream if it drops (e.g. the
+        sidecar restarts).
+        """
+        client = self._http_client
+        if client is None:
+            return
+        url = f"http://{self._sidecar_bind}:{self._sidecar_port}/inbound"
+        headers = {"X-Hermes-Sidecar-Token": self._sidecar_token}
+        backoff = 1.0
+        while self._inbound_running:
             try:
-                await self._runner.cleanup()
-            except Exception:
-                pass
-            self._runner = None
-
-    async def _handle_webhook(self, request: "web.Request") -> "web.Response":
-        body = await request.read()
-        if self._webhook_secret:
-            ts = request.headers.get("X-Spectrum-Timestamp", "")
-            sig = request.headers.get("X-Spectrum-Signature", "")
-            if not verify_signature(
-                body=body,
-                timestamp_header=ts,
-                signature_header=sig,
-                signing_secret=self._webhook_secret,
-            ):
-                logger.warning("[photon] rejected webhook with bad signature")
-                return web.Response(status=401, text="invalid signature")
-        else:
-            logger.warning(
-                "[photon] PHOTON_WEBHOOK_SECRET unset — accepting unsigned "
-                "deliveries. Set the per-URL signing secret returned by "
-                "register-webhook to enable verification."
-            )
+                async with client.stream(
+                    "GET", url, headers=headers, timeout=None,
+                ) as resp:
+                    if resp.status_code != 200:
+                        raise RuntimeError(f"/inbound returned {resp.status_code}")
+                    backoff = 1.0  # reset on a successful connect
+                    async for line in resp.aiter_lines():
+                        if not self._inbound_running:
+                            break
+                        line = line.strip()
+                        if not line:
+                            continue  # heartbeat
+                        await self._on_inbound_line(line)
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                if not self._inbound_running:
+                    break
+                logger.warning(
+                    "[photon] inbound stream dropped (%s); reconnecting in %.1fs",
+                    e, backoff,
+                )
+                await asyncio.sleep(backoff)
+                backoff = min(backoff * 2, 30.0)
 
+    async def _on_inbound_line(self, line: str) -> None:
         try:
-            payload = json.loads(body or b"{}")
+            event = json.loads(line)
         except json.JSONDecodeError:
-            return web.Response(status=400, text="invalid json")
-        if payload.get("event") != "messages":
-            # Photon currently emits only `messages`; any future event
-            # types are ack'd 200 so they don't retry.
-            return web.Response(text="ok")
-
-        msg = payload.get("message") or {}
-        msg_id = msg.get("id")
-        if not msg_id:
-            return web.Response(status=400, text="missing message.id")
-        if self._is_duplicate(msg_id):
-            return web.Response(text="ok (dup)")
-
+            logger.debug("[photon] skipping non-JSON inbound line")
+            return
+        msg_id = event.get("messageId")
+        if msg_id and self._is_duplicate(msg_id):
+            return
         try:
-            await self._dispatch_inbound(payload)
+            await self._dispatch_inbound(event)
         except Exception:
             logger.exception("[photon] inbound dispatch failed")
-            # 200 anyway — we own the dedup; failing here would cause
-            # Photon to retry the same id.
-        return web.Response(text="ok")
 
     def _is_duplicate(self, msg_id: str) -> bool:
         now = time.time()
@@ -488,44 +423,77 @@ class PhotonAdapter(BasePlatformAdapter):
         self._seen_messages[msg_id] = now
         return False
 
-    async def _dispatch_inbound(self, payload: Dict[str, Any]) -> None:
-        msg = payload.get("message") or {}
-        space = msg.get("space") or payload.get("space") or {}
-        sender = msg.get("sender") or {}
-        content = msg.get("content") or {}
+    async def _dispatch_inbound(self, event: Dict[str, Any]) -> None:
+        """Normalize a sidecar inbound event and dispatch it to the gateway.
+
+        Event shape (from ``sidecar/index.mjs``)::
+
+            {
+              "messageId": "...",
+              "platform": "iMessage",
+              "space": {"id": "...", "type": "dm"|"group", "phone": "+E164"},
+              "sender": {"id": "+E164"},
+              "content": {"type": "text", "text": "..."}
+                       | {"type": "attachment", "id", "name", "mimeType",
+                          "size", "data"?, "encoding"?},
+              "timestamp": "2026-05-14T19:06:32.000Z"
+
+        Attachment content carries the bytes inline as base64 ``data`` (with
+        ``encoding == "base64"``) when the sidecar could read them within its
+        size cap; otherwise only metadata is present and we surface a marker.
+            }
+        """
+        space = event.get("space") or {}
+        sender = event.get("sender") or {}
+        content = event.get("content") or {}
 
         space_id = space.get("id") or ""
-        sender_id = sender.get("id") or ""
         if not space_id:
             logger.warning("[photon] inbound missing space.id")
             return
 
-        # Space type — Photon documents iMessage DM ids as `any;-;+E164`
-        # and group ids as `any;+;<chat-guid>`.  Use that as the
-        # heuristic; everything else is treated as DM.
-        chat_type = "group" if ";+;" in space_id else "dm"
+        # iMessage spaces carry their type directly — no id string-sniffing.
+        chat_type = "group" if space.get("type") == "group" else "dm"
+        sender_id = sender.get("id") or space.get("phone") or space_id
 
-        # Timestamp — ISO 8601 from the platform.
-        ts_str = msg.get("timestamp") or ""
+        ts_str = event.get("timestamp") or ""
         try:
-            timestamp = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
+            timestamp = (
+                datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
+                if ts_str
+                else datetime.now(tz=timezone.utc)
+            )
         except ValueError:
             timestamp = datetime.now(tz=timezone.utc)
 
-        # Content normalization.  Spectrum is a discriminated union;
-        # text vs attachment metadata.  Attachments are metadata-only
-        # today (no download URL) — log + carry the name so the agent
-        # at least knows something was sent.
-        if content.get("type") == "text":
+        # Media attachments (local cached paths) handed to the agent via the
+        # gateway's image-routing path, exactly like the BlueBubbles channel.
+        media_urls: List[str] = []
+        media_types: List[str] = []
+
+        ctype = content.get("type")
+        if ctype == "text":
             text = content.get("text") or ""
             mtype = MessageType.TEXT
-        elif content.get("type") == "attachment":
+        elif ctype == "attachment":
             name = content.get("name") or "(unnamed)"
             mime = content.get("mimeType") or ""
-            text = f"[Photon attachment received: {name} ({mime}) — no download URL yet]"
             mtype = _attachment_message_type(mime)
+            cached = _cache_inbound_attachment(content, name, mime)
+            if cached:
+                media_urls.append(cached)
+                media_types.append(mime or "application/octet-stream")
+                # The real bytes are attached, so the agent sees the media
+                # itself — a short marker is enough text, and it keeps group
+                # mention-gating consistent with plain messages.
+                text = "(attachment)"
+            else:
+                # No bytes (over the sidecar cap, a failed read, or a caching
+                # failure) — fall back to a metadata marker so the agent still
+                # knows something arrived.
+                text = f"[Photon attachment received: {name} ({mime})]"
         else:
-            text = f"[Photon content type not handled: {content.get('type')}]"
+            text = f"[Photon content type not handled: {ctype}]"
             mtype = MessageType.TEXT
 
         # Group-mention gating (parity with BlueBubbles). In group chats with
@@ -545,18 +513,20 @@ class PhotonAdapter(BasePlatformAdapter):
             chat_id=space_id,
             chat_name=space_id,
             chat_type=chat_type,
-            user_id=sender_id or space_id,
+            user_id=sender_id,
             user_name=sender_id or None,
         )
-        event = MessageEvent(
+        message_event = MessageEvent(
             text=text,
             message_type=mtype,
             source=source,
-            message_id=msg.get("id"),
-            raw_message=payload,
+            message_id=event.get("messageId"),
+            raw_message=event,
             timestamp=timestamp,
+            media_urls=media_urls,
+            media_types=media_types,
         )
-        await self.handle_message(event)
+        await self.handle_message(message_event)
 
     # -- Sidecar lifecycle -------------------------------------------------
 
@@ -670,7 +640,7 @@ class PhotonAdapter(BasePlatformAdapter):
         reply_to: Optional[str] = None,
         metadata: Optional[Dict[str, Any]] = None,
     ) -> SendResult:
-        return await self._sidecar_send(chat_id, content, reply_to=reply_to)
+        return await self._sidecar_send(chat_id, content)
 
     # -- Outbound media (parity with the BlueBubbles iMessage channel) -----
     #
@@ -696,7 +666,7 @@ class PhotonAdapter(BasePlatformAdapter):
             # 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,
+            chat_id, local_path, caption=caption,
         )
 
     async def send_image_file(
@@ -709,7 +679,7 @@ class PhotonAdapter(BasePlatformAdapter):
         **kwargs,
     ) -> SendResult:
         return await self._sidecar_send_attachment(
-            chat_id, image_path, caption=caption, reply_to=reply_to,
+            chat_id, image_path, caption=caption,
         )
 
     async def send_voice(
@@ -722,7 +692,7 @@ class PhotonAdapter(BasePlatformAdapter):
         **kwargs,
     ) -> SendResult:
         return await self._sidecar_send_attachment(
-            chat_id, audio_path, caption=caption, reply_to=reply_to, kind="voice",
+            chat_id, audio_path, caption=caption, kind="voice",
         )
 
     async def send_video(
@@ -735,7 +705,7 @@ class PhotonAdapter(BasePlatformAdapter):
         **kwargs,
     ) -> SendResult:
         return await self._sidecar_send_attachment(
-            chat_id, video_path, caption=caption, reply_to=reply_to,
+            chat_id, video_path, caption=caption,
         )
 
     async def send_document(
@@ -749,7 +719,7 @@ class PhotonAdapter(BasePlatformAdapter):
         **kwargs,
     ) -> SendResult:
         return await self._sidecar_send_attachment(
-            chat_id, file_path, name=file_name, caption=caption, reply_to=reply_to,
+            chat_id, file_path, name=file_name, caption=caption,
         )
 
     async def send_animation(
@@ -767,23 +737,29 @@ class PhotonAdapter(BasePlatformAdapter):
 
     async def send_typing(self, chat_id: str, metadata=None) -> None:
         try:
-            await self._sidecar_call("/typing", {"spaceId": chat_id})
+            await self._sidecar_call(
+                "/typing", {"spaceId": chat_id, "state": "start"}
+            )
         except Exception as e:
             logger.debug("[photon] send_typing failed: %s", e)
 
+    async def stop_typing(self, chat_id: str) -> None:
+        try:
+            await self._sidecar_call(
+                "/typing", {"spaceId": chat_id, "state": "stop"}
+            )
+        except Exception as e:
+            logger.debug("[photon] stop_typing failed: %s", e)
+
     async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
         """Return whatever we know about a Spectrum space id.
 
-        Photon's `space.id` is opaque (`any;-;+E164` for DMs,
-        `any;+;<guid>` for groups). We surface that shape directly so
-        the gateway has something to show in session pickers / logs.
+        Photon's ``space.id`` is opaque; the inbound event also carries the
+        DM/group type, but here we only have the id, so infer conservatively.
         """
-        chat_type = "group" if ";+;" in chat_id else "dm"
-        return {"name": chat_id, "type": chat_type, "id": chat_id}
+        return {"name": chat_id, "type": "dm", "id": chat_id}
 
-    async def _sidecar_send(
-        self, space_id: str, text: str, *, reply_to: Optional[str] = None,
-    ) -> SendResult:
+    async def _sidecar_send(self, space_id: str, text: str) -> SendResult:
         if len(text) > self.MAX_MESSAGE_LENGTH:
             logger.warning(
                 "[photon] truncating outbound from %d to %d chars",
@@ -791,8 +767,6 @@ class PhotonAdapter(BasePlatformAdapter):
             )
             text = text[: self.MAX_MESSAGE_LENGTH]
         body: Dict[str, Any] = {"spaceId": space_id, "text": text}
-        if reply_to:
-            body["replyTo"] = reply_to
         try:
             data = await self._sidecar_call("/send", body)
         except Exception as e:
@@ -807,7 +781,6 @@ class PhotonAdapter(BasePlatformAdapter):
         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.
@@ -842,8 +815,6 @@ class PhotonAdapter(BasePlatformAdapter):
             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:
@@ -887,11 +858,81 @@ def _attachment_message_type(mime: str) -> MessageType:
     return MessageType.DOCUMENT
 
 
+# MIME → file-extension maps for caching inbound attachment bytes. These mirror
+# the BlueBubbles iMessage channel so both adapters name cached media the same.
+_IMAGE_EXT_BY_MIME = {
+    "image/jpeg": ".jpg",
+    "image/png": ".png",
+    "image/gif": ".gif",
+    "image/webp": ".webp",
+    "image/heic": ".jpg",
+    "image/heif": ".jpg",
+    "image/tiff": ".jpg",
+}
+_AUDIO_EXT_BY_MIME = {
+    "audio/mp3": ".mp3",
+    "audio/mpeg": ".mp3",
+    "audio/ogg": ".ogg",
+    "audio/wav": ".wav",
+    "audio/x-caf": ".mp3",
+    "audio/mp4": ".m4a",
+    "audio/aac": ".m4a",
+}
+
+
+def _cache_inbound_attachment(
+    content: Dict[str, Any], name: str, mime: str
+) -> Optional[str]:
+    """Decode a base64-inlined inbound attachment and cache it locally.
+
+    The sidecar inlines the attachment bytes as ``content["data"]`` (base64).
+    We decode them and route to the shared media cache by MIME type, returning
+    the cached absolute path so the caller can populate ``media_urls`` (which
+    the gateway then hands to the model). Returns ``None`` when there are no
+    bytes (over the sidecar's inline cap or a failed read) or when caching
+    fails, so the caller can fall back to a text marker.
+    """
+    data_b64 = content.get("data")
+    if not data_b64:
+        return None
+    try:
+        raw = base64.b64decode(data_b64)
+    except (ValueError, TypeError) as exc:
+        logger.warning("[photon] failed to decode inbound attachment bytes: %s", exc)
+        return None
+
+    from gateway.platforms.base import (
+        cache_audio_from_bytes,
+        cache_document_from_bytes,
+        cache_image_from_bytes,
+    )
+
+    mime = (mime or "").lower()
+    # Prefer the real extension from the filename; fall back to the MIME map.
+    suffix = Path(name).suffix if name else ""
+    try:
+        if mime.startswith("image/"):
+            ext = suffix or _IMAGE_EXT_BY_MIME.get(mime, ".jpg")
+            try:
+                return cache_image_from_bytes(raw, ext)
+            except ValueError:
+                # Bytes don't look like a supported image (e.g. HEIC magic) —
+                # still deliver them as a document rather than dropping them.
+                return cache_document_from_bytes(raw, name)
+        if mime.startswith("audio/"):
+            ext = suffix or _AUDIO_EXT_BY_MIME.get(mime, ".mp3")
+            return cache_audio_from_bytes(raw, ext)
+        # Video, application/*, and everything else → document cache.
+        return cache_document_from_bytes(raw, name)
+    except Exception as exc:
+        logger.warning("[photon] failed to cache inbound attachment %s: %s", name, exc)
+        return None
+
+
 # ---------------------------------------------------------------------------
 # Standalone (out-of-process) send for cron deliveries when the gateway
-# is not co-resident.  Spins up an ephemeral sidecar call by spawning
-# the existing sidecar binary one-shot; if a live sidecar is already
-# listening on the configured port we reuse it.
+# is not co-resident.  Reuses a live sidecar already listening on the
+# configured port (cron processes cannot spawn the sidecar themselves).
 
 async def _standalone_send(
     pconfig: PlatformConfig,
@@ -980,7 +1021,7 @@ def register(ctx) -> None:
 
     ctx.register_platform(
         name="photon",
-        label="Photon iMessage",
+        label="iMessage via Photon",
         adapter_factory=lambda cfg: PhotonAdapter(cfg),
         check_fn=check_requirements,
         validate_config=validate_config,
@@ -1011,7 +1052,7 @@ def register(ctx) -> None:
             "Treat replies like regular text messages — short, friendly, no "
             "markdown rendering. Recipient identifiers are E.164 phone "
             "numbers; never expose them in responses unless the user asked. "
-            "Attachments arrive as metadata only (no download URL yet)."
+            "Attachments arrive as metadata only."
         ),
     )
 

plugins/platforms/photon/cli.py

@@ -7,25 +7,26 @@ Subcommands:
     setup              full first-time setup (device login + project + user + sidecar)
     status             show login + project + sidecar dep state
     install-sidecar    npm install inside plugins/platforms/photon/sidecar/
-    webhook register   register the local webhook URL with Photon
-    webhook list       list registered webhooks
-    webhook delete     delete a webhook by id
 
 The device-code login runs automatically as the first step of ``setup``;
 there is no standalone ``login`` verb (matching how every other Hermes
 gateway channel onboards through a single setup surface).
+
+Photon uses the spectrum-ts gRPC stream for inbound — there is no webhook
+to register, so there are no webhook subcommands.
 """
 from __future__ import annotations
 
 import argparse
 import getpass
-import json
 import os
 import shutil
 import subprocess
 import sys
 from pathlib import Path
 
+from hermes_cli.colors import Colors, color
+
 from . import auth as photon_auth
 
 _SIDECAR_DIR = Path(__file__).parent / "sidecar"
@@ -38,9 +39,14 @@ def register_cli(parser: argparse.ArgumentParser) -> None:
     """Wire up `hermes photon ...` subcommands."""
     subs = parser.add_subparsers(dest="photon_command", required=False)
 
-    p_setup = subs.add_parser("setup", help="First-time setup (device login + project + user + sidecar)")
-    p_setup.add_argument("--project-name", default=None, help="Project name (default: 'Hermes Agent')")
-    p_setup.add_argument("--phone", default=None, help="Your E.164 phone number (e.g. +15551234567)")
+    p_setup = subs.add_parser(
+        "setup",
+        help="First-time setup (device login + project + user + sidecar)",
+    )
+    p_setup.add_argument("--project-name", default=None,
+                         help="Project name (default: 'Hermes Agent')")
+    p_setup.add_argument("--phone", default=None,
+                         help="Your E.164 phone number (e.g. +15551234567)")
     p_setup.add_argument("--first-name", default=None)
     p_setup.add_argument("--last-name", default=None)
     p_setup.add_argument("--email", default=None)
@@ -52,14 +58,6 @@ def register_cli(parser: argparse.ArgumentParser) -> None:
     subs.add_parser("status", help="Show login + project + sidecar dep state")
     subs.add_parser("install-sidecar", help="Run npm install inside the sidecar directory")
 
-    p_hook = subs.add_parser("webhook", help="Manage Photon webhook registrations")
-    hook_subs = p_hook.add_subparsers(dest="photon_webhook_command", required=True)
-    p_hook_reg = hook_subs.add_parser("register", help="Register a webhook URL")
-    p_hook_reg.add_argument("url", help="Publicly reachable URL Photon should POST to")
-    hook_subs.add_parser("list", help="List registered webhooks for the current project")
-    p_hook_del = hook_subs.add_parser("delete", help="Delete a webhook by id")
-    p_hook_del.add_argument("webhook_id")
-
     parser.set_defaults(func=dispatch)
 
 
@@ -77,8 +75,6 @@ def dispatch(args: argparse.Namespace) -> int:
         return _cmd_status(args)
     if sub == "install-sidecar":
         return _cmd_install_sidecar(args)
-    if sub == "webhook":
-        return _cmd_webhook(args)
     print(f"unknown subcommand: {sub}", file=sys.stderr)
     return 2
 
@@ -122,7 +118,7 @@ def _cmd_setup(args: argparse.Namespace) -> int:
     # 1. Login (skip if we already have a token).
     token = photon_auth.load_photon_token()
     if not token:
-        print("[1/4] No Photon token found — running device login...")
+        print("[1/5] No Photon token found — running device login...")
         rc = _run_device_login(args)
         if rc != 0:
             return rc
@@ -131,85 +127,163 @@ def _cmd_setup(args: argparse.Namespace) -> int:
             print("login completed but token was not stored", file=sys.stderr)
             return 1
     else:
-        print("[1/4] Reusing existing Photon token")
+        print("[1/5] Reusing existing Photon token")
 
-    # 2. Create (or surface existing) project.
-    existing_id, existing_secret = photon_auth.load_project_credentials()
-    project_id: str
-    project_secret: str
-    if existing_id and existing_secret:
-        project_id, project_secret = existing_id, existing_secret
-        # `project_id` is a Photon-assigned UUID, not a secret — but we
-        # keep the print terse to avoid CodeQL flow noise.
-        print("[2/4] Reusing existing Photon project")
-    else:
-        name = args.project_name or "Hermes Agent"
-        print(f"[2/4] Creating Photon project '{name}' (spectrum=true, imessage)...")
-        try:
-            data = photon_auth.create_project(token, name=name)
-        except Exception as e:
-            print(f"create-project failed: {e}", file=sys.stderr)
-            return 1
-        project_id = data.get("spectrumProjectId") or data.get("id") or ""
-        project_secret = data.get("projectSecret") or ""
-        if not project_id or not project_secret:
-            print(
-                "create-project did not return spectrumProjectId + "
-                "projectSecret. Re-run after enabling Spectrum on the "
-                "project, or open https://app.photon.codes/ to fetch the "
-                "secret manually.",
-                file=sys.stderr,
-            )
-            return 1
-        photon_auth.store_project_credentials(project_id, project_secret, name=name)
-        print("  ✓ project provisioned (run `hermes photon status` to see the id)")
+    # 2. Find or create the "Hermes Agent" project.
+    name = args.project_name or photon_auth.DEFAULT_PROJECT_NAME
+    dashboard_id = photon_auth.load_dashboard_project_id()
+    try:
+        if dashboard_id:
+            print("[2/5] Reusing configured Photon project")
+        else:
+            existing = photon_auth.find_project_by_name(token, name)
+            if existing and existing.get("id"):
+                dashboard_id = existing["id"]
+                print(f"[2/5] Found existing project '{name}'")
+            else:
+                print(f"[2/5] Creating Photon project '{name}'...")
+                created = photon_auth.create_project(token, name=name)
+                dashboard_id = created.get("id")
+                print("  ✓ project created")
+    except Exception as e:
+        print(f"project setup failed: {e}", file=sys.stderr)
+        return 1
+    if not dashboard_id:
+        print("could not resolve a Photon project id", file=sys.stderr)
+        return 1
 
-    # 3. Create a Spectrum user for the operator.
+    # 3. Enable Spectrum, fetch the spectrum project id, rotate the secret,
+    #    and persist both (runtime creds -> ~/.hermes/.env, ids -> auth.json).
+    try:
+        print("[3/5] Enabling Spectrum and provisioning credentials...")
+        proj = photon_auth.ensure_spectrum_enabled(token, dashboard_id)
+        spectrum_id = proj.get("spectrumProjectId")
+        if not spectrum_id:
+            print("spectrum provisioning failed: no spectrum project id", file=sys.stderr)
+            return 1
+        spectrum_id = str(spectrum_id)
+        secret = photon_auth.regenerate_project_secret(token, dashboard_id)
+        photon_auth.store_project_credentials(
+            spectrum_project_id=spectrum_id,
+            project_secret=secret,
+            dashboard_project_id=dashboard_id,
+            name=name,
+        )
+        # spectrum_id is an opaque non-secret id; safe to show.
+        print(f"  ✓ Spectrum enabled (project id {spectrum_id}) — secret saved")
+    except Exception as e:
+        print(f"spectrum provisioning failed: {e}", file=sys.stderr)
+        return 1
+
+    # 4. Register the operator's phone number as a Spectrum user (idempotent).
     phone = args.phone or _prompt(
-        "Your iMessage phone number (E.164, e.g. +15551234567): "
+        color(
+            "[4/5] Your iMessage phone number (E.164, e.g. +15551234567): ",
+            Colors.CYAN,
+        )
     )
+    agent_number = None
     if not phone:
-        print("[3/4] Skipped user creation (no phone given). Re-run with --phone later.")
+        print("      Skipped user registration (no phone given). Re-run with --phone later.")
     else:
-        print("[3/4] Creating shared Spectrum user...")
+        # Name/email are optional and never prompted for — pass --first-name /
+        # --email if you want them sent to the dashboard.
+        first_name = args.first_name
+        email = args.email
         try:
-            photon_auth.create_user(
-                project_id, project_secret,
+            user, created = photon_auth.register_user_if_absent(
+                token, dashboard_id,
                 phone_number=phone,
-                first_name=args.first_name,
+                first_name=first_name,
                 last_name=args.last_name,
-                email=args.email,
+                email=email,
             )
-        except Exception as e:
-            print(f"create-user failed: {e}", file=sys.stderr)
+        except ValueError as e:
+            print(f"      invalid phone number: {e}", file=sys.stderr)
             return 1
-        print("  ✓ user created — check `hermes photon status` or the dashboard for the assigned iMessage line")
+        except Exception as e:
+            print(f"      user registration failed: {e}", file=sys.stderr)
+            return 1
+        print("  ✓ phone registered" if created else "  ✓ phone already registered")
+        # The number to text the agent is the user's assigned iMessage line
+        # (the dashboard's "TEXTS ON" column). On shared-number plans there is
+        # no dedicated entry in /lines, so this per-user field is the source of
+        # truth — and we already have it from the (reused) user object.
+        agent_number = photon_auth.user_assigned_line(user)
+        # Allowlist the operator and make their DM the cron home channel —
+        # otherwise the gateway denies their own inbound messages
+        # ("Unauthorized user") and has no default space for cron delivery.
+        _autoconfigure_access(phone)
 
-    # 4. Sidecar deps.
-    if args.skip_sidecar_install:
-        print("[4/4] Skipping sidecar npm install (--skip-sidecar-install)")
+    # 5. Surface the agent's iMessage number (the number to text the agent).
+    if not agent_number:
+        # No per-user assignment — fall back to a dedicated line if the project
+        # has one provisioned in its line inventory.
+        try:
+            line = photon_auth.get_imessage_line(token, dashboard_id)
+            if line:
+                agent_number = line.get("phoneNumber")
+        except Exception as e:
+            print(f"      (could not fetch the assigned line: {e})", file=sys.stderr)
+    if agent_number:
+        print()
+        print(color("┌─ Your agent's iMessage number ───────────────────────────────", Colors.GREEN))
+        print(
+            color("│  📱 ", Colors.GREEN)
+            + color(str(agent_number), Colors.GREEN, Colors.BOLD)
+        )
+        print(color("│  Text this number from your phone to talk to your agent.", Colors.GREEN))
+        print(color("└──────────────────────────────────────────────────────────────", Colors.GREEN))
     else:
-        print("[4/4] Installing Node sidecar deps (spectrum-ts)...")
+        print("      No iMessage line assigned yet — check the Photon dashboard.")
+
+    # 6. Sidecar deps (spectrum-ts).
+    if args.skip_sidecar_install:
+        print("[5/5] Skipping sidecar npm install (--skip-sidecar-install)")
+    else:
+        print("[5/5] Installing Node sidecar deps (spectrum-ts)...")
         rc = _install_sidecar()
         if rc != 0:
             return rc
 
     print()
     print("✓ Photon setup complete.")
-    print("  Next: register a webhook URL Photon can reach:")
-    print("        hermes photon webhook register https://YOUR-PUBLIC-URL/photon/webhook")
-    print("  Then start the gateway:")
-    print("        hermes gateway start --platform photon")
+    print("  Start the gateway:  hermes gateway start --platform photon")
     return 0
 
 
+def _autoconfigure_access(phone: str) -> None:
+    """Allowlist the operator and set their DM as the cron home channel.
+
+    Writes ``PHOTON_ALLOWED_USERS`` (so the gateway authorizes the operator's
+    own inbound messages instead of denying them) and ``PHOTON_HOME_CHANNEL``
+    (the default space for cron delivery) to the operator's E.164 number. Each
+    is only filled when unset, so a hand-tuned allowlist / home channel is
+    never clobbered on a re-run.
+    """
+    try:
+        from hermes_cli.config import get_env_value, save_env_value
+    except ImportError:
+        return
+    for key, label in (
+        ("PHOTON_ALLOWED_USERS", "allowlisted your number"),
+        ("PHOTON_HOME_CHANNEL", "set your DM as the cron home channel"),
+    ):
+        try:
+            if get_env_value(key):
+                print(f"      {key} already set — leaving it as-is.")
+                continue
+            save_env_value(key, phone)
+            print(f"  ✓ {label} ({key})")
+        except Exception as e:
+            print(f"      could not set {key}: {e}", file=sys.stderr)
+
+
 def _cmd_status(_args: argparse.Namespace) -> int:
-    # Defer the whole table to auth.print_credential_summary — its emit
+    # Defer the credential rows to auth.print_credential_summary — its emit
     # callback is the only sink that sees credential-derived strings, so
     # cli.py keeps zero taint flow according to CodeQL.
     photon_auth.print_credential_summary(print)
-    # The two non-credential rows live here so the helper stays purely
-    # about credentials.
     node_bin = os.getenv("PHOTON_NODE_BIN") or shutil.which("node")
     sidecar_installed = (_SIDECAR_DIR / "node_modules").exists()
     print(f"  node binary         : {node_bin or '✗ missing (install Node 18+)'}")
@@ -218,8 +292,7 @@ def _cmd_status(_args: argparse.Namespace) -> int:
 
 
 def _cmd_install_sidecar(_args: argparse.Namespace) -> int:
-    rc = _install_sidecar()
-    return rc
+    return _install_sidecar()
 
 
 def _install_sidecar() -> int:
@@ -242,64 +315,6 @@ def _install_sidecar() -> int:
     return proc.returncode
 
 
-def _cmd_webhook(args: argparse.Namespace) -> int:
-    sub = getattr(args, "photon_webhook_command", None)
-    project_id, project_secret = photon_auth.load_project_credentials()
-    if not (project_id and project_secret):
-        print(
-            "no Photon project configured — run `hermes photon setup` first",
-            file=sys.stderr,
-        )
-        return 1
-
-    if sub == "register":
-        try:
-            data = photon_auth.register_webhook(
-                project_id, project_secret, webhook_url=args.url
-            )
-        except Exception as e:
-            print(f"register failed: {e}", file=sys.stderr)
-            return 1
-        # The helper does all the formatting + writing; cli.py never
-        # touches the signing-secret value, the path it was written
-        # to, or even the redacted-response dict. on_summary is a
-        # plain printer callback.
-        ok = photon_auth.persist_webhook_signing_secret(data, on_summary=print)
-        if not ok:
-            print(
-                "‼  Photon returned no signing secret in the response, "
-                "or the file write failed. Inspect your home directory "
-                "permissions and re-run; do not retry without first "
-                "deleting the orphaned webhook from the Photon dashboard.",
-                file=sys.stderr,
-            )
-            return 1
-        return 0
-
-    if sub == "list":
-        try:
-            data = photon_auth.list_webhooks(project_id, project_secret)
-        except Exception as e:
-            print(f"list failed: {e}", file=sys.stderr)
-            return 1
-        print(json.dumps(data, indent=2))
-        return 0
-
-    if sub == "delete":
-        try:
-            photon_auth.delete_webhook(
-                project_id, project_secret, webhook_id=args.webhook_id
-            )
-        except Exception as e:
-            print(f"delete failed: {e}", file=sys.stderr)
-            return 1
-        print(f"deleted webhook {args.webhook_id}")
-        return 0
-
-    print(f"unknown webhook subcommand: {sub}", file=sys.stderr)
-    return 2
-
-
 # ---------------------------------------------------------------------------
 # Gateway-setup entry point
 #

plugins/platforms/photon/plugin.yaml

@@ -1,52 +1,37 @@
 name: photon-platform
-label: Photon iMessage
+label: iMessage via Photon
 kind: platform
-version: 0.1.0
+version: 0.2.0
 description: >
   Photon Spectrum gateway adapter for Hermes Agent.
   Connects to iMessage (and other Spectrum interfaces) through Photon's
-  managed Spectrum platform. Inbound messages arrive as signed webhooks
-  on a local aiohttp server; outbound messages are sent via a small
-  supervised Node sidecar that runs the `spectrum-ts` SDK (Photon does
-  not currently expose a public HTTP send endpoint).
+  managed Spectrum platform. Both directions run over the `spectrum-ts`
+  SDK's long-lived gRPC stream via a small supervised Node sidecar —
+  inbound messages arrive on the SDK's `app.messages` stream (no webhook,
+  no public URL, no signing secret), and outbound messages are sent over
+  the same sidecar.
 
-  The plugin ships with a `hermes photon` CLI for the one-time login
-  + project + user setup, persists Spectrum credentials to
-  ``~/.hermes/auth.json`` under ``credential_pool.photon`` (token) and
-  ``credential_pool.photon_project`` (project id + secret), and exposes
-  Photon's free shared-line model so users can get started without a
-  paid plan.
+  The plugin ships with a `hermes photon` CLI for the one-time device
+  login + project + user setup. Runtime credentials are written to
+  ``~/.hermes/.env`` (``PHOTON_PROJECT_ID`` = the Spectrum project id,
+  ``PHOTON_PROJECT_SECRET``) like every other channel, with management
+  metadata (device token, dashboard project id) in ``~/.hermes/auth.json``.
+  Photon's free shared-line model lets users get started without a paid plan.
 author: NousResearch
 requires_env:
   - name: PHOTON_PROJECT_ID
-    description: "Spectrum project ID (set by `hermes photon setup`)"
-    prompt: "Photon Spectrum project ID"
+    description: "Spectrum project id (the project's spectrumProjectId; set by `hermes photon setup`)"
+    prompt: "Photon Spectrum project id"
     url: "https://app.photon.codes/"
     password: false
   - name: PHOTON_PROJECT_SECRET
-    description: "Spectrum project secret (set by `hermes photon setup`)"
-    prompt: "Photon Spectrum project secret"
+    description: "Project secret paired with the Spectrum project id (set by `hermes photon setup`)"
+    prompt: "Photon project secret"
     url: "https://app.photon.codes/"
     password: true
 optional_env:
-  - name: PHOTON_WEBHOOK_SECRET
-    description: "Per-URL HMAC-SHA256 signing secret returned at webhook registration"
-    prompt: "Photon webhook signing secret"
-    password: true
-  - name: PHOTON_WEBHOOK_PORT
-    description: "Local port the webhook receiver listens on (default 8788)"
-    prompt: "Webhook receiver port"
-    password: false
-  - name: PHOTON_WEBHOOK_PATH
-    description: "Path the webhook receiver listens on (default /photon/webhook)"
-    prompt: "Webhook receiver path"
-    password: false
-  - name: PHOTON_WEBHOOK_BIND
-    description: "Bind address for the webhook receiver (default 0.0.0.0)"
-    prompt: "Webhook bind address"
-    password: false
   - name: PHOTON_SIDECAR_PORT
-    description: "Loopback port for the Node sidecar control channel (default 8789)"
+    description: "Loopback port for the Node sidecar control + inbound channel (default 8789)"
     prompt: "Sidecar control port"
     password: false
   - name: PHOTON_SIDECAR_AUTOSTART
@@ -57,12 +42,8 @@ optional_env:
     description: "Path to the node binary (default: shutil.which('node'))"
     prompt: "Node executable path"
     password: false
-  - name: PHOTON_API_HOST
-    description: "Spectrum management API host (default https://spectrum.photon.codes)"
-    prompt: "Spectrum API host"
-    password: false
   - name: PHOTON_DASHBOARD_HOST
-    description: "Dashboard API host (default https://app.photon.codes)"
+    description: "Photon Dashboard API host (default https://app.photon.codes)"
     prompt: "Dashboard host"
     password: false
   - name: PHOTON_ALLOWED_USERS
@@ -82,8 +63,8 @@ optional_env:
     prompt: "Group mention patterns"
     password: false
   - name: PHOTON_HOME_CHANNEL
-    description: "Default Spectrum space ID for cron / notification delivery"
-    prompt: "Home space ID"
+    description: "Default Photon target for cron / notification delivery: Spectrum space id, DM GUID, or bare E.164 phone number"
+    prompt: "Home Photon target"
     password: false
   - name: PHOTON_HOME_CHANNEL_NAME
     description: "Human label for the home channel"

plugins/platforms/photon/sidecar/index.mjs

@@ -1,40 +1,46 @@
 // Hermes Agent — Photon Spectrum sidecar
 //
-// Spawned by `plugins/platforms/photon/adapter.py` to bridge outbound
-// messaging to Photon's Spectrum platform. Inbound messages go directly
-// from Photon's webhook to Hermes' Python aiohttp receiver — this
-// sidecar handles ONLY outbound calls (which require the spectrum-ts
-// SDK because Photon has no public HTTP send endpoint today).
+// Spawned by `plugins/platforms/photon/adapter.py` to bridge BOTH directions
+// of messaging to Photon's Spectrum platform via the `spectrum-ts` SDK (the
+// SDK is TypeScript-only, so a Node sidecar is unavoidable — there is no
+// Python SDK and no public HTTP message API).
 //
-// Protocol:
-//   - The sidecar listens on http://127.0.0.1:${PORT} (loopback only)
-//   - Each request must include `X-Hermes-Sidecar-Token: ${TOKEN}`
-//   - POST /healthz                     -> {"ok": true}
-//   - POST /send                        -> {"ok": true, "messageId": "..."}
-//       body: {"spaceId": "...", "text": "...", "replyTo": "..." | null}
-//   - POST /send-attachment             -> {"ok": true, "messageId": "..."}
+// Inbound  (gRPC -> Hermes): the SDK's `app.messages` async iterator is a
+//   long-lived gRPC stream. We serialize each `[space, message]` to a
+//   normalized JSON event and stream it to the Python adapter over a
+//   loopback `GET /inbound` (NDJSON). We pause pulling from the stream while
+//   no consumer is attached so a backlog isn't pulled-and-lost before the
+//   gateway connects.
+// Outbound (Hermes -> gRPC): `/send` drives `space.send(...)`; `/typing`
+//   sends the documented `typing("start" | "stop")` content builder.
+//
+// Protocol (all requests require `X-Hermes-Sidecar-Token: ${TOKEN}`):
+//   - GET  /inbound    -> 200 NDJSON stream; one JSON event per line, blank
+//                         lines are heartbeats. One consumer at a time.
+//   - POST /healthz     -> {"ok": true}
+//   - POST /send        -> {"ok": true, "messageId": "..."}
+//       body: {"spaceId": "...", "text": "..."}
+//   - 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
+//              "kind": "attachment" | "voice"}
+//   - POST /typing      -> {"ok": true}
+//       body: {"spaceId": "...", "state": "start" | "stop"}
+//   - POST /shutdown    -> {"ok": true}; then process exits
 //
 // On SIGINT/SIGTERM the sidecar calls `app.stop()` (3s graceful) before
-// exiting. Errors are logged to stderr; Python supervises restart.
+// exiting. Logs go to stderr; Python supervises restart.
 //
-// Env vars (all required):
-//   PHOTON_PROJECT_ID
+// Env vars (required):
+//   PHOTON_PROJECT_ID      (== the project's spectrumProjectId)
 //   PHOTON_PROJECT_SECRET
 //   PHOTON_SIDECAR_PORT
 //   PHOTON_SIDECAR_TOKEN
-//
 // Optional:
-//   PHOTON_SIDECAR_BIND  (default 127.0.0.1)
-//   PHOTON_API_HOST      (passed through to spectrum-ts if its config
-//                         honours it)
+//   PHOTON_SIDECAR_BIND    (default 127.0.0.1)
 
 import http from "node:http";
+import { once } from "node:events";
 
 const projectId = process.env.PHOTON_PROJECT_ID;
 const projectSecret = process.env.PHOTON_PROJECT_SECRET;
@@ -42,6 +48,17 @@ const port = parseInt(process.env.PHOTON_SIDECAR_PORT || "8789", 10);
 const bind = process.env.PHOTON_SIDECAR_BIND || "127.0.0.1";
 const sharedToken = process.env.PHOTON_SIDECAR_TOKEN;
 
+// Inbound attachments are read into memory and base64-inlined on the NDJSON
+// event so the Python adapter can cache the real bytes (and the agent can see
+// the image). Cap the size we inline — above it we forward metadata only and
+// the adapter surfaces a text marker, so one large video can't balloon a
+// single NDJSON line. Override via PHOTON_MAX_INLINE_ATTACHMENT_BYTES.
+const MAX_INLINE_ATTACHMENT_BYTES =
+  Number(process.env.PHOTON_MAX_INLINE_ATTACHMENT_BYTES) || 20 * 1024 * 1024;
+const DM_CHAT_GUID_RE = /^any;-;(\+\d{6,})$/;
+const E164_RE = /^\+\d{6,}$/;
+const MAX_KNOWN_SPACES = 2048;
+
 if (!projectId || !projectSecret || !sharedToken) {
   console.error(
     "photon-sidecar: PHOTON_PROJECT_ID, PHOTON_PROJECT_SECRET and " +
@@ -52,9 +69,15 @@ 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, attachment, voice;
+let Spectrum, imessage, attachment, voice, spectrumText, spectrumTyping;
 try {
-  ({ Spectrum, attachment, voice } = await import("spectrum-ts"));
+  ({
+    Spectrum,
+    attachment,
+    voice,
+    text: spectrumText,
+    typing: spectrumTyping,
+  } = await import("spectrum-ts"));
   ({ imessage } = await import("spectrum-ts/providers/imessage"));
 } catch (e) {
   console.error(
@@ -71,17 +94,168 @@ const app = await Spectrum({
   providers: [imessage.config()],
 });
 
-// Drain the inbound stream — Photon's webhook is the canonical inbound
-// path, but we still consume `app.messages` so spectrum-ts' internal
-// reconnect/heartbeat logic keeps running.  Each event is logged at
-// debug level; everything else is a no-op here.
+// ---------------------------------------------------------------------------
+// Inbound: forward `app.messages` (gRPC stream) to the Python consumer.
+
+// At most one Python consumer is attached at a time (the gateway adapter).
+let consumerRes = null;
+let consumerWaiters = [];
+const knownSpaces = new Map();
+
+function rememberKnownSpace(id, space) {
+  if (!id || typeof id !== "string" || !space) return;
+  if (knownSpaces.has(id)) knownSpaces.delete(id);
+  knownSpaces.set(id, space);
+  if (knownSpaces.size > MAX_KNOWN_SPACES) {
+    const oldest = knownSpaces.keys().next().value;
+    if (oldest) knownSpaces.delete(oldest);
+  }
+}
+
+function phoneTargetFromSpaceId(spaceId) {
+  if (typeof spaceId !== "string") return null;
+  if (E164_RE.test(spaceId)) return spaceId;
+  const dmGuid = spaceId.match(DM_CHAT_GUID_RE);
+  return dmGuid ? dmGuid[1] : null;
+}
+
+function rememberInboundSpace(space, message) {
+  const msgSpace = message?.space || {};
+  const ids = [space?.id, msgSpace.id];
+  for (const id of ids) {
+    rememberKnownSpace(id, space);
+    const phone = phoneTargetFromSpaceId(id);
+    if (phone) rememberKnownSpace(phone, space);
+  }
+}
+
+function waitForConsumer() {
+  if (consumerRes) return Promise.resolve();
+  return new Promise((resolve) => consumerWaiters.push(resolve));
+}
+
+function setConsumer(res) {
+  consumerRes = res;
+  const waiters = consumerWaiters;
+  consumerWaiters = [];
+  for (const resolve of waiters) resolve();
+}
+
+function clearConsumer(res) {
+  if (consumerRes === res) consumerRes = null;
+}
+
+// Write one NDJSON line to the active consumer. Blocks until a consumer is
+// connected; if the write fails (consumer vanished mid-flight) we wait for a
+// new consumer and retry, so a message is never silently dropped here.
+async function deliver(line) {
+  for (;;) {
+    await waitForConsumer();
+    const res = consumerRes;
+    if (!res) continue;
+    try {
+      const flushed = res.write(line + "\n");
+      if (!flushed) await once(res, "drain");
+      return;
+    } catch {
+      clearConsumer(res);
+    }
+  }
+}
+
+async function normalizeContent(content) {
+  if (!content || typeof content !== "object") {
+    return { type: "unknown" };
+  }
+  if (content.type === "text") {
+    return { type: "text", text: content.text || "" };
+  }
+  if (content.type === "attachment") {
+    const meta = {
+      type: "attachment",
+      id: content.id ?? null,
+      name: content.name ?? null,
+      mimeType: content.mimeType ?? null,
+      size: typeof content.size === "number" ? content.size : null,
+    };
+    // Read the bytes eagerly and base64-inline them as `data` so the Python
+    // adapter can cache the real file (the agent then sees the image itself).
+    // The spectrum-ts attachment object may not outlive this stream
+    // iteration, so a lazy/on-demand fetch isn't safe. Over-cap attachments
+    // (when size is known up front) are forwarded as metadata only and the
+    // adapter falls back to a text marker. A read failure must never break
+    // the inbound loop — we just drop `data` and forward metadata.
+    if (meta.size !== null && meta.size > MAX_INLINE_ATTACHMENT_BYTES) {
+      console.error(
+        `photon-sidecar: attachment ${meta.name ?? meta.id} (${meta.size} bytes) ` +
+          `exceeds inline cap ${MAX_INLINE_ATTACHMENT_BYTES}; forwarding metadata only`
+      );
+      return meta;
+    }
+    if (typeof content.read === "function") {
+      try {
+        const buf = await content.read();
+        // Guard the case where size was unknown but the bytes turn out to be
+        // over the cap.
+        if (buf && buf.length > MAX_INLINE_ATTACHMENT_BYTES) {
+          console.error(
+            `photon-sidecar: attachment ${meta.name ?? meta.id} (${buf.length} bytes) ` +
+              `exceeds inline cap after read; forwarding metadata only`
+          );
+          return meta;
+        }
+        meta.data = Buffer.from(buf).toString("base64");
+        meta.encoding = "base64";
+      } catch (e) {
+        console.error(
+          "photon-sidecar: failed to read attachment bytes " +
+            "(forwarding metadata only): " +
+            (e && e.stack ? e.stack : String(e))
+        );
+      }
+    }
+    return meta;
+  }
+  return { type: content.type || "unknown" };
+}
+
+async function normalizeEvent(space, message) {
+  try {
+    const msgSpace = message.space || {};
+    const ts = message.timestamp;
+    return {
+      messageId: message.id ?? null,
+      platform: message.platform || space.__platform || "iMessage",
+      space: {
+        id: space.id ?? msgSpace.id ?? null,
+        // iMessage spaces carry `type` ("dm"|"group") and `phone` directly.
+        type: space.type ?? msgSpace.type ?? "dm",
+        phone: space.phone ?? msgSpace.phone ?? null,
+      },
+      sender: { id: message.sender ? message.sender.id : null },
+      content: await normalizeContent(message.content),
+      timestamp:
+        ts instanceof Date ? ts.toISOString() : ts ? String(ts) : null,
+    };
+  } catch (e) {
+    console.error(
+      "photon-sidecar: failed to normalize inbound message: " + String(e)
+    );
+    return null;
+  }
+}
+
 (async () => {
   try {
-    for await (const [, message] of app.messages) {
-      console.error(
-        `photon-sidecar: drained inbound from ${message.platform} ` +
-          `space=${message.space?.id}`
-      );
+    for await (const [space, message] of app.messages) {
+      // Only forward inbound messages (ignore our own outbound echoes).
+      if (message && message.direction && message.direction !== "inbound") {
+        continue;
+      }
+      rememberInboundSpace(space, message);
+      const event = await normalizeEvent(space, message);
+      if (!event) continue;
+      await deliver(JSON.stringify(event));
     }
   } catch (e) {
     console.error(
@@ -91,6 +265,9 @@ const app = await Spectrum({
   }
 })();
 
+// ---------------------------------------------------------------------------
+// HTTP control + inbound server (loopback only).
+
 async function readBody(req) {
   const chunks = [];
   for await (const chunk of req) chunks.push(chunk);
@@ -130,27 +307,73 @@ function ok(res, data) {
   res.end(JSON.stringify({ ok: true, ...data }));
 }
 
-async function resolveSpace(spaceId) {
-  // spectrum-ts exposes the same Space methods via `app.space(spaceId)` /
-  // narrowed helpers; we fall back through a few accessor shapes to
-  // tolerate small SDK API drift.
-  if (typeof app.space === "function") {
-    return await app.space(spaceId);
-  }
-  if (app.spaces && typeof app.spaces.get === "function") {
-    return await app.spaces.get(spaceId);
-  }
-  // Last resort — the platform-narrowed helper.
-  if (imessage) {
-    const im = imessage(app);
-    if (typeof im.space === "function") {
-      try {
-        return await im.space({ id: spaceId });
-      } catch {
-        /* fall through */
-      }
+function handleInbound(req, res) {
+  res.statusCode = 200;
+  res.setHeader("Content-Type", "application/x-ndjson");
+  res.setHeader("Cache-Control", "no-store");
+  res.setHeader("Connection", "keep-alive");
+  // One consumer at a time — a fresh connection (e.g. after a reconnect)
+  // supersedes the previous one.
+  if (consumerRes && consumerRes !== res) {
+    try {
+      consumerRes.end();
+    } catch {
+      /* ignore */
     }
   }
+  setConsumer(res);
+  // Heartbeat keeps the socket warm through idle periods and lets the Python
+  // side detect a dead pipe promptly.
+  const heartbeat = setInterval(() => {
+    try {
+      res.write("\n");
+    } catch {
+      /* ignore */
+    }
+  }, 25000);
+  const cleanup = () => {
+    clearInterval(heartbeat);
+    clearConsumer(res);
+  };
+  req.on("close", cleanup);
+  req.on("aborted", cleanup);
+  res.on("error", cleanup);
+}
+
+async function resolveSpace(spaceId) {
+  const cached = knownSpaces.get(spaceId);
+  if (cached) return cached;
+
+  const phoneTarget = phoneTargetFromSpaceId(spaceId);
+  // A bare E.164 phone number addresses a DM. Resolve the user, then the (DM)
+  // space — `imessage(app).user(phone)` -> `im.space(user)` — so callers can
+  // pass just "+1..." (e.g. PHOTON_HOME_CHANNEL for cron delivery) instead of
+  // an opaque inbound space id. Photon also represents DM chat ids as
+  // `any;-;+1...`; normalize those through the same path so replies to inbound
+  // DMs still resolve after Python stores the inbound `space.id`.
+  if (phoneTarget && imessage) {
+    try {
+      const im = imessage(app);
+      const user = await im.user(phoneTarget);
+      const space = await im.space(user);
+      rememberKnownSpace(spaceId, space);
+      rememberKnownSpace(phoneTarget, space);
+      rememberKnownSpace(space?.id, space);
+      return space;
+    } catch (e) {
+      console.error(
+        "photon-sidecar: phone->DM resolution failed: " +
+          (e && e.stack ? e.stack : String(e))
+      );
+    }
+  }
+  // No cache hit and not a phone/DM target. spectrum-ts exposes no API to
+  // rehydrate an arbitrary opaque space id: a Space is only obtained from the
+  // inbound `[space, message]` stream (cached above in `knownSpaces`) or
+  // reconstructed for a DM from its phone number. So a group space whose cache
+  // entry was lost — e.g. after a sidecar restart with no fresh inbound message
+  // in that group — cannot be resolved here; a new inbound message in the group
+  // re-warms the cache. DMs are unaffected (reconstructed from the phone).
   throw new Error(`unable to resolve space id ${spaceId}`);
 }
 
@@ -158,6 +381,10 @@ const server = http.createServer(async (req, res) => {
   if (req.headers["x-hermes-sidecar-token"] !== sharedToken) {
     return unauthorized(res);
   }
+  // Long-lived inbound NDJSON stream.
+  if (req.method === "GET" && req.url === "/inbound") {
+    return handleInbound(req, res);
+  }
   if (req.method !== "POST") {
     res.statusCode = 405;
     return res.end();
@@ -173,18 +400,16 @@ const server = http.createServer(async (req, res) => {
     }
     const body = await readBody(req);
     if (req.url === "/send") {
-      const { spaceId, text, replyTo } = body || {};
+      const { spaceId, text } = body || {};
       if (!spaceId || typeof text !== "string") {
         return badRequest(res, "spaceId and text are required");
       }
       const space = await resolveSpace(spaceId);
-      const result = replyTo
-        ? await space.send(text, { replyTo })
-        : await space.send(text);
-      return ok(res, { messageId: result?.id || result?.messageId || null });
+      const result = await space.send(spectrumText(text));
+      return ok(res, { messageId: result?.id || null });
     }
     if (req.url === "/send-attachment") {
-      const { spaceId, path, name, mimeType, caption, kind, replyTo } =
+      const { spaceId, path, name, mimeType, caption, kind } =
         body || {};
       if (!spaceId || typeof path !== "string" || !path) {
         return badRequest(res, "spaceId and path are required");
@@ -202,16 +427,13 @@ const server = http.createServer(async (req, res) => {
           ? 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);
+      const result = 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);
+          await space.send(spectrumText(caption));
         } catch (e) {
           console.error(
             "photon-sidecar: attachment sent but caption failed: " +
@@ -219,17 +441,16 @@ const server = http.createServer(async (req, res) => {
           );
         }
       }
-      return ok(res, { messageId: result?.id || result?.messageId || null });
+      return ok(res, { messageId: result?.id || null });
     }
     if (req.url === "/typing") {
-      const { spaceId } = body || {};
+      const { spaceId, state = "start" } = body || {};
       if (!spaceId) return badRequest(res, "spaceId is required");
-      const space = await resolveSpace(spaceId);
-      if (typeof space.typing === "function") {
-        await space.typing();
-      } else if (typeof space.setTyping === "function") {
-        await space.setTyping(true);
+      if (state !== "start" && state !== "stop") {
+        return badRequest(res, "state must be start or stop");
       }
+      const space = await resolveSpace(spaceId);
+      await space.send(spectrumTyping(state));
       return ok(res, {});
     }
     res.statusCode = 404;

plugins/platforms/photon/sidecar/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@hermes-agent/photon-sidecar",
   "private": true,
-  "version": "0.1.0",
+  "version": "0.2.0",
   "description": "Spectrum-ts bridge for the Hermes Agent Photon platform plugin.",
   "type": "module",
   "main": "index.mjs",
@@ -12,6 +12,6 @@
     "node": ">=18.17"
   },
   "dependencies": {
-    "spectrum-ts": "^0.1.0"
+    "spectrum-ts": "^1.17.1"
   }
 }

plugins/platforms/simplex/plugin.yaml

@@ -1,7 +1,7 @@
 name: simplex-platform
 label: SimpleX Chat
 kind: platform
-version: 1.0.0
+version: 1.1.0
 description: >
   SimpleX Chat gateway adapter for Hermes Agent.
   Connects to a local simplex-chat daemon via WebSocket and relays
@@ -9,7 +9,7 @@ description: >
   SimpleX is decentralised and assigns no persistent user IDs —
   every contact is an opaque internal ID generated at connection
   time, making it one of the most private messengers available.
-author: Mibayy
+author: Mibayy, jooray
 # ``requires_env`` and ``optional_env`` entries are surfaced in the
 # ``hermes config`` UI via the platform-plugin env var injector in
 # ``hermes_cli/config.py``.
@@ -27,6 +27,18 @@ optional_env:
     description: "Allow any contact to talk to the bot (dev only — disables allowlist)"
     prompt: "Allow all contacts? (true/false)"
     password: false
+  - name: SIMPLEX_AUTO_ACCEPT
+    description: "Auto-accept incoming contact requests (default: true)"
+    prompt: "Auto-accept contact requests? (true/false)"
+    password: false
+  - name: SIMPLEX_GROUP_ALLOWED
+    description: >-
+      Comma-separated SimpleX group IDs the bot should participate in, or
+      '*' to allow any group. Omit to ignore group messages entirely
+      (safer default — a bot in a group otherwise processes every
+      member's traffic).
+    prompt: "Allowed group IDs (comma-separated, or '*' for any)"
+    password: false
   - name: SIMPLEX_HOME_CHANNEL
     description: "Default contact/group ID for cron / notification delivery"
     prompt: "Home channel contact/group ID (or empty)"
@@ -35,3 +47,10 @@ optional_env:
     description: "Human label for the home channel (defaults to the ID)"
     prompt: "Home channel display name (or empty)"
     password: false
+  - name: HERMES_SIMPLEX_TEXT_BATCH_DELAY
+    description: >-
+      Quiet-period seconds (default: 0.8) used to concatenate rapid-fire
+      inbound text messages into a single MessageEvent — same pattern as
+      Telegram's text batching.
+    prompt: "Text batch flush delay in seconds (default 0.8)"
+    password: false

scripts/release.py

@@ -45,6 +45,8 @@ ACP_REGISTRY_MANIFEST = REPO_ROOT / "acp_registry" / "agent.json"
 
 # Auto-extracted from noreply emails + manual overrides
 AUTHOR_MAP = {
+    "zhuhaoyu0909@icloud.com": "underthestars-zhy",
+    "raysun12142006@gmail.com": "yanxue06",
     "alberto.regalado@ymail.com": "ARegalado1",
     "alchemistchaos@protonmail.com": "AlchemistChaos",  # co-author only
     "gilad@smiti.ai": "giladbau",
@@ -971,6 +973,7 @@ AUTHOR_MAP = {
     "draixagent@gmail.com": "draix",
     "martin.alca@gmail.com": "draix",
     "junminliu@gmail.com": "JimLiu",
+    "juraj@bednar.io": "jooray",
     "jarvischer@gmail.com": "maxchernin",
     "levantam.98.2324@gmail.com": "LVT382009",
     "zhurongcheng@rcrai.com": "heykb",
@@ -1241,6 +1244,7 @@ AUTHOR_MAP = {
     "charliekerfoot@gmail.com": "CharlieKerfoot",  # PR #18951
     # Debug share upload-time redaction (May 2026)
     "dhuysamen@gmail.com": "GodsBoy",  # PR #19318
+    "github@nadyahermes.anonaddy.com": "ruangraung",  # PR #42308
     "mrcoferland@gmail.com": "mrcoferland",  # PR #19023
     "chenlinfeng@ruije.com.cn": "noOne-list",  # PR #19050
     "briansu@Mac-mini.attlocal.net": "likejudy",  # PR #19052

tests/cli/test_slash_confirm_windows.py

@@ -1,29 +1,31 @@
-"""Regression tests for issue #30768 and #32383.
+"""Regression tests for #30768, #32383, and #33961.
 
-``_prompt_text_input_modal`` uses a queue-based modal that relies on
-prompt_toolkit key bindings receiving keyboard events.  On Windows the
-prompt_toolkit input channel can deadlock when the modal is entered from
-the ``process_loop`` daemon thread.  The fix falls back to the simpler
-``_prompt_text_input`` (stdin-based) prompt on Windows.
+``_prompt_text_input_modal`` answers destructive-slash confirmations through a
+queue-based modal driven by prompt_toolkit key bindings.  When invoked from the
+``process_loop`` daemon thread it sets the modal up on the app's event loop via
+``call_soon_threadsafe``, so it is safe on every platform — including native
+Windows (#33961), where the earlier ``sys.platform == "win32"`` → raw ``input()``
+fallback deadlocked the daemon thread against prompt_toolkit's stdin ownership.
 
 These tests verify:
-1. Windows detection triggers the stdin fallback
-2. Non-Windows daemon threads still use the modal via the app loop
-3. macOS/Linux main-thread path still uses the modal (no regression)
-4. No-app path still uses the stdin fallback (existing behavior)
-5. Empty choices returns None (existing behavior)
+1. Daemon-thread confirm uses the modal via the app loop on Linux AND native
+   Windows (#33961) — never the raw stdin fallback, never a hang.
+2. Main-thread confirm with a running app uses the modal.
+3. The raw stdin fallback is kept ONLY for the safe cases: no running app, and
+   (on win32, off-thread) a scheduling failure degrades to a clean cancel.
+4. Empty choices returns None.
 """
 
-import queue
 import sys
 import threading
 import time
 from unittest.mock import MagicMock, patch
 
+import pytest
 
 
 def _make_cli():
-    """Minimal HermesCLI shell exposing prompt/modal helpers."""
+    """Minimal HermesCLI shell exposing the prompt/modal helpers."""
     import cli as cli_mod
 
     obj = object.__new__(cli_mod.HermesCLI)
@@ -37,9 +39,6 @@ def _make_cli():
     return obj
 
 
-# ---------------------------------------------------------------------------
-# Sample choices used across tests
-# ---------------------------------------------------------------------------
 _SAMPLE_CHOICES = [
     ("once", "Approve Once", "proceed this time only"),
     ("always", "Always Approve", "proceed and silence this prompt permanently"),
@@ -47,119 +46,106 @@ _SAMPLE_CHOICES = [
 ]
 
 
-class TestModalWindowsFallback:
-    """Windows dead-lock regression tests for _prompt_text_input_modal."""
+def _answer_modal_when_open(cli, response, stop=None):
+    """Push ``response`` onto the modal's response_queue once it opens.
 
-    def test_windows_falls_back_to_stdin(self):
-        """On Windows, _prompt_text_input_modal should use _prompt_text_input."""
+    Gives up after ~2s, or early when ``stop`` is set (the modal will never open,
+    e.g. a scheduling failure) so degraded-path tests don't wait the full budget.
+    """
+    for _ in range(100):
+        if stop is not None and stop.is_set():
+            return
+        state = cli._slash_confirm_state
+        if state and "response_queue" in state:
+            state["response_queue"].put(response)
+            return
+        time.sleep(0.02)
+
+
+def _run_on_daemon(call, cli, *, platform, response, schedule=None):
+    """Invoke ``call`` on a daemon thread — as the process_loop does — answering
+    the modal with ``response`` once it opens.
+
+    Returns ``{result, stdin_called, capture, restore}``.  ``schedule`` overrides
+    the ``call_soon_threadsafe`` side effect (default: run the callback inline);
+    pass a raiser to simulate a scheduling failure.  Fails if the worker hangs,
+    which is the deadlock canary for #33961.
+    """
+    outcome = {"capture": [], "restore": [], "result": None, "stdin_called": False}
+    done = threading.Event()
+
+    def _worker():
+        try:
+            with patch.object(sys, "platform", platform), \
+                 patch.object(cli._app.loop, "call_soon_threadsafe", side_effect=schedule or (lambda cb: cb())), \
+                 patch.object(cli, "_prompt_text_input") as mock_stdin, \
+                 patch.object(cli, "_invalidate"), \
+                 patch.object(cli, "_capture_modal_input_snapshot", side_effect=lambda: outcome["capture"].append(1)), \
+                 patch.object(cli, "_restore_modal_input_snapshot", side_effect=lambda: outcome["restore"].append(1)):
+                outcome["result"] = call()
+                outcome["stdin_called"] = mock_stdin.called
+        finally:
+            done.set()
+
+    worker = threading.Thread(target=_worker, daemon=True)
+    answerer = threading.Thread(target=_answer_modal_when_open, args=(cli, response, done), daemon=True)
+    answerer.start()
+    worker.start()
+    worker.join(timeout=2.0)
+    answerer.join(timeout=2.0)
+    assert not worker.is_alive(), "daemon thread hung — modal deadlocked"
+    return outcome
+
+
+class TestModal:
+    """Behaviour of _prompt_text_input_modal across platforms and threads."""
+
+    @pytest.mark.parametrize("platform", ["linux", "win32"])
+    def test_daemon_thread_uses_modal_via_app_loop(self, platform):
+        """Off the process_loop daemon thread, the confirm uses the modal via
+        call_soon_threadsafe on every platform — including native Windows, where
+        the old win32 early-return deadlocked on raw input() (#33961)."""
         cli = _make_cli()
-
-        with patch.object(sys, "platform", "win32"), \
-             patch.object(cli, "_prompt_text_input", return_value="1") as mock_stdin:
-            result = cli._prompt_text_input_modal(
-                title="⚠️  /new — destroys conversation state",
+        outcome = _run_on_daemon(
+            lambda: cli._prompt_text_input_modal(
+                title="⚠️  /reset",
                 detail="This starts a fresh session.",
                 choices=_SAMPLE_CHOICES,
-            )
+                timeout=5,
+            ),
+            cli,
+            platform=platform,
+            response="once",
+        )
+        assert outcome["stdin_called"] is False, "must use the modal, not raw input()"
+        assert outcome["result"] == "once"
+        assert outcome["capture"] == [1]
+        assert outcome["restore"] == [1]
+        assert cli._slash_confirm_state is None
 
-        # The stdin-based fallback was used, not the modal queue path.
-        mock_stdin.assert_called_once_with("Choice [1/2/3]: ")
-        assert result == "1"
-
-    def test_non_main_thread_uses_modal_via_app_loop(self):
-        """Off the main thread on Linux, keep the modal path via app-loop setup."""
+    def test_main_thread_with_app_uses_modal(self):
+        """On the main thread with a running app, the queue-based modal is used."""
         cli = _make_cli()
-        result_holder = {}
-        setup_calls = []
-        teardown_calls = []
-
-        def _call_soon_threadsafe(callback):
-            callback()
-
-        def run_on_daemon():
-            with patch.object(sys, "platform", "linux"), \
-                 patch.object(cli._app.loop, "call_soon_threadsafe", side_effect=_call_soon_threadsafe), \
-                 patch.object(cli, "_prompt_text_input") as mock_stdin, \
-                 patch.object(cli, "_capture_modal_input_snapshot", side_effect=lambda: setup_calls.append("capture")), \
-                 patch.object(cli, "_restore_modal_input_snapshot", side_effect=lambda: teardown_calls.append("restore")):
-                result_holder["result"] = cli._prompt_text_input_modal(
-                    title="⚠️  /reset",
-                    detail="This starts a fresh session.",
-                    choices=_SAMPLE_CHOICES,
-                    timeout=5,
-                )
-                result_holder["stdin_called"] = mock_stdin.called
-
-        def _submit_after_delay():
-            time.sleep(0.2)
-            state = cli._slash_confirm_state
-            if state and "response_queue" in state:
-                state["response_queue"].put("once")
-
-        submitter = threading.Thread(target=_submit_after_delay, daemon=True)
-        t = threading.Thread(target=run_on_daemon, daemon=True)
-        submitter.start()
-        t.start()
-        t.join(timeout=2.0)
-        submitter.join(timeout=2.0)
-        assert not t.is_alive(), "daemon thread hung — modal deadlocked"
-        assert result_holder["stdin_called"] is False
-        assert result_holder["result"] == "once"
-        assert setup_calls == ["capture"]
-        assert teardown_calls == ["restore"]
-
-    def test_main_thread_non_windows_uses_modal(self):
-        """On macOS/Linux main thread, the queue-based modal is still used."""
-        cli = _make_cli()
-
-        # We need to simulate the modal receiving a response. We'll patch
-        # the response_queue to immediately return a value.
         with patch.object(sys, "platform", "darwin"), \
              patch.object(cli, "_capture_modal_input_snapshot"), \
              patch.object(cli, "_restore_modal_input_snapshot"), \
-             patch.object(cli, "_invalidate"):
-            # Start the modal in a way that it will receive a response
-            # immediately via the queue.
-            original_queue = queue.Queue
-            original_time = time.monotonic
+             patch.object(cli, "_invalidate"), \
+             patch.object(cli, "_prompt_text_input") as mock_stdin:
+            answerer = threading.Thread(target=_answer_modal_when_open, args=(cli, "once"), daemon=True)
+            answerer.start()
+            result = cli._prompt_text_input_modal(
+                title="⚠️  /new",
+                detail="This starts a fresh session.",
+                choices=_SAMPLE_CHOICES,
+                timeout=5,
+            )
+            answerer.join(timeout=2.0)
 
-            def _fake_modal_flow(*args, **kwargs):
-                """Simulate the modal flow: set state, put response, return."""
-                # We'll directly test that the modal path is entered by
-                # checking that _slash_confirm_state was set.
-                pass
-
-            # Since we can't easily mock the internal queue, let's test
-            # that the modal path is entered by checking that
-            # _prompt_text_input was NOT called.
-            with patch.object(cli, "_prompt_text_input") as mock_stdin:
-                # Set up a response that will be put into the queue
-                # after the modal starts waiting.
-                def _submit_after_delay():
-                    time.sleep(0.2)
-                    state = cli._slash_confirm_state
-                    if state and "response_queue" in state:
-                        state["response_queue"].put("once")
-
-                submitter = threading.Thread(target=_submit_after_delay, daemon=True)
-                submitter.start()
-
-                result = cli._prompt_text_input_modal(
-                    title="⚠️  /new",
-                    detail="This starts a fresh session.",
-                    choices=_SAMPLE_CHOICES,
-                    timeout=5,
-                )
-
-                submitter.join(timeout=2.0)
-
-            # The stdin fallback should NOT have been called.
-            mock_stdin.assert_not_called()
-            # The result should be "once" from the simulated modal response.
-            assert result == "once"
+        mock_stdin.assert_not_called()
+        assert result == "once"
 
     def test_no_app_falls_back_to_stdin(self):
-        """Without a prompt_toolkit app, always use stdin fallback."""
+        """Without a running app (oneshot / non-interactive), use the stdin prompt."""
         cli = _make_cli()
         cli._app = None
 
@@ -173,78 +159,102 @@ class TestModalWindowsFallback:
         mock_stdin.assert_called_once_with("Choice [1/2/3]: ")
         assert result == "3"
 
-    def test_empty_choices_returns_none(self):
-        """Empty choices list should return None without prompting."""
-        cli = _make_cli()
-
-        with patch.object(cli, "_prompt_text_input") as mock_stdin:
-            result = cli._prompt_text_input_modal(
-                title="Test",
-                detail="Test",
-                choices=[],
-            )
-
-        mock_stdin.assert_not_called()
-        assert result is None
-
-    def test_windows_fallback_does_not_set_modal_state(self):
-        """Verify Windows fallback doesn't leave _slash_confirm_state set."""
+    def test_windows_no_app_falls_back_to_stdin(self):
+        """win32 without a running app keeps stdin — the only case where the raw
+        prompt is safe on Windows, since no app owns the console to deadlock."""
         cli = _make_cli()
+        cli._app = None
 
         with patch.object(sys, "platform", "win32"), \
-             patch.object(cli, "_prompt_text_input", return_value="1"):
-            cli._prompt_text_input_modal(
-                title="⚠️  /reset",
+             patch.object(cli, "_prompt_text_input", return_value="1") as mock_stdin:
+            result = cli._prompt_text_input_modal(
+                title="⚠️  /new — destroys conversation state",
                 detail="This starts a fresh session.",
                 choices=_SAMPLE_CHOICES,
             )
 
+        mock_stdin.assert_called_once_with("Choice [1/2/3]: ")
+        assert result == "1"
+
+    def test_windows_scheduling_failure_clean_cancels(self):
+        """win32 off the main thread: if marshaling onto the app loop fails, cancel
+        cleanly (None) rather than fall to raw input() (which deadlocks on native
+        Windows) or hang. Asserts the _stdin_fallback guard (#33961)."""
+        cli = _make_cli()
+
+        def _raise(_cb):
+            raise RuntimeError("loop closed")
+
+        outcome = _run_on_daemon(
+            lambda: cli._prompt_text_input_modal(
+                title="⚠️  /reset",
+                detail="This starts a fresh session.",
+                choices=_SAMPLE_CHOICES,
+                timeout=5,
+            ),
+            cli,
+            platform="win32",
+            response="once",
+            schedule=_raise,
+        )
+        assert outcome["stdin_called"] is False, "win32 off-thread must NOT call raw input()"
+        assert outcome["result"] is None
         assert cli._slash_confirm_state is None
 
-    def test_non_main_thread_modal_clears_state(self):
-        """Verify daemon-thread modal teardown does not leave state behind."""
+    @pytest.mark.parametrize(
+        "platform, expect_stdin, expect_result",
+        [("win32", False, None), ("linux", True, "1")],
+    )
+    def test_daemon_thread_no_app_loop_uses_fallback(self, platform, expect_stdin, expect_result):
+        """Off the daemon thread with no resolvable app loop (``self._app.loop``
+        is None / raises), the modal can never be scheduled, so the method short-
+        circuits at the app_loop-is-None site (cli.py ~7260) — a distinct path
+        from a call_soon_threadsafe failure. win32 clean-cancels (None) instead of
+        deadlocking on raw input(); other platforms keep the stdin prompt."""
         cli = _make_cli()
-        errors = []
+        cli._app.loop = None  # forces app_loop is None, off the main thread
 
-        def _call_soon_threadsafe(callback):
-            callback()
+        outcome = {"result": None, "stdin_called": False}
+        done = threading.Event()
 
-        def run_on_daemon():
+        def _worker():
             try:
-                with patch.object(sys, "platform", "linux"), \
-                     patch.object(cli._app.loop, "call_soon_threadsafe", side_effect=_call_soon_threadsafe):
-                    def _submit_after_delay():
-                        time.sleep(0.2)
-                        state = cli._slash_confirm_state
-                        if state and "response_queue" in state:
-                            state["response_queue"].put("cancel")
-
-                    submitter = threading.Thread(target=_submit_after_delay, daemon=True)
-                    submitter.start()
-                    cli._prompt_text_input_modal(
-                        title="⚠️  /new",
+                with patch.object(sys, "platform", platform), \
+                     patch.object(cli, "_prompt_text_input", return_value="1") as mock_stdin, \
+                     patch.object(cli, "_invalidate"):
+                    outcome["result"] = cli._prompt_text_input_modal(
+                        title="⚠️  /reset",
                         detail="This starts a fresh session.",
                         choices=_SAMPLE_CHOICES,
                         timeout=5,
                     )
-                    submitter.join(timeout=2.0)
-                if cli._slash_confirm_state is not None:
-                    errors.append("_slash_confirm_state should be None")
-            except Exception as exc:
-                errors.append(str(exc))
+                    outcome["stdin_called"] = mock_stdin.called
+            finally:
+                done.set()
 
-        t = threading.Thread(target=run_on_daemon, daemon=True)
-        t.start()
-        t.join(timeout=2.0)
-        assert not errors, f"unexpected errors: {errors}"
+        worker = threading.Thread(target=_worker, daemon=True)
+        worker.start()
+        worker.join(timeout=2.0)
+        assert not worker.is_alive(), "daemon thread hung — modal deadlocked"
+        assert outcome["stdin_called"] is expect_stdin
+        assert outcome["result"] == expect_result
         assert cli._slash_confirm_state is None
 
+    def test_empty_choices_returns_none(self):
+        """Empty choices returns None without prompting."""
+        cli = _make_cli()
+
+        with patch.object(cli, "_prompt_text_input") as mock_stdin:
+            result = cli._prompt_text_input_modal(title="Test", detail="Test", choices=[])
+
+        mock_stdin.assert_not_called()
+        assert result is None
+
 
 class TestConfirmDestructiveSlashWindows:
-    """Integration-level tests for _confirm_destructive_slash on Windows."""
+    """End-to-end _confirm_destructive_slash on the native-Windows daemon thread."""
 
-    def test_confirm_destructive_slash_bypasses_modal_on_windows(self):
-        """_confirm_destructive_slash should work on Windows via stdin fallback."""
+    def _make_interactive_cli(self):
         cli = _make_cli()
         cli.model = "test-model"
         cli._agent_running = False
@@ -255,37 +265,140 @@ class TestConfirmDestructiveSlashWindows:
         cli._pending_tool_info = {}
         cli._tool_start_time = 0.0
         cli._last_scrollback_tool = ""
+        return cli
 
-        with patch.object(sys, "platform", "win32"), \
-             patch.object(cli, "_prompt_text_input", return_value="1"), \
-             patch("cli.load_cli_config", return_value={"approvals": {"destructive_slash_confirm": True}}):
-            result = cli._confirm_destructive_slash(
-                "new",
-                "This starts a fresh session.\nThe current conversation history will be discarded.",
+    @pytest.mark.parametrize(
+        "response, expected",
+        [("once", "once"), ("cancel", None)],
+    )
+    def test_confirm_destructive_slash_uses_modal_on_windows(self, response, expected):
+        """On native Windows, the bare /new confirm drives the modal (not stdin)
+        and returns the chosen outcome — the bug #33961 froze this path."""
+        cli = self._make_interactive_cli()
+        with patch("cli.load_cli_config", return_value={"approvals": {"destructive_slash_confirm": True}}):
+            outcome = _run_on_daemon(
+                lambda: cli._confirm_destructive_slash(
+                    "new",
+                    "This starts a fresh session.\nThe current conversation history will be discarded.",
+                ),
+                cli,
+                platform="win32",
+                response=response,
             )
 
-        assert result == "once"
+        assert outcome["stdin_called"] is False
+        assert outcome["result"] == expected
 
-    def test_confirm_destructive_slash_cancelled_on_windows(self):
-        """Cancellation via stdin fallback works on Windows."""
+
+class TestNativeWindowsNoRawInputDeadlock:
+    """Anti-regression guard exercising the REAL ``_prompt_text_input``.
+
+    Every other test here mocks ``_prompt_text_input`` away, so they only
+    assert *routing* (modal vs. stdin) — they cannot observe the actual hang
+    that #33961 was.  The historical regression was precisely that
+    ``_prompt_text_input_modal`` delegated to the *real* ``_prompt_text_input``
+    on native Windows, which on a non-main thread runs a bare ``input()`` that
+    blocks forever against prompt_toolkit's stdin ownership.
+
+    These tests let the real ``_prompt_text_input`` run with a blocking
+    ``input()`` and assert the worker thread never hangs.  They fail on the
+    pre-#33961 code (win32 → ``_prompt_text_input`` → off-main ``input()``)
+    and pass once the modal path / clean-cancel fallback is in place.
+    """
+
+    def test_win32_daemon_thread_never_blocks_on_real_input(self):
+        """A blocking input() must NOT hang the daemon thread on win32.
+
+        Drives the genuine helper chain (no mock of ``_prompt_text_input``)
+        with ``builtins.input`` patched to block forever. The confirm must
+        resolve via the app-loop modal (answered on a background thread, as
+        the real key bindings would) and never sit in ``input()``.  On the
+        pre-#33961 code the win32 early-return routed to the real
+        ``_prompt_text_input`` → off-main ``input()`` → permanent hang.
+        """
         cli = _make_cli()
-        cli.model = "test-model"
-        cli._agent_running = False
-        cli._spinner_text = ""
-        cli._should_exit = False
-        cli._command_running = False
-        cli.session_id = "test-session"
-        cli._pending_tool_info = {}
-        cli._tool_start_time = 0.0
-        cli._last_scrollback_tool = ""
+        cli._app.loop.call_soon_threadsafe = lambda cb: cb()
 
-        with patch.object(sys, "platform", "win32"), \
-             patch.object(cli, "_prompt_text_input", return_value="3"), \
-             patch("cli.load_cli_config", return_value={"approvals": {"destructive_slash_confirm": True}}):
-            result = cli._confirm_destructive_slash(
-                "reset",
-                "This starts a fresh session.\nThe current conversation history will be discarded.",
-            )
+        def _blocking_input(prompt=""):  # stands in for "no line ever arrives"
+            time.sleep(30)
+            return "1"
 
-        # Choice "3" normalizes to "cancel", which returns None.
-        assert result is None
+        outcome = {}
+        done = threading.Event()
+
+        def _worker():
+            try:
+                with patch.object(sys, "platform", "win32"), \
+                     patch("builtins.input", side_effect=_blocking_input), \
+                     patch.object(cli, "_capture_modal_input_snapshot"), \
+                     patch.object(cli, "_restore_modal_input_snapshot"), \
+                     patch.object(cli, "_invalidate"):
+                    outcome["result"] = cli._prompt_text_input_modal(
+                        title="/new",
+                        detail="destroys conversation state",
+                        choices=_SAMPLE_CHOICES,
+                        timeout=3,
+                    )
+            finally:
+                done.set()
+
+        worker = threading.Thread(target=_worker, daemon=True)
+        answerer = threading.Thread(
+            target=_answer_modal_when_open, args=(cli, "cancel", done), daemon=True
+        )
+        answerer.start()
+        worker.start()
+        worker.join(timeout=5.0)
+        answerer.join(timeout=5.0)
+        assert not worker.is_alive(), (
+            "daemon thread hung in real input() — native-Windows confirm "
+            "deadlock regressed (#33961)"
+        )
+        # cancel → None; the point is it RETURNED rather than blocking forever.
+        assert outcome.get("result") in (None, "cancel")
+
+    def test_win32_scheduling_failure_cleanly_cancels_no_input(self):
+        """If the modal can't be marshaled onto the app loop on native Windows
+        (scheduling failure) the off-main-thread path must cancel cleanly —
+        NOT fall through to a blocking raw ``input()``.
+
+        This is the degraded branch the pre-#33961 code handled with
+        ``return self._prompt_text_input(...)`` (which deadlocks); the fix
+        returns ``None`` instead.
+        """
+        cli = _make_cli()
+
+        def _raise(cb):  # call_soon_threadsafe scheduling failure
+            raise RuntimeError("event loop closed")
+
+        cli._app.loop.call_soon_threadsafe = _raise
+
+        input_called = {"n": 0}
+
+        def _tracking_input(prompt=""):
+            input_called["n"] += 1
+            time.sleep(30)
+            return "1"
+
+        outcome = {}
+
+        def _worker():
+            with patch.object(sys, "platform", "win32"), \
+                 patch("builtins.input", side_effect=_tracking_input), \
+                 patch.object(cli, "_invalidate"):
+                outcome["result"] = cli._prompt_text_input_modal(
+                    title="/new",
+                    detail="destroys conversation state",
+                    choices=_SAMPLE_CHOICES,
+                    timeout=3,
+                )
+
+        worker = threading.Thread(target=_worker, daemon=True)
+        worker.start()
+        worker.join(timeout=5.0)
+        assert not worker.is_alive(), (
+            "daemon thread hung — win32 scheduling-failure fallback used raw "
+            "input() instead of cleanly cancelling (#33961)"
+        )
+        assert input_called["n"] == 0, "win32 off-thread fallback must not call input()"
+        assert outcome.get("result") is None

tests/gateway/test_run_progress_topics.py

@@ -1264,3 +1264,123 @@ async def test_verbose_mode_respects_explicit_tool_preview_length(monkeypatch, t
     assert VerboseAgent.LONG_CODE not in all_content
     # But should still contain the truncated portion with "..."
     assert "..." in all_content
+
+
+class CodeBlockProgressAdapter(ProgressCaptureAdapter):
+    """A markdown-capable progress adapter (declares supports_code_blocks)."""
+
+    supports_code_blocks = True
+
+
+class TerminalCommandAgent:
+    """Emits a terminal tool.started with a real, multi-line command arg."""
+
+    CMD = (
+        "set -euo pipefail\n"
+        "printf 'node: '; node --version\n"
+        "npm install -g hyperframes@latest"
+    )
+
+    def __init__(self, **kwargs):
+        self.tool_progress_callback = kwargs.get("tool_progress_callback")
+        self.tools = []
+
+    def run_conversation(self, message, conversation_history=None, task_id=None):
+        self.tool_progress_callback(
+            "tool.started", "terminal", self.CMD, {"command": self.CMD}
+        )
+        # Let the async progress task drain the queue and send before returning.
+        time.sleep(0.35)
+        return {"final_response": "done", "messages": [], "api_calls": 1}
+
+
+@pytest.mark.asyncio
+async def test_terminal_progress_is_truncated_preview_not_bash_block(monkeypatch, tmp_path):
+    """Regression for #41215: terminal progress must render as a short truncated
+    preview, never the full command in a fenced ```bash block, even on a
+    markdown-capable (supports_code_blocks) gateway."""
+    monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "all")
+
+    fake_dotenv = types.ModuleType("dotenv")
+    fake_dotenv.load_dotenv = lambda *args, **kwargs: None
+    monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
+
+    fake_run_agent = types.ModuleType("run_agent")
+    fake_run_agent.AIAgent = TerminalCommandAgent
+    monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
+    import tools.terminal_tool  # noqa: F401 - register terminal emoji
+
+    adapter = CodeBlockProgressAdapter(platform=Platform.TELEGRAM)
+    runner = _make_runner(adapter)
+    gateway_run = importlib.import_module("gateway.run")
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"})
+
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="12345",
+        chat_type="dm",
+        thread_id=None,
+    )
+
+    result = await runner._run_agent(
+        message="hello",
+        context_prompt="",
+        history=[],
+        source=source,
+        session_id="sess-terminal-no-bash-block",
+        session_key="agent:main:telegram:dm:12345",
+    )
+
+    assert result["final_response"] == "done"
+    all_content = " ".join(call["content"] for call in adapter.sent)
+    all_content += " ".join(call["content"] for call in adapter.edits)
+    # Compact truncated preview, not a fenced bash block.
+    assert "```bash" not in all_content
+    assert 'terminal: "' in all_content
+    # The full multi-line command body must not reach the chat.
+    assert "npm install -g hyperframes@latest" not in all_content
+
+
+@pytest.mark.asyncio
+async def test_terminal_progress_no_bash_block_in_verbose_mode(monkeypatch, tmp_path):
+    """#41215 also rendered the bash block in verbose mode. The revert removed it
+    from both branches, so verbose progress must not emit a fenced ```bash block
+    either (verbose still shows args by opt-in, just not as a code block)."""
+    monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "verbose")
+
+    fake_dotenv = types.ModuleType("dotenv")
+    fake_dotenv.load_dotenv = lambda *args, **kwargs: None
+    monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
+
+    fake_run_agent = types.ModuleType("run_agent")
+    fake_run_agent.AIAgent = TerminalCommandAgent
+    monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
+    import tools.terminal_tool  # noqa: F401 - register terminal emoji
+
+    adapter = CodeBlockProgressAdapter(platform=Platform.TELEGRAM)
+    runner = _make_runner(adapter)
+    gateway_run = importlib.import_module("gateway.run")
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"})
+
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="12345",
+        chat_type="dm",
+        thread_id=None,
+    )
+
+    result = await runner._run_agent(
+        message="hello",
+        context_prompt="",
+        history=[],
+        source=source,
+        session_id="sess-terminal-verbose-no-bash",
+        session_key="agent:main:telegram:dm:12345",
+    )
+
+    assert result["final_response"] == "done"
+    all_content = " ".join(call["content"] for call in adapter.sent)
+    all_content += " ".join(call["content"] for call in adapter.edits)
+    assert "```bash" not in all_content

tests/gateway/test_session_api.py

@@ -75,6 +75,54 @@ async def test_capabilities_advertises_session_control_surface(adapter):
     }
 
 
+@pytest.mark.asyncio
+async def test_run_agent_binds_api_session_context_for_tool_env(adapter, monkeypatch):
+    """API-server request sessions should reach tools and terminal subprocess env."""
+    monkeypatch.setenv("HERMES_SESSION_ID", "stale-session")
+    observed = {}
+
+    class FakeAgent:
+        session_prompt_tokens = 0
+        session_completion_tokens = 0
+        session_total_tokens = 0
+
+        def __init__(self, session_id: str):
+            self.session_id = session_id
+
+        def run_conversation(self, user_message, conversation_history, task_id):
+            from gateway.session_context import get_session_env
+            from tools.environments.local import _make_run_env
+
+            observed["task_id"] = task_id
+            observed["context_session_id"] = get_session_env("HERMES_SESSION_ID")
+            observed["context_platform"] = get_session_env("HERMES_SESSION_PLATFORM")
+            observed["context_session_key"] = get_session_env("HERMES_SESSION_KEY")
+            observed["child_session_id"] = _make_run_env({}).get("HERMES_SESSION_ID")
+            return {"final_response": "ok"}
+
+    def fake_create_agent(**kwargs):
+        return FakeAgent(kwargs["session_id"])
+
+    monkeypatch.setattr(adapter, "_create_agent", fake_create_agent)
+
+    result, usage = await adapter._run_agent(
+        user_message="hello",
+        conversation_history=[],
+        session_id="request-session",
+        gateway_session_key="request-key",
+    )
+
+    assert result["session_id"] == "request-session"
+    assert usage == {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
+    assert observed == {
+        "task_id": "request-session",
+        "context_session_id": "request-session",
+        "context_platform": "api_server",
+        "context_session_key": "request-key",
+        "child_session_id": "request-session",
+    }
+
+
 @pytest.mark.asyncio
 async def test_session_crud_and_message_history(adapter, session_db):
     app = _create_session_app(adapter)

tests/gateway/test_session_env.py

@@ -190,6 +190,17 @@ def test_session_key_falls_back_to_os_environ(monkeypatch):
     assert get_session_env("HERMES_SESSION_KEY") == ""
 
 
+def test_session_id_set_via_contextvars(monkeypatch):
+    """set_session_vars should set HERMES_SESSION_ID via contextvars."""
+    monkeypatch.setenv("HERMES_SESSION_ID", "stale-env-session")
+
+    tokens = set_session_vars(session_id="ctx-session-456")
+    assert get_session_env("HERMES_SESSION_ID") == "ctx-session-456"
+
+    clear_session_vars(tokens)
+    assert get_session_env("HERMES_SESSION_ID") == ""
+
+
 def test_set_session_env_includes_session_key():
     """_set_session_env should propagate session_key from SessionContext."""
     runner = object.__new__(GatewayRunner)

tests/gateway/test_simplex_plugin.py

@@ -205,6 +205,13 @@ def test_corr_id_pending_set_self_trims():
 
 @pytest.mark.asyncio
 async def test_send_dm():
+    """DMs use the bare ``@<id> text`` chat-command form.
+
+    The bracketed form ``@[<id>] text`` is what the daemon's man page
+    documents, but in practice both addressing styles route through
+    the same chat-command parser; bare ``@<id>`` matches what every
+    Hermes deployment has been using in production for months.
+    """
     from gateway.config import PlatformConfig
     cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"})
     adapter = SimplexAdapter(cfg)
@@ -222,6 +229,14 @@ async def test_send_dm():
 
 @pytest.mark.asyncio
 async def test_send_group():
+    """Groups use the structured ``/_send #<id> json [...]`` form.
+
+    The bracket chat-command form ``#[<id>] text`` *looks* like an exact
+    ID match in the daemon docs but is parsed as a display-name lookup
+    — so messages to groups whose display name isn't literally the ID
+    silently drop. The structured ``/_send`` form addresses by numeric
+    ID and survives newlines/quoting through ``json.dumps``.
+    """
     from gateway.config import PlatformConfig
     cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"})
     adapter = SimplexAdapter(cfg)
@@ -231,7 +246,11 @@ async def test_send_group():
 
     result = await adapter.send("group:grp-99", "Hello, group!")
     payload = json.loads(mock_ws.send.call_args[0][0])
-    assert payload["cmd"] == "#[grp-99] Hello, group!"
+    assert payload["cmd"].startswith("/_send #grp-99 json ")
+    msg_content = json.loads(payload["cmd"].split(" json ", 1)[1])[0][
+        "msgContent"
+    ]
+    assert msg_content == {"type": "text", "text": "Hello, group!"}
     assert result.success is True
 
 

tests/gateway/test_telegram_format.py

@@ -835,7 +835,7 @@ class TestEditMessageStreamingSafety:
         assert second_call == {
             "chat_id": 123,
             "message_id": 456,
-            "text": "final **bold**",
+            "text": "final bold",
         }
 
     @pytest.mark.asyncio

tests/hermes_cli/test_dashboard_register.py

@@ -27,7 +27,7 @@ import hermes_cli.dashboard_register as dr
 
 
 def _ns(**kw):
-    defaults = dict(name=None, redirect_uri=None)
+    defaults = dict(name=None, redirect_uri=None, portal_url=None)
     defaults.update(kw)
     return argparse.Namespace(**defaults)
 
@@ -76,7 +76,7 @@ def _fake_http_ok(payload: dict):
 
 class TestHappyPath:
     def _run(self, *, args, account_token="tok_abc", portal="https://portal.nousresearch.com",
-             response=None, captured=None):
+             response=None, captured=None, existing_client_id=None):
         response = response or {
             "client_id": "agent:selfhost-1",
             "id": "selfhost-1",
@@ -98,12 +98,21 @@ class TestHappyPath:
         def fake_save(key, value):
             saved[key] = value
 
+        # get_env_value is consulted twice: once for the stored client_id
+        # (idempotency key) and once for HERMES_DASHBOARD_PORTAL_URL. Route by
+        # key so a test can seed a prior client_id while keeping the portal
+        # unset (the default-portal-not-persisted path).
+        def fake_get_env(key):
+            if key == "HERMES_DASHBOARD_OAUTH_CLIENT_ID":
+                return existing_client_id
+            return None
+
         with patch(
             "hermes_cli.auth.resolve_nous_access_token", return_value=account_token
         ), patch("hermes_cli.config.is_managed", return_value=False), patch.object(
             dr, "_resolve_portal_base_url", return_value=portal
         ), patch(
-            "hermes_cli.config.get_env_value", return_value=None
+            "hermes_cli.config.get_env_value", side_effect=fake_get_env
         ), patch(
             "hermes_cli.config.save_env_value", side_effect=fake_save
         ), patch.object(
@@ -157,6 +166,394 @@ class TestHappyPath:
         )
 
 
+class TestIdempotentRerun(TestHappyPath):
+    """Re-running with a stored client_id updates instead of creating.
+
+    Inherits ``_run`` from TestHappyPath; the only new lever is
+    ``existing_client_id`` (the HERMES_DASHBOARD_OAUTH_CLIENT_ID a prior run
+    persisted), which the CLI re-sends so the portal updates that row.
+    """
+
+    def test_stored_client_id_is_sent_as_idempotency_key(self, capsys):
+        captured: dict = {}
+        # Portal echoes back the SAME id -> it updated in place.
+        self._run(
+            args=_ns(),
+            existing_client_id="agent:selfhost-1",
+            response={
+                "client_id": "agent:selfhost-1",
+                "id": "selfhost-1",
+                "name": "dreamy_tesla",
+                "kind": "SELF_HOSTED",
+                "custom_redirect_uri": None,
+                "created_at": "2026-06-04T12:00:00.000Z",
+            },
+            captured=captured,
+        )
+        assert captured["body"]["client_id"] == "agent:selfhost-1"
+
+    def test_rerun_without_name_omits_name_to_preserve_stored(self, capsys):
+        # No --name on a re-run: don't churn the portal-stored name. The CLI
+        # leaves `name` out of the body so the portal keeps what it has.
+        captured: dict = {}
+        self._run(
+            args=_ns(),
+            existing_client_id="agent:selfhost-1",
+            captured=captured,
+        )
+        assert "name" not in captured["body"]
+        assert captured["body"]["client_id"] == "agent:selfhost-1"
+
+    def test_rerun_with_explicit_name_still_sends_name(self, capsys):
+        captured: dict = {}
+        self._run(
+            args=_ns(name="renamed_box"),
+            existing_client_id="agent:selfhost-1",
+            captured=captured,
+        )
+        assert captured["body"]["name"] == "renamed_box"
+        assert captured["body"]["client_id"] == "agent:selfhost-1"
+
+    def test_rerun_prints_updated_when_same_id_returned(self, capsys):
+        self._run(
+            args=_ns(),
+            existing_client_id="agent:selfhost-1",
+            response={
+                "client_id": "agent:selfhost-1",
+                "id": "selfhost-1",
+                "name": "dreamy_tesla",
+                "kind": "SELF_HOSTED",
+                "custom_redirect_uri": None,
+                "created_at": "2026-06-04T12:00:00.000Z",
+            },
+        )
+        out = capsys.readouterr().out
+        assert "Updated dashboard" in out
+        assert "Registered dashboard" not in out
+
+    def test_rerun_persists_returned_client_id(self, capsys):
+        saved = self._run(
+            args=_ns(),
+            existing_client_id="agent:selfhost-1",
+        )
+        # Same id round-trips into .env -> idempotent, one record.
+        assert saved["HERMES_DASHBOARD_OAUTH_CLIENT_ID"] == "agent:selfhost-1"
+
+    def test_stale_id_falls_through_to_create_prints_registered(self, capsys):
+        # Stored id no longer resolves server-side -> portal created a fresh
+        # row and returns a DIFFERENT id. The CLI treats that as a create and
+        # persists the new id (re-run stays safe, never worse than first run).
+        captured: dict = {}
+        saved = self._run(
+            args=_ns(name="seed_name"),
+            existing_client_id="agent:selfhost-stale",
+            response={
+                "client_id": "agent:selfhost-new",
+                "id": "selfhost-new",
+                "name": "seed_name",
+                "kind": "SELF_HOSTED",
+                "custom_redirect_uri": None,
+                "created_at": "2026-06-04T12:00:00.000Z",
+            },
+            captured=captured,
+        )
+        # The stale id is still SENT (portal decides create-vs-update).
+        assert captured["body"]["client_id"] == "agent:selfhost-stale"
+        # Returned id differs from what we sent -> message is "Registered".
+        out = capsys.readouterr().out
+        assert "Registered dashboard" in out
+        assert "Updated dashboard" not in out
+        assert saved["HERMES_DASHBOARD_OAUTH_CLIENT_ID"] == "agent:selfhost-new"
+
+    def test_blank_stored_client_id_treated_as_first_run(self, capsys):
+        # A blank/whitespace stored value is not a usable key: treat as a
+        # first registration (auto-generate a name, don't send client_id).
+        captured: dict = {}
+        self._run(
+            args=_ns(),
+            existing_client_id="   ",
+            captured=captured,
+        )
+        assert "client_id" not in captured["body"]
+        assert captured["body"].get("name")  # auto-generated
+
+
+class TestCustomPortalPersistence:
+    """`--portal-url` / HERMES_DASHBOARD_PORTAL_URL is persisted to .env.
+
+    An *explicitly supplied* custom portal URL is an intentional choice the
+    user wants to survive across sessions, so it's always written (updating an
+    existing entry in place rather than appending a duplicate). When no custom
+    URL is supplied, the older conservative behaviour is preserved: an inferred
+    portal is only written when absent and non-default, and an existing entry
+    is never altered unexpectedly.
+    """
+
+    def _run(self, *, args, portal, existing_portal):
+        """Drive cmd_dashboard_register, capturing save_env_value calls.
+
+        `existing_portal` is what get_env_value returns for
+        HERMES_DASHBOARD_PORTAL_URL (None = not present in .env).
+        """
+        response = {
+            "client_id": "agent:selfhost-1",
+            "id": "selfhost-1",
+            "name": "dreamy_tesla",
+            "kind": "SELF_HOSTED",
+            "custom_redirect_uri": None,
+            "created_at": "2026-06-04T12:00:00.000Z",
+        }
+
+        saved: dict = {}
+
+        def fake_save(key, value):
+            saved[key] = value
+
+        def fake_get_env_value(key, *a, **kw):
+            if key == "HERMES_DASHBOARD_PORTAL_URL":
+                return existing_portal
+            return None
+
+        with patch(
+            "hermes_cli.auth.resolve_nous_access_token", return_value="tok"
+        ), patch("hermes_cli.config.is_managed", return_value=False), patch.dict(
+            dr.os.environ, {}, clear=False
+        ), patch.object(
+            dr, "_resolve_portal_base_url", return_value=portal
+        ), patch(
+            "hermes_cli.config.get_env_value", side_effect=fake_get_env_value
+        ), patch(
+            "hermes_cli.config.save_env_value", side_effect=fake_save
+        ), patch.object(
+            dr.urllib.request, "urlopen", return_value=_fake_http_ok(response)
+        ):
+            # The ambient process env may carry HERMES_DASHBOARD_PORTAL_URL
+            # (e.g. staging dev shells); drop it so `custom_portal_supplied`
+            # is driven solely by the args.portal_url under test.
+            dr.os.environ.pop("HERMES_DASHBOARD_PORTAL_URL", None)
+            dr.cmd_dashboard_register(args)
+        return saved
+
+    def test_explicit_custom_url_persisted_when_var_absent(self, capsys):
+        saved = self._run(
+            args=_ns(portal_url="https://preview.example.com"),
+            portal="https://preview.example.com",
+            existing_portal=None,
+        )
+        assert saved["HERMES_DASHBOARD_PORTAL_URL"] == "https://preview.example.com"
+
+    def test_explicit_custom_url_updates_existing_in_place(self, capsys):
+        # An entry already exists with a different value; the explicit custom
+        # URL overwrites it (save_env_value updates the matching key in place).
+        saved = self._run(
+            args=_ns(portal_url="https://new-preview.example.com"),
+            portal="https://new-preview.example.com",
+            existing_portal="https://old-preview.example.com",
+        )
+        assert (
+            saved["HERMES_DASHBOARD_PORTAL_URL"] == "https://new-preview.example.com"
+        )
+
+    def test_explicit_custom_url_persisted_even_when_equals_default(self, capsys):
+        # User explicitly asked for the production portal — honour the explicit
+        # request and persist it (the no-flag path would skip the default).
+        saved = self._run(
+            args=_ns(portal_url="https://portal.nousresearch.com"),
+            portal="https://portal.nousresearch.com",
+            existing_portal=None,
+        )
+        assert (
+            saved["HERMES_DASHBOARD_PORTAL_URL"] == "https://portal.nousresearch.com"
+        )
+
+    def test_explicit_custom_url_equal_to_existing_is_noop(self, capsys):
+        # Already persisted with the same value → no redundant write.
+        saved = self._run(
+            args=_ns(portal_url="https://preview.example.com"),
+            portal="https://preview.example.com",
+            existing_portal="https://preview.example.com",
+        )
+        assert "HERMES_DASHBOARD_PORTAL_URL" not in saved
+
+    def test_no_flag_default_portal_not_written(self, capsys):
+        # No custom URL supplied, resolves to default → not written.
+        saved = self._run(
+            args=_ns(),
+            portal="https://portal.nousresearch.com",
+            existing_portal=None,
+        )
+        assert "HERMES_DASHBOARD_PORTAL_URL" not in saved
+
+    def test_no_flag_does_not_overwrite_existing_entry(self, capsys):
+        # No custom URL supplied and the var already exists → left untouched,
+        # even if the inferred portal differs (acceptance criterion 4).
+        saved = self._run(
+            args=_ns(),
+            portal="https://inferred-from-login.example.com",
+            existing_portal="https://already-set.example.com",
+        )
+        assert "HERMES_DASHBOARD_PORTAL_URL" not in saved
+
+
+class TestPublicUrlPersistence:
+    """`--redirect-uri` derives & persists HERMES_DASHBOARD_PUBLIC_URL in .env.
+
+    --redirect-uri is the full public callback (e.g.
+    https://hermes.example.com/auth/callback). At serve time the dashboard auth
+    layer reconstructs that callback by appending "/auth/callback" to
+    HERMES_DASHBOARD_PUBLIC_URL, so the value that's actually consumed is the
+    ORIGIN (scheme://host). We derive the origin from the supplied redirect URI
+    and persist THAT as HERMES_DASHBOARD_PUBLIC_URL — the var the runtime reads
+    — so the public-URL override is genuinely wired, not just stored.
+
+    An explicitly supplied value is always written (updating an existing entry
+    in place rather than appending a duplicate); a no-op when it already
+    matches; and never written on a localhost-only install (no --redirect-uri).
+    """
+
+    def _run(self, *, args, existing_public=None):
+        """Drive cmd_dashboard_register, capturing save_env_value calls.
+
+        `existing_public` is what get_env_value returns for
+        HERMES_DASHBOARD_PUBLIC_URL (None = not present in .env).
+        """
+        response = {
+            "client_id": "agent:selfhost-1",
+            "id": "selfhost-1",
+            "name": "dreamy_tesla",
+            "kind": "SELF_HOSTED",
+            "custom_redirect_uri": getattr(args, "redirect_uri", None),
+            "created_at": "2026-06-04T12:00:00.000Z",
+        }
+
+        saved: dict = {}
+
+        def fake_save(key, value):
+            saved[key] = value
+
+        def fake_get_env_value(key, *a, **kw):
+            if key == "HERMES_DASHBOARD_PUBLIC_URL":
+                return existing_public
+            return None
+
+        with patch(
+            "hermes_cli.auth.resolve_nous_access_token", return_value="tok"
+        ), patch("hermes_cli.config.is_managed", return_value=False), patch.dict(
+            dr.os.environ, {}, clear=False
+        ), patch.object(
+            dr, "_resolve_portal_base_url", return_value="https://portal.nousresearch.com"
+        ), patch(
+            "hermes_cli.config.get_env_value", side_effect=fake_get_env_value
+        ), patch(
+            "hermes_cli.config.save_env_value", side_effect=fake_save
+        ), patch.object(
+            dr.urllib.request, "urlopen", return_value=_fake_http_ok(response)
+        ):
+            dr.os.environ.pop("HERMES_DASHBOARD_PORTAL_URL", None)
+            dr.cmd_dashboard_register(args)
+        return saved
+
+    def test_origin_derived_from_full_callback_path(self, capsys):
+        # The key behaviour: a full callback URL is reduced to its ORIGIN so
+        # the runtime's "public_url + /auth/callback" reconstruction matches.
+        saved = self._run(
+            args=_ns(redirect_uri="https://hermes.example.com/auth/callback"),
+            existing_public=None,
+        )
+        assert saved["HERMES_DASHBOARD_PUBLIC_URL"] == "https://hermes.example.com"
+        # The full callback path must NOT be persisted verbatim (would double
+        # the path at serve time).
+        assert "/auth/callback" not in saved["HERMES_DASHBOARD_PUBLIC_URL"]
+
+    def test_origin_preserves_port(self, capsys):
+        saved = self._run(
+            args=_ns(redirect_uri="https://hermes.example.com:8443/auth/callback"),
+            existing_public=None,
+        )
+        assert saved["HERMES_DASHBOARD_PUBLIC_URL"] == "https://hermes.example.com:8443"
+
+    def test_public_url_updates_existing_in_place(self, capsys):
+        # A stale public-url entry exists; the new derived origin overwrites it.
+        saved = self._run(
+            args=_ns(redirect_uri="https://new.example.com/auth/callback"),
+            existing_public="https://old.example.com",
+        )
+        assert saved["HERMES_DASHBOARD_PUBLIC_URL"] == "https://new.example.com"
+
+    def test_public_url_equal_to_existing_is_noop(self, capsys):
+        # Derived origin already matches what's stored → no redundant write.
+        saved = self._run(
+            args=_ns(redirect_uri="https://hermes.example.com/auth/callback"),
+            existing_public="https://hermes.example.com",
+        )
+        assert "HERMES_DASHBOARD_PUBLIC_URL" not in saved
+
+    def test_no_redirect_flag_not_written(self, capsys):
+        # Localhost-only install (no --redirect-uri) → var left untouched.
+        saved = self._run(
+            args=_ns(),
+            existing_public=None,
+        )
+        assert "HERMES_DASHBOARD_PUBLIC_URL" not in saved
+
+    def test_no_redirect_flag_does_not_overwrite_existing(self, capsys):
+        # No --redirect-uri supplied but a value already exists → never touch
+        # it (an existing entry is only changed by an explicit new value).
+        saved = self._run(
+            args=_ns(),
+            existing_public="https://already-set.example.com",
+        )
+        assert "HERMES_DASHBOARD_PUBLIC_URL" not in saved
+
+    def test_non_http_redirect_not_persisted(self, capsys):
+        # A malformed / non-http(s) redirect yields no derivable origin → skip.
+        saved = self._run(
+            args=_ns(redirect_uri="not-a-url"),
+            existing_public=None,
+        )
+        assert "HERMES_DASHBOARD_PUBLIC_URL" not in saved
+
+    def test_public_url_persisted_alongside_portal_url(self, capsys):
+        # Both --portal-url and --redirect-uri supplied → portal_url AND the
+        # derived public_url are both persisted (ADD semantics: the public-url
+        # write does not displace portal-url persistence).
+        response = {
+            "client_id": "agent:selfhost-1",
+            "id": "selfhost-1",
+            "name": "dreamy_tesla",
+            "kind": "SELF_HOSTED",
+            "custom_redirect_uri": "https://hermes.example.com/auth/callback",
+            "created_at": "2026-06-04T12:00:00.000Z",
+        }
+        saved: dict = {}
+
+        def fake_save(key, value):
+            saved[key] = value
+
+        with patch(
+            "hermes_cli.auth.resolve_nous_access_token", return_value="tok"
+        ), patch("hermes_cli.config.is_managed", return_value=False), patch.dict(
+            dr.os.environ, {}, clear=False
+        ), patch.object(
+            dr, "_resolve_portal_base_url", return_value="https://preview.example.com"
+        ), patch(
+            "hermes_cli.config.get_env_value", return_value=None
+        ), patch(
+            "hermes_cli.config.save_env_value", side_effect=fake_save
+        ), patch.object(
+            dr.urllib.request, "urlopen", return_value=_fake_http_ok(response)
+        ):
+            dr.os.environ.pop("HERMES_DASHBOARD_PORTAL_URL", None)
+            dr.cmd_dashboard_register(
+                _ns(
+                    portal_url="https://preview.example.com",
+                    redirect_uri="https://hermes.example.com/auth/callback",
+                )
+            )
+        assert saved["HERMES_DASHBOARD_PORTAL_URL"] == "https://preview.example.com"
+        assert saved["HERMES_DASHBOARD_PUBLIC_URL"] == "https://hermes.example.com"
+
+
 class TestPortalResolution:
     def test_override_arg_wins(self):
         assert (

tests/hermes_cli/test_doctor.py

@@ -540,6 +540,54 @@ def test_run_doctor_accepts_hermes_provider_ids_that_catalog_aliases(
         )
 
 
+def test_run_doctor_accepts_vendor_slugs_for_named_custom_provider(monkeypatch, tmp_path):
+    home = tmp_path / ".hermes"
+    home.mkdir(parents=True, exist_ok=True)
+    (home / "config.yaml").write_text(
+        "model:\n"
+        "  provider: custom:hpc-ai\n"
+        "  default: deepseek/deepseek-v4-flash\n"
+        "custom_providers:\n"
+        "  - name: hpc-ai\n"
+        "    base_url: https://hpc-ai.example/v1\n"
+        "    api_key: test-key\n",
+        encoding="utf-8",
+    )
+
+    monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
+    monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", tmp_path / "project")
+    monkeypatch.setattr(doctor_mod, "_DHH", str(home))
+    (tmp_path / "project").mkdir(exist_ok=True)
+
+    fake_model_tools = types.SimpleNamespace(
+        check_tool_availability=lambda *a, **kw: ([], []),
+        TOOLSET_REQUIREMENTS={},
+    )
+    monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
+
+    try:
+        from hermes_cli import auth as _auth_mod
+        monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
+        monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
+        monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {})
+    except Exception:
+        pass
+
+    buf = io.StringIO()
+    with contextlib.redirect_stdout(buf):
+        doctor_mod.run_doctor(Namespace(fix=False))
+
+    out = buf.getvalue()
+    assert "model.provider 'custom:hpc-ai' is not a recognised provider" not in out
+    assert "model.provider 'custom:hpc-ai' is unknown" not in out
+    assert (
+        "model.default 'deepseek/deepseek-v4-flash' uses a vendor/model slug but provider is "
+        "'custom:hpc-ai'"
+        not in out
+    )
+    assert "Either set model.provider to 'openrouter', or drop the vendor prefix." not in out
+
+
 
 
 def test_run_doctor_accepts_kimi_coding_cn_provider(monkeypatch, tmp_path):

tests/hermes_cli/test_model_switch_custom_providers.py

@@ -65,6 +65,15 @@ def test_resolve_provider_full_finds_named_custom_provider():
     assert resolved.source == "user-config"
 
 
+def test_is_aggregator_recognizes_named_custom_provider():
+    assert providers_mod.is_aggregator("custom:hpc-ai") is True
+    assert providers_mod.is_aggregator("custom:litellm") is True
+
+
+def test_is_aggregator_leaves_unknown_provider_non_aggregator():
+    assert providers_mod.is_aggregator("not-a-provider") is False
+
+
 def test_switch_model_accepts_explicit_named_custom_provider(monkeypatch):
     """Shared /model switch pipeline should accept --provider for custom_providers."""
     monkeypatch.setattr(

tests/plugins/platforms/photon/test_auth.py

@@ -1,8 +1,8 @@
-"""Tests for the Photon auth module (device login + project + user creation)."""
+"""Tests for the Photon auth module (device login + dashboard API)."""
 from __future__ import annotations
 
 import json
-import time
+import os
 from pathlib import Path
 from typing import Any, Dict
 
@@ -36,51 +36,91 @@ class _FakeResponse:
             raise RuntimeError(f"HTTP {self.status_code}")
 
 
+_PHOTON_ENV = (
+    "PHOTON_PROJECT_ID",
+    "PHOTON_PROJECT_SECRET",
+    "PHOTON_DASHBOARD_PROJECT_ID",
+)
+
+
 @pytest.fixture
-def tmp_hermes_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+def tmp_hermes_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
     home = tmp_path / "hermes"
     home.mkdir()
     monkeypatch.setenv("HERMES_HOME", str(home))
-    # The auth module memoises by reading get_hermes_home at call time
-    # so the env var is what matters.
-    return home
+    for key in _PHOTON_ENV:
+        monkeypatch.delenv(key, raising=False)
+    yield home
+    # save_env_value() mutates os.environ directly, so scrub any leakage.
+    for key in _PHOTON_ENV:
+        os.environ.pop(key, None)
 
 
+# ---------------------------------------------------------------------------
+# Credential storage
+
 def test_store_and_load_photon_token(tmp_hermes_home: Path) -> None:
     photon_auth.store_photon_token("abc123def456")
     assert photon_auth.load_photon_token() == "abc123def456"
 
     auth_json = json.loads((tmp_hermes_home / "auth.json").read_text())
-    assert "credential_pool" in auth_json
     assert auth_json["credential_pool"]["photon"][0]["access_token"] == "abc123def456"
 
 
-def test_store_and_load_project_credentials(tmp_hermes_home: Path) -> None:
+def test_store_project_credentials_round_trip(
+    tmp_hermes_home: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Don't touch .env / os.environ here — exercise the auth.json path.
+    monkeypatch.setattr(photon_auth, "_persist_runtime_env", lambda *a, **k: None)
     photon_auth.store_project_credentials(
-        "proj-uuid", "secret-key", name="Test Project",
+        spectrum_project_id="sp-123",
+        project_secret="secret-key",
+        dashboard_project_id="dash-456",
+        name="Hermes Agent",
     )
-    pid, secret = photon_auth.load_project_credentials()
-    assert pid == "proj-uuid"
+    for key in _PHOTON_ENV:
+        monkeypatch.delenv(key, raising=False)
+
+    sid, secret = photon_auth.load_project_credentials()
+    assert sid == "sp-123"
     assert secret == "secret-key"
+    assert photon_auth.load_dashboard_project_id() == "dash-456"
+
+
+def test_store_project_credentials_writes_env(tmp_hermes_home: Path) -> None:
+    photon_auth.store_project_credentials(
+        spectrum_project_id="sp-789",
+        project_secret="sek-ret",
+        dashboard_project_id="dash-1",
+    )
+    env_text = (tmp_hermes_home / ".env").read_text()
+    assert "PHOTON_PROJECT_ID=sp-789" in env_text
+    assert "PHOTON_PROJECT_SECRET=sek-ret" in env_text
 
 
 def test_load_project_credentials_env_override(
     tmp_hermes_home: Path, monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    photon_auth.store_project_credentials("from-file", "secret-file")
+    monkeypatch.setattr(photon_auth, "_persist_runtime_env", lambda *a, **k: None)
+    photon_auth.store_project_credentials(
+        spectrum_project_id="from-file", project_secret="secret-file",
+    )
     monkeypatch.setenv("PHOTON_PROJECT_ID", "from-env")
     monkeypatch.setenv("PHOTON_PROJECT_SECRET", "secret-env")
-    pid, secret = photon_auth.load_project_credentials()
-    assert pid == "from-env"
+    sid, secret = photon_auth.load_project_credentials()
+    assert sid == "from-env"
     assert secret == "secret-env"
 
 
-def test_request_device_code(monkeypatch: pytest.MonkeyPatch) -> None:
+# ---------------------------------------------------------------------------
+# Device login flow
+
+def test_request_device_code_uses_photon_cli(monkeypatch: pytest.MonkeyPatch) -> None:
     captured: Dict[str, Any] = {}
 
-    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
         captured["url"] = url
-        captured["body"] = json
+        captured["body"] = kwargs.get("json")
         return _FakeResponse(json_body={
             "device_code": "dev-code-xyz",
             "user_code": "ABCD-1234",
@@ -95,189 +135,408 @@ def test_request_device_code(monkeypatch: pytest.MonkeyPatch) -> None:
     code = photon_auth.request_device_code()
     assert code.device_code == "dev-code-xyz"
     assert code.user_code == "ABCD-1234"
-    assert code.expires_in == 600
     assert "/api/auth/device/code" in captured["url"]
-    assert captured["body"]["client_id"] == "hermes-agent"
+    # Hosted Photon allowlists registered device clients — an unregistered
+    # client_id is rejected with 400 invalid_client. We use Photon's published
+    # CLI device client and send the standard scope.
+    assert captured["body"]["client_id"] == "photon-cli"
+    assert captured["body"]["scope"] == "openid profile email"
 
 
-def test_poll_for_token_via_header(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Token from set-auth-token header is the documented mechanism."""
-
-    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
-        return _FakeResponse(
-            status=200,
-            json_body={"session": {}, "user": {}},
-            headers={"set-auth-token": "bearer-xyz"},
-        )
-
-    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
-
-    code = photon_auth.DeviceCode(
+def _device_code() -> "photon_auth.DeviceCode":
+    return photon_auth.DeviceCode(
         device_code="d", user_code="u",
         verification_uri="https://x", verification_uri_complete=None,
         expires_in=10, interval=0,
     )
-    token = photon_auth.poll_for_token(code, interval=0, timeout=2)
-    assert token == "bearer-xyz"
 
 
-def test_poll_for_token_via_body_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
-    """If the header is absent we fall back to session.access_token."""
-
-    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
-        return _FakeResponse(
-            status=200,
-            json_body={"session": {"access_token": "from-body"}, "user": {}},
-        )
+def test_poll_for_token_body_access_token(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(status=200, json_body={"access_token": "tok-body"})
 
     monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
-    code = photon_auth.DeviceCode(
-        device_code="d", user_code="u",
-        verification_uri="https://x", verification_uri_complete=None,
-        expires_in=10, interval=0,
-    )
-    assert photon_auth.poll_for_token(code, interval=0, timeout=2) == "from-body"
+    assert photon_auth.poll_for_token(_device_code(), interval=0, timeout=2) == "tok-body"
 
 
-def test_poll_for_token_propagates_access_denied(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
-        return _FakeResponse(
-            status=400, json_body={"error": "access_denied"},
-        )
+def test_poll_for_token_session_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(status=200, json_body={"session": {"access_token": "tok-sess"}})
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    assert photon_auth.poll_for_token(_device_code(), interval=0, timeout=2) == "tok-sess"
+
+
+def test_poll_for_token_header_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(status=200, json_body={}, headers={"set-auth-token": "tok-hdr"})
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    assert photon_auth.poll_for_token(_device_code(), interval=0, timeout=2) == "tok-hdr"
+
+
+def test_poll_for_token_pending_then_success(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = {"n": 0}
+
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return _FakeResponse(status=400, json_body={"error": "authorization_pending"})
+        return _FakeResponse(status=200, json_body={"access_token": "tok-eventual"})
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    assert photon_auth.poll_for_token(_device_code(), interval=0, timeout=5) == "tok-eventual"
+    assert calls["n"] == 2
+
+
+def test_poll_for_token_access_denied(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(status=400, json_body={"error": "access_denied"})
 
     monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
-    code = photon_auth.DeviceCode(
-        device_code="d", user_code="u",
-        verification_uri="https://x", verification_uri_complete=None,
-        expires_in=10, interval=0,
-    )
     with pytest.raises(RuntimeError, match="access_denied"):
-        photon_auth.poll_for_token(code, interval=0, timeout=2)
+        photon_auth.poll_for_token(_device_code(), interval=0, timeout=2)
 
 
+# ---------------------------------------------------------------------------
+# Projects
+
+def test_list_projects_unwraps_list(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_get(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(json_body=[{"id": "p1", "name": "Hermes Agent"}])
+
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+    projects = photon_auth.list_projects("tok")
+    assert projects[0]["id"] == "p1"
+
+
+def test_find_project_by_name_case_insensitive(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_get(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(json_body={"data": [
+            {"id": "p1", "name": "Other"},
+            {"id": "p2", "name": "hermes agent"},
+        ]})
+
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+    proj = photon_auth.find_project_by_name("tok", "Hermes Agent")
+    assert proj is not None and proj["id"] == "p2"
+
+
+def test_create_project_sends_spectrum_true(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: Dict[str, Any] = {}
+
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        captured["url"] = url
+        captured["body"] = kwargs.get("json")
+        captured["headers"] = kwargs.get("headers")
+        return _FakeResponse(json_body={"success": True, "id": "new-proj"})
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    data = photon_auth.create_project("tok", name="Hermes Agent")
+    assert data["id"] == "new-proj"
+    assert captured["body"]["spectrum"] is True
+    assert captured["body"]["name"] == "Hermes Agent"
+    assert captured["headers"]["Authorization"] == "Bearer tok"
+    assert captured["url"].endswith("/api/projects")
+
+
+def test_create_project_raises_without_id(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(json_body={"success": True})
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    with pytest.raises(RuntimeError, match="project id"):
+        photon_auth.create_project("tok")
+
+
+def test_ensure_spectrum_enabled_toggles_when_off(monkeypatch: pytest.MonkeyPatch) -> None:
+    get_calls = {"n": 0}
+    posted = {"toggle": False}
+
+    def fake_get(url: str, **kwargs: Any) -> _FakeResponse:
+        get_calls["n"] += 1
+        if get_calls["n"] == 1:
+            return _FakeResponse(json_body={"id": "p", "spectrum": False, "spectrumProjectId": None})
+        return _FakeResponse(json_body={"id": "p", "spectrum": True, "spectrumProjectId": "sp-1"})
+
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        if url.endswith("/spectrum/toggle"):
+            posted["toggle"] = True
+        return _FakeResponse(json_body={"success": True})
+
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    proj = photon_auth.ensure_spectrum_enabled("tok", "p")
+    assert posted["toggle"] is True
+    assert proj["spectrumProjectId"] == "sp-1"
+
+
+def test_ensure_spectrum_enabled_skips_toggle_when_on(monkeypatch: pytest.MonkeyPatch) -> None:
+    posted = {"toggle": False}
+
+    def fake_get(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(json_body={"id": "p", "spectrum": True, "spectrumProjectId": "sp-1"})
+
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        if url.endswith("/spectrum/toggle"):
+            posted["toggle"] = True
+        return _FakeResponse(json_body={"success": True})
+
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    proj = photon_auth.ensure_spectrum_enabled("tok", "p")
+    assert posted["toggle"] is False
+    assert proj["spectrumProjectId"] == "sp-1"
+
+
+def test_regenerate_project_secret(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        assert url.endswith("/regenerate-secret")
+        return _FakeResponse(json_body={"success": True, "projectSecret": "rotated"})
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    assert photon_auth.regenerate_project_secret("tok", "p") == "rotated"
+
+
+# ---------------------------------------------------------------------------
+# Users
+
 def test_create_user_rejects_invalid_phone() -> None:
     with pytest.raises(ValueError, match="E.164"):
-        photon_auth.create_user(
-            "proj", "secret", phone_number="not-a-number",
-        )
+        photon_auth.create_user("tok", "proj", phone_number="not-a-number")
 
 
-def test_create_user_posts_shared_type(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_create_user_posts_dashboard_shape(monkeypatch: pytest.MonkeyPatch) -> None:
     captured: Dict[str, Any] = {}
 
-    def fake_post(url: str, *, json: Dict[str, Any], auth: tuple, timeout: float) -> _FakeResponse:
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
         captured["url"] = url
-        captured["body"] = json
-        captured["auth"] = auth
-        return _FakeResponse(json_body={
-            "succeed": True,
-            "data": {
-                "id": "user-uuid",
-                "phoneNumber": "+15551234567",
-                "assignedPhoneNumber": "+15559999999",
-            },
-        })
+        captured["body"] = kwargs.get("json")
+        captured["headers"] = kwargs.get("headers")
+        return _FakeResponse(json_body={"success": True, "user": {
+            "id": "user-uuid", "phoneNumber": "+15551234567",
+        }})
 
     monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
-    user = photon_auth.create_user(
-        "proj-id", "proj-secret",
-        phone_number="+15551234567",
-    )
-    assert user["assignedPhoneNumber"] == "+15559999999"
-    assert captured["auth"] == ("proj-id", "proj-secret")
-    assert captured["body"]["type"] == "shared"
+    user = photon_auth.create_user("tok", "proj-id", phone_number="+15551234567")
+    assert user["id"] == "user-uuid"
     assert captured["body"]["phoneNumber"] == "+15551234567"
-    assert "/projects/proj-id/users/" in captured["url"]
+    assert captured["headers"]["Authorization"] == "Bearer tok"
+    assert "/projects/proj-id/spectrum/users" in captured["url"]
 
 
-def test_register_webhook_surfaces_secret(monkeypatch: pytest.MonkeyPatch) -> None:
-    def fake_post(url: str, *, json: Dict[str, Any], auth: tuple, timeout: float) -> _FakeResponse:
-        return _FakeResponse(json_body={
-            "succeed": True,
-            "data": {
-                "id": "wh-uuid",
-                "webhookUrl": json["webhookUrl"],
-                "signingSecret": "0" * 64,
-            },
-        })
+def test_register_user_if_absent_dedup(monkeypatch: pytest.MonkeyPatch) -> None:
+    posted = {"n": 0}
 
+    def fake_get(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(json_body=[{
+            "id": "u1",
+            "phoneNumber": "+1 (555) 123-4567",
+            "assignedPhoneNumber": "+16282679185",
+        }])
+
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        posted["n"] += 1
+        return _FakeResponse(json_body={"success": True, "user": {}})
+
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
     monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
-    data = photon_auth.register_webhook(
-        "proj", "secret", webhook_url="https://x.example.com/hook",
+    # Same number, different formatting — should match and NOT create.
+    user, created = photon_auth.register_user_if_absent(
+        "tok", "proj", phone_number="+15551234567",
     )
-    assert data["signingSecret"] == "0" * 64
-    assert data["webhookUrl"] == "https://x.example.com/hook"
+    assert created is False
+    assert user["id"] == "u1"
+    assert posted["n"] == 0
+    # The reused user carries the assigned iMessage line ("TEXTS ON").
+    assert photon_auth.user_assigned_line(user) == "+16282679185"
 
 
-def test_persist_webhook_signing_secret_writes_env(
-    tmp_hermes_home: Path,
-) -> None:
-    """The helper hands the secret to save_env_value, never returns it."""
-    summary: list = []
-    response = {
-        "id": "wh-uuid",
-        "webhookUrl": "https://x.example.com/hook",
-        "signingSecret": "ABCDEF1234567890" * 4,
-    }
-    ok = photon_auth.persist_webhook_signing_secret(
-        response, on_summary=summary.append,
+def test_user_assigned_line() -> None:
+    assert (
+        photon_auth.user_assigned_line({"assignedPhoneNumber": "+16282679185"})
+        == "+16282679185"
     )
-
-    assert ok is True
-    env_path = tmp_hermes_home / ".env"
-    assert env_path.exists()
-    env_text = env_path.read_text()
-    assert "PHOTON_WEBHOOK_SECRET=ABCDEF1234567890" in env_text
-    # The on_summary callback gets the redacted response + a saved-to path;
-    # none of those strings should leak the raw secret.
-    joined = "\n".join(summary)
-    assert "<redacted>" in joined
-    assert "ABCDEF1234567890" not in joined
+    # Own number present but no assignment yet (e.g. freshly created user).
+    assert photon_auth.user_assigned_line({"phoneNumber": "+15551234567"}) is None
+    assert photon_auth.user_assigned_line({"assignedPhoneNumber": ""}) is None
+    assert photon_auth.user_assigned_line({}) is None
+    assert photon_auth.user_assigned_line(None) is None
 
 
-def test_persist_webhook_signing_secret_no_secret_no_write(
-    tmp_hermes_home: Path,
-) -> None:
-    summary: list = []
-    ok = photon_auth.persist_webhook_signing_secret(
-        {"id": "wh-uuid", "webhookUrl": "https://x"},
-        on_summary=summary.append,
+def test_register_user_if_absent_creates(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_get(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(json_body=[])
+
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(json_body={"success": True, "user": {"id": "u-new"}})
+
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    user, created = photon_auth.register_user_if_absent(
+        "tok", "proj", phone_number="+15551234567",
     )
-    assert ok is False
-    # No env file written; summary callback still received the redacted
-    # response (without a signingSecret key, nothing to redact).
-    assert not (tmp_hermes_home / ".env").exists()
+    assert created is True
+    assert user["id"] == "u-new"
 
 
-def test_credential_summary_returns_only_display_strings(
-    tmp_hermes_home: Path,
+# ---------------------------------------------------------------------------
+# Lines (assigned number)
+
+def test_get_imessage_line_returns_existing(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_get(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(json_body=[
+            {"id": "l1", "platform": "imessage", "phoneNumber": "+15559999999", "status": "active"},
+        ])
+
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+    line = photon_auth.get_imessage_line("tok", "proj")
+    assert line is not None and line["phoneNumber"] == "+15559999999"
+
+
+def test_get_imessage_line_provisions_when_missing(monkeypatch: pytest.MonkeyPatch) -> None:
+    added = {"n": 0}
+
+    def fake_get(url: str, **kwargs: Any) -> _FakeResponse:
+        return _FakeResponse(json_body=[])
+
+    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+        added["n"] += 1
+        assert kwargs.get("json", {}).get("platform") == "imessage"
+        return _FakeResponse(json_body={"success": True, "line": {
+            "id": "l-new", "platform": "imessage", "phoneNumber": "+15558888888",
+        }})
+
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    line = photon_auth.get_imessage_line("tok", "proj")
+    assert added["n"] == 1
+    assert line["phoneNumber"] == "+15558888888"
+
+
+# ---------------------------------------------------------------------------
+# Credential summary (no secret leakage)
+
+def test_credential_summary_no_secret_leak(
+    tmp_hermes_home: Path, monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """credential_summary must not leak raw token/secret material."""
+    monkeypatch.setattr(photon_auth, "_persist_runtime_env", lambda *a, **k: None)
     photon_auth.store_photon_token("token-aaaaaaaaaaaaaaaa")
-    photon_auth.store_project_credentials("proj-uuid", "secret-bbbbbbbbbbb")
+    photon_auth.store_project_credentials(
+        spectrum_project_id="sp-uuid",
+        project_secret="secret-bbbbbbbbbbb",
+        dashboard_project_id="dash-uuid",
+    )
     summary = photon_auth.credential_summary()
     blob = "\n".join(summary.values())
     assert "token-aaaa" not in blob
     assert "secret-bbbb" not in blob
     assert summary["device_token"].startswith("✓")
     assert summary["project_key"].startswith("✓")
-    assert summary["project_id"] == "proj-uuid"
+    assert summary["spectrum_project_id"] == "sp-uuid"
+    assert summary["dashboard_project_id"] == "dash-uuid"
 
 
-def test_print_credential_summary_emits_only_display_strings(
-    tmp_hermes_home: Path,
+# ---------------------------------------------------------------------------
+# Device-token candidate extraction + dashboard validation.
+
+def test_device_response_candidates_covers_known_shapes() -> None:
+    candidates = photon_auth._device_response_token_candidates(
+        {
+            "access_token": "tok-snake",
+            "accessToken": "tok-camel",
+            "data": {"access_token": "tok-data"},
+        },
+        headers={"set-auth-token": "Bearer tok-header"},
+    )
+    by_source = {c.source: c.token for c in candidates}
+    assert by_source["access_token"] == "tok-snake"
+    assert by_source["accessToken"] == "tok-camel"
+    assert by_source["data.access_token"] == "tok-data"
+    # "Bearer " prefix is stripped from the header value.
+    assert by_source["set-auth-token"] == "tok-header"
+
+
+def test_device_response_candidates_dedupes() -> None:
+    candidates = photon_auth._device_response_token_candidates(
+        {"access_token": "same", "accessToken": "same"},
+    )
+    assert [c.token for c in candidates] == ["same"]
+
+
+def test_validate_photon_token_rejects_unrecognized_session(
+    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """The emit callback must never receive raw credential bytes."""
-    photon_auth.store_photon_token("token-aaaaaaaaaaaaaaaa")
-    photon_auth.store_project_credentials("proj-uuid", "secret-bbbbbbbbbbb")
-    lines: list = []
-    photon_auth.print_credential_summary(lines.append)
-    blob = "\n".join(lines)
-    assert "token-aaaa" not in blob
-    assert "secret-bbbb" not in blob
-    assert "✓ stored" in blob   # device token line
-    assert "proj-uuid" in blob   # project id is intentionally surfaced
-    # Header is always emitted
-    assert any("Photon iMessage status" in line for line in lines)
+    def fake_get(url: str, *, headers: Dict[str, str], timeout: float) -> _FakeResponse:
+        if url.endswith("/api/auth/get-session"):
+            return _FakeResponse(json_body={})  # no "user" key
+        return _FakeResponse(json_body=[])
+
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+    with pytest.raises(photon_auth.PhotonDashboardAuthError):
+        photon_auth.validate_photon_token("some-token")
+
+
+def test_validate_photon_token_rejects_project_api_denial(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def fake_get(url: str, *, headers: Dict[str, str], timeout: float) -> _FakeResponse:
+        if url.endswith("/api/auth/get-session"):
+            return _FakeResponse(json_body={"user": {"id": "u1"}})
+        return _FakeResponse(status=403)  # project API rejects
+
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+    with pytest.raises(photon_auth.PhotonDashboardAuthError):
+        photon_auth.validate_photon_token("some-token")
+
+
+def test_login_device_flow_validates_before_persisting(
+    tmp_hermes_home: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
+        if url.endswith("/api/auth/device/code"):
+            return _FakeResponse(json_body={
+                "device_code": "dev", "user_code": "AAAA",
+                "verification_uri": "https://app.photon.codes/device",
+                "verification_uri_complete": None,
+                "expires_in": 600, "interval": 0,
+            })
+        # device/token approval
+        return _FakeResponse(json_body={"access_token": "good-token"})
+
+    def fake_get(url: str, *, headers: Dict[str, str], timeout: float) -> _FakeResponse:
+        if url.endswith("/api/auth/get-session"):
+            return _FakeResponse(json_body={"user": {"id": "u1"}})
+        return _FakeResponse(json_body=[])  # projects OK
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+
+    token = photon_auth.login_device_flow(open_browser=False)
+    assert token == "good-token"
+    assert photon_auth.load_photon_token() == "good-token"
+
+
+def test_login_device_flow_raises_when_token_invalid(
+    tmp_hermes_home: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
+        if url.endswith("/api/auth/device/code"):
+            return _FakeResponse(json_body={
+                "device_code": "dev", "user_code": "AAAA",
+                "verification_uri": "https://app.photon.codes/device",
+                "verification_uri_complete": None,
+                "expires_in": 600, "interval": 0,
+            })
+        return _FakeResponse(json_body={"access_token": "bad-token"})
+
+    def fake_get(url: str, *, headers: Dict[str, str], timeout: float) -> _FakeResponse:
+        return _FakeResponse(status=401)  # session lookup rejects
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    monkeypatch.setattr(photon_auth.httpx, "get", fake_get)
+
+    with pytest.raises(photon_auth.PhotonDashboardAuthError):
+        photon_auth.login_device_flow(open_browser=False)
+    # A token that failed validation must never be persisted.
+    assert photon_auth.load_photon_token() is None

tests/plugins/platforms/photon/test_inbound.py

@@ -1,12 +1,15 @@
 """Inbound dispatch + dedup tests for PhotonAdapter.
 
-These tests bypass the aiohttp server — they call ``_dispatch_inbound``
-and ``_is_duplicate`` directly. That keeps them fast and means we can
-exercise the message-shape parsing logic without binding ports.
+These bypass the loopback HTTP stream — they call ``_dispatch_inbound`` /
+``_on_inbound_line`` / ``_is_duplicate`` directly, exercising the
+sidecar-event parsing without spawning the Node sidecar or binding ports.
 """
 from __future__ import annotations
 
-from typing import List
+import base64
+import json
+from pathlib import Path
+from typing import Any, Dict, List
 
 import pytest
 
@@ -16,38 +19,39 @@ from plugins.platforms.photon.adapter import PhotonAdapter
 
 
 def _make_adapter(monkeypatch: pytest.MonkeyPatch) -> PhotonAdapter:
-    # Avoid touching real auth.json / env.
     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)
 
 
-@pytest.mark.asyncio
-async def test_dispatch_text_dm(monkeypatch: pytest.MonkeyPatch) -> None:
-    adapter = _make_adapter(monkeypatch)
+def _capture(adapter: PhotonAdapter, monkeypatch: pytest.MonkeyPatch) -> List[MessageEvent]:
     captured: List[MessageEvent] = []
 
     async def fake_handle(event: MessageEvent) -> None:
         captured.append(event)
 
     monkeypatch.setattr(adapter, "handle_message", fake_handle)
+    return captured
 
-    payload = {
-        "event": "messages",
-        "space": {"id": "any;-;+15551234567", "platform": "iMessage"},
-        "message": {
-            "id": "spc-msg-abc",
-            "platform": "iMessage",
-            "direction": "inbound",
-            "timestamp": "2026-05-14T19:06:32.000Z",
-            "sender": {"id": "+15551234567", "platform": "iMessage"},
-            "space": {"id": "any;-;+15551234567", "platform": "iMessage"},
-            "content": {"type": "text", "text": "hello world"},
-        },
+
+def _dm_event(text: str, msg_id: str = "spc-msg-abc") -> Dict[str, Any]:
+    return {
+        "messageId": msg_id,
+        "platform": "iMessage",
+        "space": {"id": "+15551234567", "type": "dm", "phone": "+15551234567"},
+        "sender": {"id": "+15551234567"},
+        "content": {"type": "text", "text": text},
+        "timestamp": "2026-05-14T19:06:32.000Z",
     }
-    await adapter._dispatch_inbound(payload)
+
+
+@pytest.mark.asyncio
+async def test_dispatch_text_dm(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch)
+    captured = _capture(adapter, monkeypatch)
+
+    await adapter._dispatch_inbound(_dm_event("hello world"))
 
     assert len(captured) == 1
     event = captured[0]
@@ -57,70 +61,157 @@ async def test_dispatch_text_dm(monkeypatch: pytest.MonkeyPatch) -> None:
     src = event.source
     assert src is not None
     assert src.platform == Platform("photon")
-    assert src.chat_id == "any;-;+15551234567"
+    assert src.chat_id == "+15551234567"
     assert src.chat_type == "dm"
     assert src.user_id == "+15551234567"
 
 
 @pytest.mark.asyncio
-async def test_dispatch_group_id_detected(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_dispatch_group_type(monkeypatch: pytest.MonkeyPatch) -> None:
     adapter = _make_adapter(monkeypatch)
-    captured: List[MessageEvent] = []
+    captured = _capture(adapter, monkeypatch)
 
-    async def fake_handle(event: MessageEvent) -> None:
-        captured.append(event)
-
-    monkeypatch.setattr(adapter, "handle_message", fake_handle)
-
-    payload = {
-        "event": "messages",
-        "space": {"id": "any;+;group-guid-xyz", "platform": "iMessage"},
-        "message": {
-            "id": "spc-msg-grp",
-            "timestamp": "2026-05-14T19:06:32.000Z",
-            "sender": {"id": "+15551234567"},
-            "space": {"id": "any;+;group-guid-xyz"},
-            "content": {"type": "text", "text": "hi group"},
-        },
+    event = {
+        "messageId": "spc-msg-grp",
+        "space": {"id": "group-guid-xyz", "type": "group", "phone": None},
+        "sender": {"id": "+15551234567"},
+        "content": {"type": "text", "text": "hi group"},
+        "timestamp": "2026-05-14T19:06:32.000Z",
     }
-    await adapter._dispatch_inbound(payload)
+    await adapter._dispatch_inbound(event)
     assert captured[0].source.chat_type == "group"
 
 
+# A real 1x1 transparent PNG (passes base.py's _looks_like_image magic check).
+_PNG_1X1_B64 = (
+    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYPhf"
+    "DwAChwGA60e6kgAAAABJRU5ErkJggg=="
+)
+
+
+def _attachment_event(
+    content: Dict[str, Any], msg_id: str = "spc-msg-att"
+) -> Dict[str, Any]:
+    return {
+        "messageId": msg_id,
+        "space": {"id": "+15551234567", "type": "dm", "phone": "+15551234567"},
+        "sender": {"id": "+15551234567"},
+        "content": {"type": "attachment", **content},
+        "timestamp": "2026-05-14T19:06:32.000Z",
+    }
+
+
 @pytest.mark.asyncio
-async def test_dispatch_attachment_surfaces_marker(
+async def test_dispatch_attachment_without_bytes_surfaces_marker(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """No inline ``data`` (over cap / failed sidecar read) -> text marker, no media."""
+    adapter = _make_adapter(monkeypatch)
+    captured = _capture(adapter, monkeypatch)
+
+    event = _attachment_event(
+        {"name": "IMG_4127.HEIC", "mimeType": "image/heic", "size": 12345}
+    )
+    await adapter._dispatch_inbound(event)
+    assert len(captured) == 1
+    ev = captured[0]
+    assert "Photon attachment received" in ev.text
+    assert "IMG_4127.HEIC" in ev.text
+    assert ev.message_type == MessageType.PHOTO
+    assert ev.media_urls == []
+    assert ev.media_types == []
+
+
+@pytest.mark.asyncio
+async def test_dispatch_attachment_downloads_image(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Inline base64 image bytes are decoded, cached, and exposed as media."""
+    adapter = _make_adapter(monkeypatch)
+    captured = _capture(adapter, monkeypatch)
+
+    raw = base64.b64decode(_PNG_1X1_B64)
+    event = _attachment_event(
+        {
+            "name": "photo.png",
+            "mimeType": "image/png",
+            "size": len(raw),
+            "data": _PNG_1X1_B64,
+            "encoding": "base64",
+        }
+    )
+    await adapter._dispatch_inbound(event)
+
+    assert len(captured) == 1
+    ev = captured[0]
+    assert ev.message_type == MessageType.PHOTO
+    assert ev.media_types == ["image/png"]
+    assert len(ev.media_urls) == 1
+    cached = Path(ev.media_urls[0])
+    try:
+        assert cached.is_file()
+        assert cached.read_bytes() == raw
+        assert ev.text == "(attachment)"
+    finally:
+        cached.unlink(missing_ok=True)
+
+
+@pytest.mark.asyncio
+async def test_dispatch_attachment_downloads_document(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Non-image attachments route through the document cache as DOCUMENT."""
+    adapter = _make_adapter(monkeypatch)
+    captured = _capture(adapter, monkeypatch)
+
+    raw = b"%PDF-1.4 hermes test document"
+    event = _attachment_event(
+        {
+            "name": "report.pdf",
+            "mimeType": "application/pdf",
+            "size": len(raw),
+            "data": base64.b64encode(raw).decode("ascii"),
+            "encoding": "base64",
+        }
+    )
+    await adapter._dispatch_inbound(event)
+
+    assert len(captured) == 1
+    ev = captured[0]
+    assert ev.message_type == MessageType.DOCUMENT
+    assert ev.media_types == ["application/pdf"]
+    assert len(ev.media_urls) == 1
+    cached = Path(ev.media_urls[0])
+    try:
+        assert cached.is_file()
+        assert cached.read_bytes() == raw
+        assert ev.text == "(attachment)"
+    finally:
+        cached.unlink(missing_ok=True)
+
+
+@pytest.mark.asyncio
+async def test_on_inbound_line_dispatches_and_dedups(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     adapter = _make_adapter(monkeypatch)
-    captured: List[MessageEvent] = []
+    captured = _capture(adapter, monkeypatch)
 
-    async def fake_handle(event: MessageEvent) -> None:
-        captured.append(event)
+    line = json.dumps(_dm_event("ping", msg_id="dup-1"))
+    await adapter._on_inbound_line(line)
+    await adapter._on_inbound_line(line)  # same messageId -> deduped
 
-    monkeypatch.setattr(adapter, "handle_message", fake_handle)
-
-    payload = {
-        "event": "messages",
-        "message": {
-            "id": "spc-msg-att",
-            "timestamp": "2026-05-14T19:06:32.000Z",
-            "sender": {"id": "+15551234567"},
-            "space": {"id": "any;-;+15551234567"},
-            "content": {
-                "type": "attachment",
-                "name": "IMG_4127.HEIC",
-                "mimeType": "image/heic",
-                "size": 12345,
-            },
-        },
-    }
-    await adapter._dispatch_inbound(payload)
     assert len(captured) == 1
-    event = captured[0]
-    # Attachment carries metadata marker; mime → MessageType.PHOTO.
-    assert "Photon attachment received" in event.text
-    assert "IMG_4127.HEIC" in event.text
-    assert event.message_type == MessageType.PHOTO
+    assert captured[0].text == "ping"
+
+
+@pytest.mark.asyncio
+async def test_on_inbound_line_ignores_bad_json(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch)
+    captured = _capture(adapter, monkeypatch)
+
+    await adapter._on_inbound_line("{not json")
+    assert captured == []
 
 
 def test_is_duplicate_window(monkeypatch: pytest.MonkeyPatch) -> None:

tests/plugins/platforms/photon/test_mention_gating.py

@@ -22,7 +22,6 @@ from plugins.platforms.photon.adapter import PhotonAdapter
 def _make_adapter(monkeypatch: pytest.MonkeyPatch, extra: dict | None = None) -> PhotonAdapter:
     monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id")
     monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret")
-    monkeypatch.delenv("PHOTON_WEBHOOK_SECRET", raising=False)
     monkeypatch.delenv("PHOTON_REQUIRE_MENTION", raising=False)
     monkeypatch.delenv("PHOTON_MENTION_PATTERNS", raising=False)
     cfg = PlatformConfig(enabled=True, token="", extra=extra or {})
@@ -31,27 +30,21 @@ def _make_adapter(monkeypatch: pytest.MonkeyPatch, extra: dict | None = None) ->
 
 def _group_payload(text: str) -> dict:
     return {
-        "event": "messages",
-        "message": {
-            "id": f"grp-{abs(hash(text))}",
-            "timestamp": "2026-05-14T19:06:32.000Z",
-            "sender": {"id": "+15551234567"},
-            "space": {"id": "any;+;group-guid-xyz"},
-            "content": {"type": "text", "text": text},
-        },
+        "messageId": f"grp-{abs(hash(text))}",
+        "space": {"id": "group-guid-xyz", "type": "group", "phone": None},
+        "sender": {"id": "+15551234567"},
+        "content": {"type": "text", "text": text},
+        "timestamp": "2026-05-14T19:06:32.000Z",
     }
 
 
 def _dm_payload(text: str) -> dict:
     return {
-        "event": "messages",
-        "message": {
-            "id": f"dm-{abs(hash(text))}",
-            "timestamp": "2026-05-14T19:06:32.000Z",
-            "sender": {"id": "+15551234567"},
-            "space": {"id": "any;-;+15551234567"},
-            "content": {"type": "text", "text": text},
-        },
+        "messageId": f"dm-{abs(hash(text))}",
+        "space": {"id": "+15551234567", "type": "dm", "phone": "+15551234567"},
+        "sender": {"id": "+15551234567"},
+        "content": {"type": "text", "text": text},
+        "timestamp": "2026-05-14T19:06:32.000Z",
     }
 
 
@@ -126,7 +119,6 @@ def test_custom_mention_patterns_from_config(monkeypatch: pytest.MonkeyPatch) ->
 def test_mention_patterns_env_comma_separated(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id")
     monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret")
-    monkeypatch.delenv("PHOTON_WEBHOOK_SECRET", raising=False)
     monkeypatch.setenv("PHOTON_REQUIRE_MENTION", "true")
     monkeypatch.setenv("PHOTON_MENTION_PATTERNS", r"bot\b, assistant\b")
     cfg = PlatformConfig(enabled=True, token="", extra={})

tests/plugins/platforms/photon/test_setup_access.py

@@ -0,0 +1,69 @@
+"""Tests for `hermes photon setup`'s access auto-configuration.
+
+`_autoconfigure_access` allowlists the operator and points the cron home
+channel at their DM, writing to the per-test ~/.hermes/.env (the hermetic
+HERMES_HOME fixture isolates this). It must fill only unset keys so a re-run
+never clobbers a hand-tuned allowlist.
+"""
+from __future__ import annotations
+
+import pytest
+
+from hermes_cli.config import get_env_value, save_env_value
+from plugins.platforms.photon.adapter import _env_enablement
+from plugins.platforms.photon import cli
+
+
+def test_autoconfigure_access_fills_unset(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PHOTON_ALLOWED_USERS", raising=False)
+    monkeypatch.delenv("PHOTON_HOME_CHANNEL", raising=False)
+
+    cli._autoconfigure_access("+15551234567")
+
+    assert get_env_value("PHOTON_ALLOWED_USERS") == "+15551234567"
+    assert get_env_value("PHOTON_HOME_CHANNEL") == "+15551234567"
+
+
+def test_autoconfigure_access_preserves_existing_allowlist(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("PHOTON_ALLOWED_USERS", raising=False)
+    monkeypatch.delenv("PHOTON_HOME_CHANNEL", raising=False)
+    # A hand-tuned allowlist already in place must survive a setup re-run.
+    save_env_value("PHOTON_ALLOWED_USERS", "+19998887777,+15551112222")
+
+    cli._autoconfigure_access("+15551234567")
+
+    assert get_env_value("PHOTON_ALLOWED_USERS") == "+19998887777,+15551112222"
+    # The still-unset home channel is filled.
+    assert get_env_value("PHOTON_HOME_CHANNEL") == "+15551234567"
+
+
+def test_env_enablement_seeds_home_channel(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("PHOTON_PROJECT_ID", "project_123")
+    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "secret_123")
+    monkeypatch.setenv("PHOTON_HOME_CHANNEL", "+15551234567")
+    monkeypatch.setenv("PHOTON_HOME_CHANNEL_NAME", "Primary DM")
+
+    seed = _env_enablement()
+
+    assert seed is not None
+    assert seed["home_channel"] == {
+        "chat_id": "+15551234567",
+        "name": "Primary DM",
+    }
+
+
+def test_env_enablement_home_channel_defaults_name(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("PHOTON_PROJECT_ID", "project_123")
+    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "secret_123")
+    monkeypatch.setenv("PHOTON_HOME_CHANNEL", "+15551234567")
+    monkeypatch.delenv("PHOTON_HOME_CHANNEL_NAME", raising=False)
+
+    seed = _env_enablement()
+
+    assert seed is not None
+    assert seed["home_channel"] == {
+        "chat_id": "+15551234567",
+        "name": "Home",
+    }

tests/plugins/platforms/photon/test_signature.py

@@ -1,95 +0,0 @@
-"""Signature verification tests for the Photon webhook receiver."""
-from __future__ import annotations
-
-import hashlib
-import hmac
-import time
-
-import pytest
-
-from plugins.platforms.photon.adapter import verify_signature
-
-
-def _sign(secret: str, body: bytes, ts: int) -> str:
-    return "v0=" + hmac.new(
-        secret.encode(), f"v0:{ts}:".encode() + body, hashlib.sha256,
-    ).hexdigest()
-
-
-def test_accepts_valid_signature() -> None:
-    secret = "topsecret-32chars-or-whatever"
-    body = b'{"event":"messages"}'
-    ts = int(time.time())
-    sig = _sign(secret, body, ts)
-    assert verify_signature(
-        body=body, timestamp_header=str(ts), signature_header=sig,
-        signing_secret=secret,
-    )
-
-
-def test_rejects_tampered_body() -> None:
-    secret = "s"
-    body = b'{"event":"messages"}'
-    ts = int(time.time())
-    sig = _sign(secret, body, ts)
-    assert not verify_signature(
-        body=body + b" tamper", timestamp_header=str(ts),
-        signature_header=sig, signing_secret=secret,
-    )
-
-
-def test_rejects_wrong_secret() -> None:
-    body = b"x"
-    ts = int(time.time())
-    sig = _sign("right", body, ts)
-    assert not verify_signature(
-        body=body, timestamp_header=str(ts), signature_header=sig,
-        signing_secret="wrong",
-    )
-
-
-def test_rejects_drifted_timestamp() -> None:
-    secret = "s"
-    body = b"x"
-    ts = int(time.time()) - 3600  # 1h old; drift window is 5 min
-    sig = _sign(secret, body, ts)
-    assert not verify_signature(
-        body=body, timestamp_header=str(ts), signature_header=sig,
-        signing_secret=secret,
-    )
-
-
-def test_rejects_missing_v0_prefix() -> None:
-    secret = "s"
-    body = b"x"
-    ts = int(time.time())
-    raw_hex = hmac.new(
-        secret.encode(), f"v0:{ts}:".encode() + body, hashlib.sha256,
-    ).hexdigest()
-    # Strip the "v0=" prefix — verify_signature must reject.
-    assert not verify_signature(
-        body=body, timestamp_header=str(ts), signature_header=raw_hex,
-        signing_secret=secret,
-    )
-
-
-def test_rejects_empty_inputs() -> None:
-    assert not verify_signature(
-        body=b"x", timestamp_header="", signature_header="v0=abc",
-        signing_secret="s",
-    )
-    assert not verify_signature(
-        body=b"x", timestamp_header="123", signature_header="",
-        signing_secret="s",
-    )
-    assert not verify_signature(
-        body=b"x", timestamp_header="123", signature_header="v0=abc",
-        signing_secret="",
-    )
-
-
-def test_rejects_non_integer_timestamp() -> None:
-    assert not verify_signature(
-        body=b"x", timestamp_header="not-an-int",
-        signature_header="v0=abc", signing_secret="s",
-    )

tests/tools/test_send_message_target_parse.py

@@ -0,0 +1,20 @@
+"""Parser-only tests for send_message targets.
+
+These stay separate from ``test_send_message_tool.py`` because that module
+skips wholesale when optional Telegram dependencies are not installed.
+"""
+
+from tools.send_message_tool import _parse_target_ref
+
+
+def test_photon_e164_target_is_explicit() -> None:
+    chat_id, thread_id, is_explicit = _parse_target_ref("photon", "+15551234567")
+
+    assert chat_id == "+15551234567"
+    assert thread_id is None
+    assert is_explicit is True
+
+
+def test_e164_target_still_requires_phone_platform() -> None:
+    assert _parse_target_ref("matrix", "+15551234567")[2] is False
+

tests/tools/test_send_message_tool.py

@@ -1199,6 +1199,11 @@ class TestParseTargetRefE164:
         assert chat_id == "+15551234567"
         assert is_explicit is True
 
+    def test_photon_e164_is_explicit(self):
+        chat_id, _, is_explicit = _parse_target_ref("photon", "+15551234567")
+        assert chat_id == "+15551234567"
+        assert is_explicit is True
+
     def test_signal_bare_digits_still_work(self):
         """Bare digit strings continue to match the generic numeric branch."""
         chat_id, _, is_explicit = _parse_target_ref("signal", "15551234567")

tools/send_message_tool.py

@@ -38,7 +38,7 @@ _NUMERIC_TOPIC_RE = _TELEGRAM_TOPIC_TARGET_RE
 # below and falls through to channel-name resolution, which has no way to
 # resolve a raw phone number. Keeping the '+' preserves the E.164 form that
 # downstream adapters (signal, etc.) expect.
-_PHONE_PLATFORMS = frozenset({"signal", "sms", "whatsapp"})
+_PHONE_PLATFORMS = frozenset({"photon", "signal", "sms", "whatsapp"})
 _E164_TARGET_RE = re.compile(r"^\s*\+(\d{7,15})\s*$")
 # Email addresses — a valid email like "user@domain.com" should be treated as
 # an explicit target for the email platform, not fall through to channel-name

tui_gateway/server.py

@@ -175,6 +175,7 @@ _LONG_HANDLERS = frozenset(
     {
         "browser.manage",
         "cli.exec",
+        "plugins.manage",
         "session.branch",
         "session.compress",
         "session.resume",
@@ -9020,7 +9021,83 @@ def _(rid, params: dict) -> dict:
         return _err(rid, 5025, str(e))
 
 
-# ── Methods: shell ───────────────────────────────────────────────────
+@method("plugins.manage")
+def _(rid, params: dict) -> dict:
+    """List installed plugins with activation state, or toggle one on/off.
+
+    Backs the TUI Plugins Hub. Uses the same disk-discovery + enable/disable
+    primitives as ``hermes plugins`` / the dashboard, so the three surfaces
+    agree on what's installed and what's enabled.
+
+    Actions:
+      - ``list``   → {"plugins": [{name, version, description, source,
+                       status}], "user_count": N, "bundled_count": M}
+      - ``toggle`` → flip ``name`` based on ``enable`` (bool). Returns the
+                       refreshed row plus {"ok", "unchanged"}.
+    """
+    action = params.get("action", "list")
+    try:
+        from hermes_cli.plugins_cmd import (
+            _discover_all_plugins,
+            _get_disabled_set,
+            _get_enabled_set,
+            _plugin_status,
+        )
+
+        def _rows():
+            enabled = _get_enabled_set()
+            disabled = _get_disabled_set()
+            out = []
+            for name, version, desc, source, _dir, key in sorted(
+                _discover_all_plugins()
+            ):
+                out.append(
+                    {
+                        "name": name,
+                        "version": str(version or ""),
+                        "description": desc or "",
+                        "source": source,
+                        "status": _plugin_status(name, enabled, disabled, key=key),
+                    }
+                )
+            return out
+
+        if action == "list":
+            rows = _rows()
+            user_count = sum(1 for r in rows if r["source"] != "bundled")
+            return _ok(
+                rid,
+                {
+                    "plugins": rows,
+                    "user_count": user_count,
+                    "bundled_count": len(rows) - user_count,
+                },
+            )
+
+        if action == "toggle":
+            from hermes_cli.plugins_cmd import dashboard_set_agent_plugin_enabled
+
+            name = (params.get("name") or "").strip()
+            if not name:
+                return _err(rid, 4019, "plugins.toggle requires a 'name'")
+            enable = bool(params.get("enable"))
+            result = dashboard_set_agent_plugin_enabled(name, enabled=enable)
+            if not result.get("ok"):
+                return _err(rid, 5026, result.get("error") or "toggle failed")
+            row = next((r for r in _rows() if r["name"] == name), None)
+            return _ok(
+                rid,
+                {
+                    "ok": True,
+                    "unchanged": bool(result.get("unchanged")),
+                    "name": name,
+                    "plugin": row,
+                },
+            )
+
+        return _err(rid, 4017, f"unknown plugins action: {action}")
+    except Exception as e:
+        return _err(rid, 5026, str(e))
 
 
 @method("shell.exec")

ui-tui/packages/hermes-ink/src/ink/app-rawmode-mouse.test.ts

@@ -0,0 +1,91 @@
+import { EventEmitter } from 'events'
+import React, { useContext, useEffect } from 'react'
+import { describe, expect, it } from 'vitest'
+
+import StdinContext from './components/StdinContext.js'
+import Text from './components/Text.js'
+import Ink from './ink.js'
+import { DISABLE_MOUSE_TRACKING } from './termio/dec.js'
+
+class FakeTty extends EventEmitter {
+  chunks: string[] = []
+  columns = 80
+  rows = 24
+  isTTY = true
+  isRaw = false
+
+  ref(): void {}
+  unref(): void {}
+  read(): null {
+    return null
+  }
+  setEncoding(): this {
+    return this
+  }
+  setRawMode(mode: boolean): this {
+    this.isRaw = mode
+    return this
+  }
+  write(chunk: string | Uint8Array, cb?: (err?: Error | null) => void): boolean {
+    this.chunks.push(typeof chunk === 'string' ? chunk : Buffer.from(chunk).toString('utf8'))
+    cb?.()
+    return true
+  }
+}
+
+const tick = () => new Promise<void>(resolve => setImmediate(resolve))
+
+// A child that grabs the last useInput consumer's raw-mode toggle. Mounting
+// enables raw mode (count 0→1); unmounting disables it (count 1→0), which is
+// the teardown path that must DISABLE_MOUSE_TRACKING so DEC 1003 hover can't
+// leak as cooked-mode `35;col;row M` text over the prompt.
+function RawModeConsumer({ active }: { active: boolean }) {
+  const { setRawMode, isRawModeSupported } = useContext(StdinContext)
+
+  useEffect(() => {
+    if (!active || !isRawModeSupported) {
+      return
+    }
+
+    setRawMode(true)
+
+    return () => setRawMode(false)
+  }, [active, isRawModeSupported, setRawMode])
+
+  return React.createElement(Text, null, 'x')
+}
+
+describe('App raw-mode teardown', () => {
+  it('disables mouse tracking when the last raw-mode consumer detaches', async () => {
+    const stdout = new FakeTty()
+    const stdin = new FakeTty()
+    const stderr = new FakeTty()
+    const ink = new Ink({
+      exitOnCtrlC: false,
+      patchConsole: false,
+      stderr: stderr as unknown as NodeJS.WriteStream,
+      stdin: stdin as unknown as NodeJS.ReadStream,
+      stdout: stdout as unknown as NodeJS.WriteStream
+    })
+
+    // Mouse tracking is asserted on the alt screen; the teardown path lives in
+    // App, independent of who enabled tracking.
+    ink.setAltScreenActive(true, 'all')
+    ink.render(React.createElement(RawModeConsumer, { active: true }))
+    ink.onRender()
+    await tick()
+    expect(stdin.isRaw).toBe(true)
+
+    stdout.chunks = []
+
+    // Drop the consumer → raw-mode count hits 0 → teardown runs.
+    ink.render(React.createElement(RawModeConsumer, { active: false }))
+    ink.onRender()
+    await tick()
+
+    expect(stdin.isRaw).toBe(false)
+    expect(stdout.chunks.join('')).toContain(DISABLE_MOUSE_TRACKING)
+
+    ink.unmount()
+  })
+})

ui-tui/packages/hermes-ink/src/ink/components/App.tsx

@@ -9,6 +9,7 @@ import type { DOMElement } from '../dom.js'
 import { EventEmitter } from '../events/emitter.js'
 import { InputEvent } from '../events/input-event.js'
 import { TerminalFocusEvent } from '../events/terminal-focus-event.js'
+import instances from '../instances.js'
 import {
   INITIAL_STATE,
   type ParsedInput,
@@ -309,6 +310,21 @@ export default class App extends PureComponent<Props, State> {
             }
           })
         })
+
+        // Re-assert mouse tracking on raw-mode re-entry. <AlternateScreen>
+        // owns the initial enable, but its effect only re-runs on a
+        // mode/writeRaw change — NOT on a raw-mode bounce (count 1→0→1, e.g.
+        // an overlay that briefly drops the last useInput consumer). The
+        // teardown above now DISABLE_MOUSE_TRACKING's to stop the cooked-echo
+        // leak, so without this the terminal would be left with tracking off
+        // and the mouse silently dead until the next stdin-gap/resize
+        // re-assert. reassertTerminalModes() is gated on altScreenActive and
+        // idempotent, so it's a no-op when there's nothing to restore.
+        // Deferred (same setImmediate discipline as the XTVERSION probe) so it
+        // lands after any alt-screen enable writes in this render cycle.
+        setImmediate(() => {
+          instances.get(this.props.stdout)?.reassertTerminalModes()
+        })
       }
 
       this.rawModeEnabledCount++
@@ -324,6 +340,15 @@ export default class App extends PureComponent<Props, State> {
       this.props.stdout.write(DFE)
       // Disable bracketed paste mode
       this.props.stdout.write(DBP)
+      // Disable mouse tracking. Tracking is asserted by <AlternateScreen> /
+      // the Ink instance, NOT here — but dropping raw mode + detaching the
+      // readable listener while DEC 1003 hover stays on means the terminal
+      // falls back to cooked-mode echo and every mouse move leaks as text
+      // (`35;col;row M` shards over the prompt). Same hazard handleSuspend()
+      // already guards against; this teardown path missed it. Idempotent
+      // (no-op if tracking was never on), and re-enabling raw mode below
+      // re-asserts tracking so a transient drop→re-add round-trips cleanly.
+      this.props.stdout.write(DISABLE_MOUSE_TRACKING)
       stdin.setRawMode(false)
       stdin.removeListener('readable', this.handleReadable)
       stdin.unref()

ui-tui/src/app/interfaces.ts

@@ -93,6 +93,7 @@ export interface OverlayState {
   confirm: ConfirmReq | null
   modelPicker: boolean
   pager: null | PagerState
+  pluginsHub: boolean
   secret: null | SecretReq
   sessions: boolean
   skillsHub: boolean

ui-tui/src/app/overlayStore.ts

@@ -10,6 +10,7 @@ const buildOverlayState = (): OverlayState => ({
   confirm: null,
   modelPicker: false,
   pager: null,
+  pluginsHub: false,
   secret: null,
   sessions: false,
   skillsHub: false,
@@ -20,8 +21,10 @@ export const $overlayState = atom<OverlayState>(buildOverlayState())
 
 export const $isBlocked = computed(
   $overlayState,
-  ({ agents, approval, clarify, confirm, modelPicker, pager, secret, sessions, skillsHub, sudo }) =>
-    Boolean(agents || approval || clarify || confirm || modelPicker || pager || secret || sessions || skillsHub || sudo)
+  ({ agents, approval, clarify, confirm, modelPicker, pager, pluginsHub, secret, sessions, skillsHub, sudo }) =>
+    Boolean(
+      agents || approval || clarify || confirm || modelPicker || pager || pluginsHub || secret || sessions || skillsHub || sudo
+    )
 )
 
 export const getOverlayState = () => $overlayState.get()
@@ -46,6 +49,7 @@ export const resetFlowOverlays = () =>
     agents: $overlayState.get().agents,
     agentsInitialHistoryIndex: $overlayState.get().agentsInitialHistoryIndex,
     modelPicker: $overlayState.get().modelPicker,
+    pluginsHub: $overlayState.get().pluginsHub,
     sessions: $overlayState.get().sessions,
     skillsHub: $overlayState.get().skillsHub
   })

ui-tui/src/app/slash/commands/ops.ts

@@ -652,6 +652,34 @@ export const opsCommands: SlashCommand[] = [
     }
   },
 
+  {
+    help: 'view & toggle plugins (no arg opens the hub; enable/disable <name> for direct toggle)',
+    name: 'plugins',
+    run: (arg, ctx, cmd) => {
+      // No argument → open the interactive Plugins Hub overlay. Any
+      // subcommand (enable/disable/list/install/…) falls through to the
+      // text slash worker so it stays at parity with `hermes plugins`.
+      if (!arg.trim()) {
+        return patchOverlayState({ pluginsHub: true })
+      }
+
+      ctx.gateway.gw
+        .request<SlashExecResponse>('slash.exec', { command: cmd.slice(1), session_id: ctx.sid })
+        .then(r => {
+          if (ctx.stale()) {
+            return
+          }
+
+          const body = r?.output || '/plugins: no output'
+          const text = r?.warning ? `warning: ${r.warning}\n${body}` : body
+          const long = text.length > 180 || text.split('\n').filter(Boolean).length > 2
+
+          long ? ctx.transcript.page(text, 'Plugins') : ctx.transcript.sys(text)
+        })
+        .catch(ctx.guardedErr)
+    }
+  },
+
   {
     help: 'enable or disable tools (client-side history reset on change)',
     name: 'tools',

ui-tui/src/app/useInputHandlers.ts

@@ -151,6 +151,10 @@ export function useInputHandlers(ctx: InputHandlerContext): InputHandlerResult {
       return patchOverlayState({ skillsHub: false })
     }
 
+    if (overlay.pluginsHub) {
+      return patchOverlayState({ pluginsHub: false })
+    }
+
     if (overlay.sessions) {
       return patchOverlayState({ sessions: false })
     }

ui-tui/src/components/appOverlays.tsx

@@ -11,6 +11,7 @@ import { FloatBox } from './appChrome.js'
 import { MaskedPrompt } from './maskedPrompt.js'
 import { ModelPicker } from './modelPicker.js'
 import { OverlayHint } from './overlayControls.js'
+import { PluginsHub } from './pluginsHub.js'
 import { ApprovalPrompt, ClarifyPrompt, ConfirmPrompt } from './prompts.js'
 import { SkillsHub } from './skillsHub.js'
 
@@ -125,6 +126,7 @@ export function FloatingOverlays({
     overlay.pager ||
     overlay.sessions ||
     overlay.skillsHub ||
+    overlay.pluginsHub ||
     completions.length
 
   if (!hasAny) {
@@ -174,6 +176,12 @@ export function FloatingOverlays({
         </FloatBox>
       )}
 
+      {overlay.pluginsHub && (
+        <FloatBox color={theme.color.border}>
+          <PluginsHub gw={gw} onClose={() => patchOverlayState({ pluginsHub: false })} t={theme} />
+        </FloatBox>
+      )}
+
       {overlay.pager && (
         <FloatBox color={theme.color.border}>
           <Box flexDirection="column" paddingX={1} paddingY={1}>

ui-tui/src/components/pluginsHub.tsx

@@ -0,0 +1,238 @@
+import { Box, Text, useInput, useStdout } from '@hermes/ink'
+import { useEffect, useState } from 'react'
+
+import type { GatewayClient } from '../gatewayClient.js'
+import { rpcErrorMessage } from '../lib/rpc.js'
+import type { Theme } from '../theme.js'
+
+import { OverlayHint, useOverlayKeys, windowItems, windowOffset } from './overlayControls.js'
+
+const VISIBLE = 12
+const MIN_WIDTH = 44
+const MAX_WIDTH = 96
+
+interface PluginRow {
+  description?: string
+  name: string
+  source?: string
+  status?: string
+  version?: string
+}
+
+interface PluginsListResponse {
+  bundled_count?: number
+  plugins?: PluginRow[]
+  user_count?: number
+}
+
+interface PluginsToggleResponse {
+  name?: string
+  ok?: boolean
+  plugin?: PluginRow
+  unchanged?: boolean
+}
+
+type Scope = 'all' | 'user'
+
+const GLYPH: Record<string, string> = {
+  disabled: '✗',
+  enabled: '✓'
+}
+
+export function PluginsHub({ gw, onClose, t }: PluginsHubProps) {
+  const [rows, setRows] = useState<PluginRow[]>([])
+  const [bundledCount, setBundledCount] = useState(0)
+  const [userCount, setUserCount] = useState(0)
+  const [idx, setIdx] = useState(0)
+  const [scope, setScope] = useState<Scope>('user')
+  const [busy, setBusy] = useState(false)
+  const [err, setErr] = useState('')
+  const [loading, setLoading] = useState(true)
+
+  const { stdout } = useStdout()
+  const width = Math.max(MIN_WIDTH, Math.min(MAX_WIDTH, (stdout?.columns ?? 80) - 6))
+
+  const load = () => {
+    gw.request<PluginsListResponse>('plugins.manage', { action: 'list' })
+      .then(r => {
+        setRows(r?.plugins ?? [])
+        setUserCount(Number(r?.user_count ?? 0))
+        setBundledCount(Number(r?.bundled_count ?? 0))
+        setErr('')
+        setLoading(false)
+      })
+      .catch((e: unknown) => {
+        setErr(rpcErrorMessage(e))
+        setLoading(false)
+      })
+  }
+
+  useEffect(load, [gw])
+
+  // Default to user plugins; fall back to all when there are none so the
+  // overlay is never empty when bundled plugins exist.
+  const visibleRows = scope === 'user' ? rows.filter(r => r.source !== 'bundled') : rows
+  const effectiveRows = scope === 'user' && !visibleRows.length && rows.length ? rows : visibleRows
+  const effectiveScope: Scope = effectiveRows === visibleRows ? scope : 'all'
+  const clampedIdx = Math.min(idx, Math.max(0, effectiveRows.length - 1))
+
+  useOverlayKeys({ disabled: busy, onClose })
+
+  const toggle = (row: PluginRow) => {
+    if (busy || !row) {
+      return
+    }
+
+    const enable = row.status !== 'enabled'
+    setBusy(true)
+    setErr('')
+
+    gw.request<PluginsToggleResponse>('plugins.manage', { action: 'toggle', enable, name: row.name })
+      .then(r => {
+        if (r?.plugin) {
+          setRows(prev => prev.map(p => (p.name === r.plugin!.name ? r.plugin! : p)))
+        } else {
+          load()
+        }
+      })
+      .catch((e: unknown) => setErr(rpcErrorMessage(e)))
+      .finally(() => setBusy(false))
+  }
+
+  useInput((ch, key) => {
+    if (busy) {
+      return
+    }
+
+    const count = effectiveRows.length
+
+    if (key.upArrow && clampedIdx > 0) {
+      setIdx(clampedIdx - 1)
+
+      return
+    }
+
+    if (key.downArrow && clampedIdx < count - 1) {
+      setIdx(clampedIdx + 1)
+
+      return
+    }
+
+    // Tab toggles user-only vs all (bundled) scope.
+    if (key.tab) {
+      setScope(s => (s === 'user' ? 'all' : 'user'))
+      setIdx(0)
+
+      return
+    }
+
+    if (key.return || ch === ' ') {
+      const row = effectiveRows[clampedIdx]
+
+      if (row) {
+        toggle(row)
+      }
+
+      return
+    }
+
+    const n = ch === '0' ? 10 : parseInt(ch, 10)
+
+    if (!Number.isNaN(n) && n >= 1 && n <= Math.min(10, count)) {
+      const next = windowOffset(count, clampedIdx, VISIBLE) + n - 1
+      const row = effectiveRows[next]
+
+      if (row) {
+        setIdx(next)
+        toggle(row)
+      }
+    }
+  })
+
+  if (loading) {
+    return <Text color={t.color.muted}>loading plugins…</Text>
+  }
+
+  if (err && !rows.length) {
+    return (
+      <Box flexDirection="column" width={width}>
+        <Text color={t.color.label}>error: {err}</Text>
+        <OverlayHint t={t}>Esc/q close</OverlayHint>
+      </Box>
+    )
+  }
+
+  if (!rows.length) {
+    return (
+      <Box flexDirection="column" width={width}>
+        <Text bold color={t.color.accent}>
+          Plugins Hub
+        </Text>
+        <Text color={t.color.muted}>no plugins installed</Text>
+        <Text color={t.color.muted}>install: hermes plugins install owner/repo</Text>
+        <OverlayHint t={t}>Esc/q close</OverlayHint>
+      </Box>
+    )
+  }
+
+  const labels = effectiveRows.map(r => {
+    const status = r.status ?? 'not enabled'
+    const glyph = GLYPH[status] ?? '○'
+    const ver = r.version ? ` v${r.version}` : ''
+    const src = effectiveScope === 'all' && r.source === 'bundled' ? ' [bundled]' : ''
+    const state = status === 'enabled' ? '' : ` (${status})`
+
+    return `${glyph} ${r.name}${ver}${src}${state}`
+  })
+
+  const { items, offset } = windowItems(labels, clampedIdx, VISIBLE)
+
+  const scopeLabel =
+    effectiveScope === 'user'
+      ? `${userCount} user plugin(s)${bundledCount ? ` · +${bundledCount} bundled (Tab)` : ''}`
+      : `all ${rows.length} plugins`
+
+  return (
+    <Box flexDirection="column" width={width}>
+      <Text bold color={t.color.accent}>
+        Plugins Hub
+      </Text>
+
+      <Text color={t.color.muted}>{scopeLabel}</Text>
+      {offset > 0 && <Text color={t.color.muted}> ↑ {offset} more</Text>}
+
+      {items.map((row, i) => {
+        const lineIdx = offset + i
+        const active = clampedIdx === lineIdx
+
+        return (
+          <Text
+            bold={active}
+            color={active ? t.color.accent : t.color.muted}
+            inverse={active}
+            key={effectiveRows[lineIdx]?.name ?? row}
+            wrap="truncate-end"
+          >
+            {active ? '▸ ' : '  '}
+            {i + 1}. {row}
+          </Text>
+        )
+      })}
+
+      {offset + VISIBLE < labels.length && (
+        <Text color={t.color.muted}> ↓ {labels.length - offset - VISIBLE} more</Text>
+      )}
+
+      {err ? <Text color={t.color.label}>error: {err}</Text> : null}
+      {busy ? <Text color={t.color.accent}>updating…</Text> : null}
+
+      <OverlayHint t={t}>↑/↓ select · Enter/Space toggle · Tab user/all · 1-9,0 quick · Esc/q close</OverlayHint>
+    </Box>
+  )
+}
+
+interface PluginsHubProps {
+  gw: GatewayClient
+  onClose: () => void
+  t: Theme
+}

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

@@ -22,26 +22,30 @@ your account.
 
 ## Architecture
 
-Inbound messages arrive as **signed webhooks**: Photon POSTs JSON with
-an `X-Spectrum-Signature` header to a URL you register, and Hermes'
-aiohttp listener verifies the HMAC-SHA256 signature before dispatching
-the event into the agent.
+Photon is a **persistent-connection** channel, like Discord or Slack —
+**no webhook, no public URL, no signing secret to manage.**
 
-Outbound replies go through a small supervised **Node sidecar** that
-runs the `spectrum-ts` SDK on loopback. Photon does not currently
-expose a public HTTP send-message endpoint — that's a roadmap item on
-their side — so until then the sidecar is the only way to call
-`Space.send(...)`. The Python plugin starts, supervises, and shuts
-down the sidecar automatically. When Photon ships an HTTP send
-endpoint we'll retire the sidecar in a follow-up release.
+The `spectrum-ts` SDK holds a long-lived **gRPC stream** to Photon for
+both directions. Because the SDK is TypeScript-only, Hermes runs it in a
+small supervised **Node sidecar** and talks to it over loopback:
+
+- **Inbound** — the sidecar consumes the SDK's `app.messages` gRPC
+  stream and forwards each message to the Python adapter over a loopback
+  `GET /inbound` (NDJSON). The adapter dedupes and dispatches it to the
+  agent, reconnecting automatically if the stream drops.
+- **Outbound** — replies are loopback POSTs to the sidecar, which calls
+  `space.send(...)` on the SDK.
+
+The Python plugin starts, supervises, and shuts down the sidecar
+automatically.
 
 ## Prerequisites
 
 - A Photon account — sign up at [app.photon.codes][app]
 - **Node.js 18.17 or newer** on PATH (`node --version`)
 - A phone number that can receive iMessage (used to bind your account)
-- A publicly reachable URL for the webhook receiver — Cloudflare
-  Tunnel, ngrok, or your own gateway hostname all work
+
+That's it — there is no public URL or tunnel to set up.
 
 ## First-time setup
 
@@ -58,17 +62,24 @@ hermes gateway setup
 hermes photon setup --phone +15551234567
 ```
 
-The setup:
+The setup, in order:
 
-1. Opens `https://app.photon.codes/` for device approval
-2. Creates a Spectrum-enabled project under your account
-3. Calls the Spectrum `create-user` endpoint with `type: shared` so
-   Photon allocates an iMessage line from the free pool
-4. Runs `npm install` inside the plugin's sidecar directory
+1. **Device login** (`client_id=photon-cli`) — opens
+   `https://app.photon.codes/` for approval and stores the bearer token.
+2. **Finds or creates** the `Hermes Agent` project on your account.
+3. **Enables Spectrum**, reads the project's Spectrum id, and rotates
+   the project secret.
+4. **Registers your phone number** as a Spectrum user — skipped if a
+   user with that number already exists, so re-running is safe.
+5. **Prints your assigned iMessage line** — the number you text to reach
+   your agent.
+6. **Runs `npm install`** inside the plugin's sidecar directory.
 
-Credentials are stored in `~/.hermes/auth.json` under
-`credential_pool.photon` (bearer token) and
-`credential_pool.photon_project` (project id + secret).
+Runtime credentials are written to `~/.hermes/.env`
+(`PHOTON_PROJECT_ID` = the Spectrum project id, `PHOTON_PROJECT_SECRET`),
+the same place every other channel keeps its token. Management metadata
+(device token, dashboard project id) lives in `~/.hermes/auth.json` under
+`credential_pool.photon` / `credential_pool.photon_project`.
 
 ## Authorizing users
 
@@ -131,26 +142,6 @@ Both keys also accept env vars (`PHOTON_REQUIRE_MENTION`,
 `PHOTON_MENTION_PATTERNS`). This is the same mention-gating model the
 BlueBubbles iMessage channel uses.
 
-## Registering the webhook
-
-Photon needs a public URL it can POST to. Expose your local listener
-(default port 8788, path `/photon/webhook`) via Cloudflare Tunnel or
-ngrok, then:
-
-```bash
-hermes photon webhook register https://YOUR-PUBLIC-URL/photon/webhook
-```
-
-The response includes a `signingSecret` — **Photon only returns it
-once.** Save it to `~/.hermes/.env`:
-
-```bash
-PHOTON_WEBHOOK_SECRET=v0_64-char-hex...
-```
-
-The plugin verifies every inbound `POST` against this secret and
-rejects deliveries with a timestamp drift greater than 5 minutes.
-
 ## Start the gateway
 
 ```bash
@@ -160,7 +151,7 @@ hermes gateway start --platform photon
 You'll see something like:
 
 ```
-[photon] connected — webhook at 0.0.0.0:8788/photon/webhook, sidecar on 127.0.0.1:8789
+[photon] connected — sidecar on 127.0.0.1:8789, streaming inbound over gRPC
 ```
 
 Send an iMessage to your assigned number and Hermes will reply.
@@ -177,9 +168,9 @@ Prints:
 Photon iMessage status
 ──────────────────────
   device token        : ✓ stored
-  project id          : 3c90c3cc-0d44-4b50-...
-  project key         : ✓ stored
-  webhook key         : ✓ set
+  dashboard project   : 3c90c3cc-0d44-4b50-...
+  spectrum project id : sp-...
+  project secret      : ✓ stored
   node binary         : /usr/bin/node
   sidecar deps        : ✓ installed
 ```
@@ -188,27 +179,19 @@ Common issues:
 
 - **`sidecar deps : ✗ run hermes photon install-sidecar`** — Node is
   installed but `spectrum-ts` isn't. Run the suggested command.
-- **`webhook key : ⚠ unset — verification disabled`** — the
-  plugin will accept ANY POST to the webhook URL, which is unsafe.
-  Re-run `hermes photon webhook register` and store the secret.
-- **`PHOTON_WEBHOOK_PORT` already in use** — set a different port via
-  `~/.hermes/.env`.
-- **Webhook reachable from localhost but Photon can't deliver** —
-  Photon needs a public hostname. Cloudflare Tunnel is the easiest
-  free option.
-
-## Webhook management
-
-```bash
-hermes photon webhook list                  # show registered hooks
-hermes photon webhook delete <webhook-id>   # remove one
-```
+- **`device token : ✗ missing`** — run `hermes photon setup` to log in.
+- **`No iMessage line assigned yet`** — Spectrum is enabled but no line
+  has been provisioned; re-run `hermes photon setup` or check the
+  [dashboard][app].
+- **Sidecar won't start** — confirm `node --version` is 18.17+ and that
+  `hermes photon install-sidecar` completed without errors.
 
 ## Limits today
 
-- **Inbound attachments are metadata-only.** Inbound webhooks carry the
-  filename + MIME type but no download URL — Photon documents an
-  attachment retrieval endpoint as roadmap.
+- **Inbound attachments are metadata-only.** Inbound events carry the
+  filename + MIME type; the agent sees a marker but can't yet read the
+  bytes. The SDK exposes attachment bytes via `content.read()`, so this
+  is a sidecar follow-up.
 - **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`
@@ -222,22 +205,17 @@ hermes photon webhook delete <webhook-id>   # remove one
 
 | Variable                  | Default            | Notes                                      |
 |---------------------------|--------------------|--------------------------------------------|
-| `PHOTON_PROJECT_ID`       | from `auth.json`   | Set by `hermes photon setup`               |
-| `PHOTON_PROJECT_SECRET`   | from `auth.json`   | Set by `hermes photon setup`               |
-| `PHOTON_WEBHOOK_SECRET`   | (unset)            | From `hermes photon webhook register`      |
-| `PHOTON_WEBHOOK_PORT`     | `8788`             | Local port for the aiohttp listener        |
-| `PHOTON_WEBHOOK_PATH`     | `/photon/webhook`  | Path under which the listener mounts       |
-| `PHOTON_WEBHOOK_BIND`     | `0.0.0.0`          | Bind address for the listener              |
-| `PHOTON_SIDECAR_PORT`     | `8789`             | Loopback port for sidecar control          |
+| `PHOTON_PROJECT_ID`       | from `.env`        | Spectrum project id (the SDK's `projectId`); set by setup |
+| `PHOTON_PROJECT_SECRET`   | from `.env`        | Project secret; set by setup               |
+| `PHOTON_SIDECAR_PORT`     | `8789`             | Loopback port for the sidecar control + inbound channel |
 | `PHOTON_SIDECAR_AUTOSTART`| `true`             | Whether the adapter spawns the sidecar     |
 | `PHOTON_NODE_BIN`         | `which node`       | Override the Node binary path              |
-| `PHOTON_HOME_CHANNEL`     | (unset)            | Default space ID for cron / notifications  |
+| `PHOTON_HOME_CHANNEL`     | (unset)            | Default space id for cron / notifications  |
 | `PHOTON_HOME_CHANNEL_NAME`| (unset)            | Human label for the home channel           |
 | `PHOTON_ALLOWED_USERS`    | (unset)            | Comma-separated E.164 allowlist            |
 | `PHOTON_ALLOW_ALL_USERS`  | `false`            | Dev only — accept any sender               |
 | `PHOTON_REQUIRE_MENTION`  | `false`            | Require a wake word before responding in groups |
 | `PHOTON_MENTION_PATTERNS` | Hermes wake words  | JSON list / comma / newline regex patterns for group mentions |
-| `PHOTON_API_HOST`         | `spectrum.photon.codes` | Override the Spectrum management API host |
 | `PHOTON_DASHBOARD_HOST`   | `app.photon.codes` | Override the dashboard / device-login host |
 
 [photon]: https://photon.codes/

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

@@ -54,8 +54,11 @@ SIMPLEX_HOME_CHANNEL=<contact-id>
 | `SIMPLEX_WS_URL` | Yes | WebSocket URL of the simplex-chat daemon |
 | `SIMPLEX_ALLOWED_USERS` | Recommended | Comma-separated allowlist. Each entry can be a numeric `contactId` **or** a display name — both forms work. |
 | `SIMPLEX_ALLOW_ALL_USERS` | Optional | Set `true` to allow every contact (use carefully) |
-| `SIMPLEX_HOME_CHANNEL` | Optional | Default contact ID for cron job delivery |
+| `SIMPLEX_AUTO_ACCEPT` | Optional | Auto-accept incoming contact requests (default: `true`) |
+| `SIMPLEX_GROUP_ALLOWED` | Optional | Comma-separated group IDs the bot participates in, or `*` for any group. Omit to ignore group messages entirely |
+| `SIMPLEX_HOME_CHANNEL` | Optional | Default contact/group ID for cron job delivery |
 | `SIMPLEX_HOME_CHANNEL_NAME` | Optional | Human label for the home channel |
+| `HERMES_SIMPLEX_TEXT_BATCH_DELAY` | Optional | Quiet-period seconds (default: `0.8`) used to concatenate rapid-fire inbound text messages into one event |
 
 ## Find your contact ID or display name
 
@@ -68,6 +71,37 @@ By default **all contacts are denied**. You must either:
 1. Set `SIMPLEX_ALLOWED_USERS` to a comma-separated list of `contactId`s and/or display names (e.g. `SIMPLEX_ALLOWED_USERS=4,alice` matches either contactId 4 or the contact whose display name is "alice"), or
 2. Use **DM pairing** — send any message to the bot and it will reply with a pairing code. Enter that code via `hermes pairing approve simplex <CODE>`.
 
+## Group chats
+
+By default the adapter ignores group messages — a bot in a group otherwise
+processes every member's traffic. Opt-in explicitly:
+
+```
+SIMPLEX_GROUP_ALLOWED=12,34          # specific group IDs
+# or
+SIMPLEX_GROUP_ALLOWED=*              # any group the bot is in
+```
+
+Address groups by prefixing the chat ID with `group:`, e.g.
+`simplex:group:12` in `send_message` or as a cron `deliver=` target.
+
+## Attachments
+
+The adapter supports native SimpleX attachments in both directions:
+
+- **Inbound** — incoming images, voice notes, and files are accepted via
+  the daemon's XFTP flow (`rcvFileDescrReady` → `/freceive` → wait for
+  `rcvFileComplete`) and surfaced as `MessageEvent.media_urls` with the
+  appropriate `MessageType` (`PHOTO`, `VOICE`, `TEXT` + document).
+- **Outbound** — `send_image_file`, `send_voice`, `send_document`, and
+  `send_video` all use the structured `/_send` form with `filePath`, so
+  the receiving SimpleX client renders images inline and plays voice
+  notes inline rather than offering them as downloads.
+
+Agent replies can also embed `MEDIA:/path/to/file` tags in plain text —
+the adapter strips the tag from the body and sends the file as either a
+voice note (audio extensions) or a document.
+
 ## Using SimpleX with cron jobs
 
 ```python