feat(debug): support /debug [nous|local] in the CLI/TUI slash command

The --nous flag was only wired into the argparse `hermes debug share`
subcommand. The /debug slash command (classic CLI + TUI, both via
process_command -> _handle_debug_command) built a hardcoded args
namespace with no `nous` attribute, so it always took the default
paste.rs path.

Pass cmd_original through to _handle_debug_command and parse an optional
destination word:

  /debug         -> public paste (default, unchanged)
  /debug nous    -> Nous-internal S3
  /debug local   -> stdout, no upload

local wins over nous (never touches the network); unknown words fall
back to the default. Add args_hint="[nous|local]" so help/autocomplete
surface it. New TestDebugSlashCommand covers the parsing + dispatch.

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

@@ -77,7 +77,6 @@ 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'
@@ -920,7 +919,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>
-        <LinkifiedText className="whitespace-pre-wrap" explicitOnly pretty={false} text={slashStatus.groups.output.trim()} />
+        <span className="whitespace-pre-wrap">{slashStatus.groups.output.trim()}</span>
       </MessagePrimitive.Root>
     )
   }
@@ -931,7 +930,7 @@ const SystemMessage: FC = () => {
       data-role="system"
       data-slot="aui_system-message-root"
     >
-      <LinkifiedText className="whitespace-pre-wrap" explicitOnly pretty={false} text={text} />
+      <span className="whitespace-pre-wrap">{text}</span>
     </MessagePrimitive.Root>
   )
 }

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

@@ -165,31 +165,4 @@ 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,12 +12,6 @@ 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
@@ -267,14 +261,13 @@ interface LinkifiedTextProps {
   className?: string
   text: string
   pretty?: boolean
-  explicitOnly?: boolean
 }
 
-export function LinkifiedText({ className, explicitOnly = false, pretty = true, text }: LinkifiedTextProps) {
+export function LinkifiedText({ className, pretty = true, text }: LinkifiedTextProps) {
   const nodes: ReactNode[] = []
   let cursor = 0
 
-  for (const match of text.matchAll(explicitOnly ? EXPLICIT_URL_RE : URL_RE)) {
+  for (const match of text.matchAll(URL_RE)) {
     const raw = match[0]
     const url = normalizeExternalUrl(raw)
     const index = match.index ?? 0

cli.py

@@ -7272,7 +7272,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         elif canonical == "copy":
             self._handle_copy_command(cmd_original)
         elif canonical == "debug":
-            self._handle_debug_command()
+            self._handle_debug_command(cmd_original)
         elif canonical == "update":
             if self._handle_update_command():
                 return False
@@ -7302,66 +7302,24 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
             self._handle_browser_command(cmd_original)
         elif canonical == "plugins":
             try:
-                # Discover from disk (bundled + user), matching `hermes plugins
-                # list` — so installed-but-not-enabled plugins are visible here
-                # too. The plugin manager only knows about *loaded* plugins, so
-                # using it alone made freshly-installed, not-yet-enabled plugins
-                # look like "nothing installed".
-                from hermes_cli.plugins_cmd import (
-                    _discover_all_plugins,
-                    _get_disabled_set,
-                    _get_enabled_set,
-                    _plugin_status,
-                )
-
-                entries = _discover_all_plugins()
-                enabled = _get_enabled_set()
-                disabled = _get_disabled_set()
-
-                # `/plugins` is a quick glance — default to user-installed
-                # plugins (what the user actually added). Bundled provider/
-                # platform plugins are summarized on one line; the full
-                # catalog lives behind `hermes plugins list`.
-                user_entries = [e for e in entries if e[3] != "bundled"]
-                bundled_count = len(entries) - len(user_entries)
-
-                if not user_entries:
-                    print("No user plugins installed.")
-                    print("  Install one: hermes plugins install owner/repo")
-                    print(f"  Or drop a plugin directory into {display_hermes_home()}/plugins/")
-                    if bundled_count:
-                        print(f"  ({bundled_count} bundled plugins available — see: hermes plugins list)")
+                from hermes_cli.plugins import get_plugin_manager
+                mgr = get_plugin_manager()
+                plugins = mgr.list_plugins()
+                if not plugins:
+                    print("No plugins installed.")
+                    print(f"Drop plugin directories into {display_hermes_home()}/plugins/ to get started.")
                 else:
-                    # Loaded-plugin details (tools/hooks/commands counts, errors)
-                    # keyed by name, when available.
-                    loaded: dict = {}
-                    try:
-                        from hermes_cli.plugins import get_plugin_manager
-                        for p in get_plugin_manager().list_plugins():
-                            loaded[p["name"]] = p
-                    except Exception:
-                        loaded = {}
-
-                    print(f"User plugins ({len(user_entries)}):")
-                    for name, version, _desc, source, _dir, key in sorted(user_entries):
-                        state = _plugin_status(name, enabled, disabled, key=key)
-                        glyph = {"enabled": "✓", "disabled": "✗"}.get(state, "○")
-                        ver = f" v{version}" if version else ""
-                        info = loaded.get(name) or {}
-                        bits = []
-                        if info.get("tools"):
-                            bits.append(f"{info['tools']} tools")
-                        if info.get("hooks"):
-                            bits.append(f"{info['hooks']} hooks")
-                        if info.get("commands"):
-                            bits.append(f"{info['commands']} commands")
-                        detail = f" ({', '.join(bits)})" if bits else ""
-                        label = "" if state == "enabled" else f" [{state}]"
-                        error = f" — {info['error']}" if info.get("error") else ""
-                        print(f"  {glyph} {name}{ver}{label}{detail}{error}")
-                    if bundled_count:
-                        print(f"  (+{bundled_count} bundled — see: hermes plugins list)")
-                    print("  Enable/disable: hermes plugins enable/disable <name>")
+                    print(f"Plugins ({len(plugins)}):")
+                    for p in plugins:
+                        status = "✓" if p["enabled"] else "✗"
+                        version = f" v{p['version']}" if p["version"] else ""
+                        tools = f"{p['tools']} tools" if p["tools"] else ""
+                        hooks = f"{p['hooks']} hooks" if p["hooks"] else ""
+                        commands = f"{p['commands']} commands" if p.get("commands") else ""
+                        parts = [x for x in [tools, hooks, commands] if x]
+                        detail = f" ({', '.join(parts)})" if parts else ""
+                        error = f" — {p['error']}" if p["error"] else ""
+                        print(f"  {status} {p['name']}{version}{detail}{error}")
             except Exception as e:
                 print(f"Plugin system error: {e}")
         elif canonical == "rollback":

gateway/platforms/api_server.py

@@ -3510,46 +3510,35 @@ class APIServerAdapter(BasePlatformAdapter):
         loop = asyncio.get_running_loop()
 
         def _run():
-            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 "",
+            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,
             )
-            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)
+            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
 
         return await loop.run_in_executor(None, _run)
 

gateway/session_context.py

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

hermes_cli/cli_commands_mixin.py

@@ -2090,12 +2090,25 @@ class CLICommandsMixin:
         else:
             _cprint(f"  {_ACCENT}✓ {feature_name} set to {label} (session only){_RST}")
 
-    def _handle_debug_command(self):
-        """Handle /debug — upload debug report + logs and print paste URLs."""
+    def _handle_debug_command(self, cmd_original: str = ""):
+        """Handle /debug — upload debug report + logs and print share URLs.
+
+        Accepts optional destination words after the command:
+
+        - ``/debug``        → upload to the public paste service (default)
+        - ``/debug nous``   → upload to Nous-internal storage (private, staff-only)
+        - ``/debug local``  → render the report to stdout, no upload
+
+        ``nous`` and ``local`` are mutually exclusive; if both are given,
+        ``local`` wins (it never touches the network).
+        """
         from hermes_cli.debug import run_debug_share
         from types import SimpleNamespace
 
-        args = SimpleNamespace(lines=200, expire=7, local=False)
+        words = {w.lower() for w in cmd_original.split()[1:]}
+        local = "local" in words
+        nous = "nous" in words and not local
+        args = SimpleNamespace(lines=200, expire=7, local=local, nous=nous)
         run_debug_share(args)
 
     def _handle_update_command(self) -> bool:

hermes_cli/commands.py

@@ -217,7 +217,8 @@ COMMAND_REGISTRY: list[CommandDef] = [
                cli_only=True, args_hint="<path>"),
     CommandDef("update", "Update Hermes Agent to the latest version", "Info"),
     CommandDef("version", "Show Hermes Agent version", "Info", aliases=("v",)),
-    CommandDef("debug", "Upload debug report (system info + logs) and get shareable links", "Info"),
+    CommandDef("debug", "Upload debug report (system info + logs) and get shareable links", "Info",
+               args_hint="[nous|local]"),
 
     # Exit
     CommandDef("quit", "Exit the CLI (use --delete to also remove session history)", "Exit",

hermes_cli/dashboard_register.py

@@ -94,36 +94,19 @@ def _register_self_hosted_client(
     *,
     access_token: str,
     portal_base_url: str,
-    name: Optional[str],
+    name: 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] = {}
-    if name:
-        body["name"] = name
+    body: dict[str, str] = {"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(
@@ -182,20 +165,16 @@ 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("    HERMES_DASHBOARD_OAUTH_CLIENT_ID=" + str(_cid))
+    print(f"    HERMES_DASHBOARD_OAUTH_CLIENT_ID={client_id}")
     if wrote_portal_url:
-        print("    HERMES_DASHBOARD_PORTAL_URL=" + str(portal_base_url))
-    if public_url:
-        print("    HERMES_DASHBOARD_PUBLIC_URL=" + str(public_url))
+        print(f"    HERMES_DASHBOARD_PORTAL_URL={portal_base_url}")
     print()
     print(
         "  Heads up — Nous login only *engages* on a non-loopback bind. A plain\n"
@@ -261,48 +240,12 @@ 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)
 
-    # 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()
+    name = getattr(args, "name", None) or _generate_dashboard_name()
     custom_redirect_uri = getattr(args, "redirect_uri", None)
 
     # 2. Register with the portal.
@@ -312,26 +255,20 @@ 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 or "")
+    registered_name = str(result.get("name") or 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}"')
+    print(f'✓ Registered dashboard "{registered_name}"')
 
-    # 3. Write env vars idempotently. Always set the client_id.
+    # 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).
     try:
         save_env_value("HERMES_DASHBOARD_OAUTH_CLIENT_ID", client_id)
     except Exception as exc:
@@ -339,18 +276,6 @@ 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
@@ -358,15 +283,7 @@ def cmd_dashboard_register(args) -> None:
         existing_portal = get_env_value("HERMES_DASHBOARD_PORTAL_URL")
     except Exception:
         existing_portal = None
-
-    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:
+    if not existing_portal and portal_base_url.rstrip("/") != default_portal:
         try:
             save_env_value("HERMES_DASHBOARD_PORTAL_URL", portal_base_url)
             wrote_portal_url = True
@@ -374,54 +291,10 @@ 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/debug.py

@@ -9,8 +9,16 @@ Currently supports:
                           ``~/.hermes/logs/*.log`` are not leaked into
                           the public paste service. Pass ``--no-redact``
                           to disable.
+                          Pass ``--nous`` to upload instead to Nous-internal
+                          storage (AWS S3) via a signed URL minted by the
+                          Nous account service: the bundle is private
+                          (viewable only by Nous staff / allowlisted mods via
+                          a Google-login-gated viewer) and auto-deletes after
+                          14 days, rather than going to a public paste.
 """
 
+import datetime
+import gzip
 import io
 import json
 import logging
@@ -581,6 +589,97 @@ def collect_debug_report(
     return buf.getvalue()
 
 
+# ---------------------------------------------------------------------------
+# Shared bundle collection (used by both the paste.rs and Nous-S3 paths)
+# ---------------------------------------------------------------------------
+
+# Bundle format identifier embedded in the Nous-S3 JSON envelope. The
+# discord-support viewer keys off this string to parse the bundle.
+_NOUS_BUNDLE_FORMAT = "hermes-debug-share/1"
+
+
+def collect_share_bundle(
+    log_lines: int = 200,
+    redact: bool = True,
+) -> dict[str, str]:
+    """Collect the debug report + full logs as a label→text mapping.
+
+    Returns ``{"report": ..., "agent.log": ..., "gateway.log": ...,
+    "desktop.log": ...}`` where each value is the already-redacted (when
+    ``redact`` is True) text that would be uploaded.  Keys for logs that are
+    absent/empty are simply omitted.
+
+    This is the single source of collection + redaction shared by both
+    destinations: the paste.rs path (:func:`build_debug_share`) and the
+    Nous-S3 path (``--nous``).  Centralising it guarantees the Nous bundle is
+    built from the *same* force-redacted snapshots as the public paste path —
+    redaction is the safety boundary, so the Nous path must never see raw
+    logs.
+
+    The dump header is prepended to each full log (mirroring the historical
+    paste behaviour) so every file is self-contained, and the redaction
+    banner is prepended when ``redact`` is True.
+    """
+    dump_text = _capture_dump()
+    log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact)
+
+    report = collect_debug_report(
+        log_lines=log_lines,
+        dump_text=dump_text,
+        log_snapshots=log_snapshots,
+    )
+    agent_log = log_snapshots["agent"].full_text
+    gateway_log = log_snapshots["gateway"].full_text
+    desktop_log = log_snapshots["desktop"].full_text
+
+    # Prepend dump header to each full log so every file is self-contained.
+    if agent_log:
+        agent_log = dump_text + "\n\n--- full agent.log ---\n" + agent_log
+    if gateway_log:
+        gateway_log = dump_text + "\n\n--- full gateway.log ---\n" + gateway_log
+    if desktop_log:
+        desktop_log = dump_text + "\n\n--- full desktop.log ---\n" + desktop_log
+
+    # Visible banner so reviewers know redaction was applied at upload time.
+    if redact:
+        report = _REDACTION_BANNER + report
+        if agent_log:
+            agent_log = _REDACTION_BANNER + agent_log
+        if gateway_log:
+            gateway_log = _REDACTION_BANNER + gateway_log
+        if desktop_log:
+            desktop_log = _REDACTION_BANNER + desktop_log
+
+    bundle: dict[str, str] = {"report": report}
+    if agent_log:
+        bundle["agent.log"] = agent_log
+    if gateway_log:
+        bundle["gateway.log"] = gateway_log
+    if desktop_log:
+        bundle["desktop.log"] = desktop_log
+    return bundle
+
+
+def build_nous_bundle(bundle: dict[str, str], redact: bool = True) -> bytes:
+    """Gzip-compress a :func:`collect_share_bundle` mapping into the Nous envelope.
+
+    The JSON shape is what the discord-support viewer (Repo 3) parses::
+
+        {"format": "hermes-debug-share/1",
+         "redacted": <bool>,
+         "created": <iso8601>,
+         "files": {"report": ..., "agent.log": ..., ...}}
+    """
+    created = datetime.datetime.now(datetime.timezone.utc).isoformat()
+    envelope = {
+        "format": _NOUS_BUNDLE_FORMAT,
+        "redacted": bool(redact),
+        "created": created,
+        "files": bundle,
+    }
+    return gzip.compress(json.dumps(envelope).encode("utf-8"))
+
+
 # ---------------------------------------------------------------------------
 # CLI entry points
 # ---------------------------------------------------------------------------
@@ -620,45 +719,18 @@ def build_debug_share(
     """
     _best_effort_sweep_expired_pastes()
 
-    # Capture dump once — prepended to every paste for context.
-    # The dump is already redacted at extract time via dump.py:_redact;
-    # log_snapshots are redacted by _capture_default_log_snapshots when
-    # redact=True so credentials never reach the public paste service.
-    dump_text = _capture_dump()
-    log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact)
+    # Collect the report + full logs (force-redacted when redact=True) via the
+    # shared collector so the paste.rs and Nous-S3 paths build identical,
+    # identically-redacted bundles. The dump header + redaction banner are
+    # applied inside collect_share_bundle.
+    bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
 
     if redact:
         logger.info(
             "hermes debug share: applied force-mode redaction to log snapshots before upload"
         )
 
-    report = collect_debug_report(
-        log_lines=log_lines,
-        dump_text=dump_text,
-        log_snapshots=log_snapshots,
-    )
-    agent_log = log_snapshots["agent"].full_text
-    gateway_log = log_snapshots["gateway"].full_text
-    desktop_log = log_snapshots["desktop"].full_text
-
-    # Prepend dump header to each full log so every paste is self-contained.
-    if agent_log:
-        agent_log = dump_text + "\n\n--- full agent.log ---\n" + agent_log
-    if gateway_log:
-        gateway_log = dump_text + "\n\n--- full gateway.log ---\n" + gateway_log
-    if desktop_log:
-        desktop_log = dump_text + "\n\n--- full desktop.log ---\n" + desktop_log
-
-    # Visible banner so reviewers reading the public paste know redaction
-    # was applied at upload time. Banner is omitted under --no-redact.
-    if redact:
-        report = _REDACTION_BANNER + report
-        if agent_log:
-            agent_log = _REDACTION_BANNER + agent_log
-        if gateway_log:
-            gateway_log = _REDACTION_BANNER + gateway_log
-        if desktop_log:
-            desktop_log = _REDACTION_BANNER + desktop_log
+    report = bundle["report"]
 
     urls: dict[str, str] = {}
     failures: list[str] = []
@@ -667,11 +739,8 @@ def build_debug_share(
     urls["Report"] = upload_to_pastebin(report, expiry_days=expiry)
 
     # 2-4. Full logs (optional — failures are collected, not raised)
-    for label, content in (
-        ("agent.log", agent_log),
-        ("gateway.log", gateway_log),
-        ("desktop.log", desktop_log),
-    ):
+    for label in ("agent.log", "gateway.log", "desktop.log"):
+        content = bundle.get(label)
         if not content:
             continue
         try:
@@ -696,43 +765,23 @@ def run_debug_share(args):
     log_lines = getattr(args, "lines", 200)
     expiry = getattr(args, "expire", 7)
     local_only = getattr(args, "local", False)
+    nous = getattr(args, "nous", False)
     redact = not getattr(args, "no_redact", False)
 
     if local_only:
         # Local-only path never uploads — render the report to stdout and bail
-        # before any network I/O. Mirrors the upload path's collection logic.
+        # before any network I/O. Reuses the shared collector so the rendered
+        # output matches exactly what would be uploaded.
         _best_effort_sweep_expired_pastes()
         print("Collecting debug report...")
-        dump_text = _capture_dump()
-        log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact)
-        report = collect_debug_report(
-            log_lines=log_lines,
-            dump_text=dump_text,
-            log_snapshots=log_snapshots,
-        )
-        agent_log = log_snapshots["agent"].full_text
-        gateway_log = log_snapshots["gateway"].full_text
-        desktop_log = log_snapshots["desktop"].full_text
-        if agent_log:
-            agent_log = dump_text + "\n\n--- full agent.log ---\n" + agent_log
-        if gateway_log:
-            gateway_log = dump_text + "\n\n--- full gateway.log ---\n" + gateway_log
-        if desktop_log:
-            desktop_log = dump_text + "\n\n--- full desktop.log ---\n" + desktop_log
-        if redact:
-            report = _REDACTION_BANNER + report
-            if agent_log:
-                agent_log = _REDACTION_BANNER + agent_log
-            if gateway_log:
-                gateway_log = _REDACTION_BANNER + gateway_log
-            if desktop_log:
-                desktop_log = _REDACTION_BANNER + desktop_log
-        print(report)
-        for title, body in (
-            ("FULL agent.log", agent_log),
-            ("FULL gateway.log", gateway_log),
-            ("FULL desktop.log", desktop_log),
+        bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
+        print(bundle["report"])
+        for title, label in (
+            ("FULL agent.log", "agent.log"),
+            ("FULL gateway.log", "gateway.log"),
+            ("FULL desktop.log", "desktop.log"),
         ):
+            body = bundle.get(label)
             if body:
                 print(f"\n\n{'=' * 60}")
                 print(title)
@@ -740,6 +789,10 @@ def run_debug_share(args):
                 print(body)
         return
 
+    if nous:
+        _run_debug_share_nous(log_lines=log_lines, redact=redact)
+        return
+
     print(_PRIVACY_NOTICE)
     print("Collecting debug report...")
     print("Uploading...")
@@ -773,6 +826,80 @@ def run_debug_share(args):
     print(f"\nShare these links with the Hermes team for support.")
 
 
+_NOUS_PRIVACY_NOTICE = """\
+⚠️  --nous: This uploads your debug bundle to Nous-INTERNAL storage (AWS S3),
+    NOT a public paste service. The following is included:
+  • System info (OS, Python/Hermes version, provider, which API keys are
+    configured — NOT the actual keys)
+  • Full agent.log, gateway.log, and desktop.log (up to 512 KB each — likely
+    contains conversation content, tool outputs, and file paths)
+
+  • The bundle is viewable only by Nous staff (and allowlisted Discord mods)
+    via a Google-login-gated viewer.
+  • It is NOT a public paste — there is no public URL to the contents.
+  • It auto-deletes after 14 days.
+"""
+
+
+def _run_debug_share_nous(*, log_lines: int, redact: bool) -> None:
+    """Handle ``hermes debug share --nous``: upload the bundle to Nous-S3.
+
+    Collects the same force-redacted bundle as the paste path, gzips it into
+    the Nous envelope, requests a signed URL from NAS, uploads, and prints the
+    private viewer link. On any failure falls back to a clear error that
+    suggests ``--local``.
+    """
+    from hermes_cli.diagnostics_upload import share_to_nous
+
+    print(_NOUS_PRIVACY_NOTICE)
+    if not redact:
+        print(
+            "⚠️  --no-redact is set: secrets in your logs will NOT be redacted "
+            "before upload.\n"
+        )
+    print("Collecting debug report...")
+    _best_effort_sweep_expired_pastes()
+
+    bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
+    if redact:
+        logger.info(
+            "hermes debug share --nous: applied force-mode redaction before upload"
+        )
+    blob = build_nous_bundle(bundle, redact=redact)
+
+    print("Uploading to Nous diagnostics storage...")
+    try:
+        res = share_to_nous(blob)
+    except Exception as exc:
+        print(
+            f"\nNous upload failed: {exc}\n"
+            "\nThe Nous diagnostics service may be unavailable or not yet "
+            "provisioned.\n"
+            "Run `hermes debug share --local` to print the report instead, "
+            "or `hermes debug share` to upload to a public paste service.\n",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    view_url = res.get("viewUrl") or res.get("view_url")
+    print("\nDebug bundle uploaded to Nous (private):")
+    if view_url:
+        print(f"  View URL  {view_url}")
+    else:
+        print(f"  (no view URL returned; upload id: {res.get('id', '?')})")
+
+    expires_at = res.get("expiresAt") or res.get("expires_at")
+    if expires_at:
+        print(f"\n⏱  Auto-deletes at {expires_at} (14-day retention).")
+    else:
+        print("\n⏱  Auto-deletes after 14 days.")
+
+    print(
+        "\nShare this private link with the Nous team — only Nous staff "
+        "(via Google login) can open it."
+    )
+
+
 def run_debug_delete(args):
     """Delete one or more paste URLs uploaded by /debug."""
     urls = getattr(args, "urls", [])
@@ -823,6 +950,8 @@ def run_debug(args):
         print("  --lines N    Number of log lines to include (default: 200)")
         print("  --expire N   Paste expiry in days (default: 7)")
         print("  --local      Print report locally instead of uploading")
+        print("  --nous       Upload to Nous-internal storage (private, staff-only,")
+        print("               auto-deletes in 14 days) instead of a public paste")
         print("  --no-redact  Disable upload-time secret redaction (default: redact)")
         print()
         print("Options (delete):")

hermes_cli/diagnostics_upload.py

@@ -0,0 +1,138 @@
+"""Client for uploading ``hermes debug share`` bundles to Nous-internal S3.
+
+This is the opt-in (``--nous``) destination for ``hermes debug share``.
+Unlike the public paste.rs path, bundles uploaded here go to a Nous-owned
+S3 bucket via a short-lived signed URL minted by the Nous account service
+(NAS).  The bucket auto-expires objects after 14 days, and the contents are
+only viewable by Nous staff (and allowlisted Discord mods) through a
+Google-OAuth-gated viewer.
+
+Flow:
+
+    1. POST {NAS_BASE}/api/diagnostics/upload-url  → {uploadUrl, viewUrl, id, ...}
+       (the request body carries ``sizeBytes``; NAS signs it into the presigned
+       URL's ``ContentLength``, so the PUT must send exactly that many bytes)
+    2. PUT <uploadUrl>  (the gzipped bundle, Content-Type application/gzip)
+
+NAS is stateless — the object's existence in S3 is the only state, so there is
+no confirm/callback step.
+
+Uses stdlib ``urllib`` only, matching ``debug.py`` style — no third-party deps.
+"""
+
+import json
+import os
+import urllib.request
+
+# Base URL of the Nous account service that mints the signed upload URL.
+# Overridable via env so the feature can be pointed at staging / a local dev
+# NAS instance during testing.
+NAS_BASE = os.environ.get(
+    "HERMES_DIAGNOSTICS_BASE_URL", "https://portal.nousresearch.com"
+)
+
+# Network timeout for each request (seconds). The upload itself can be larger
+# (a gzipped log bundle), so the PUT gets a more generous window.
+_REQUEST_TIMEOUT = 30
+_UPLOAD_TIMEOUT = 120
+
+_USER_AGENT = "hermes-agent/debug-share"
+
+
+def request_upload_url(
+    content_type: str = "application/gzip",
+    size_bytes: int | None = None,
+) -> dict:
+    """Ask NAS to mint a presigned PUT URL for a diagnostics bundle.
+
+    POSTs a small JSON body to ``{NAS_BASE}/api/diagnostics/upload-url`` and
+    returns the parsed JSON response, expected to contain at least
+    ``uploadUrl``, ``viewUrl`` and ``id`` (plus optional ``expiresAt`` /
+    ``uploadExpiresInSeconds``).
+
+    Raises on non-2xx responses or unparseable JSON.
+    """
+    payload: dict = {"contentType": content_type}
+    if size_bytes is not None:
+        payload["sizeBytes"] = int(size_bytes)
+
+    data = json.dumps(payload).encode("utf-8")
+    req = urllib.request.Request(
+        f"{NAS_BASE}/api/diagnostics/upload-url",
+        data=data,
+        method="POST",
+        headers={
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "User-Agent": _USER_AGENT,
+        },
+    )
+    with urllib.request.urlopen(req, timeout=_REQUEST_TIMEOUT) as resp:
+        status = getattr(resp, "status", None)
+        if status is None:
+            status = resp.getcode()
+        if not (200 <= status < 300):
+            raise RuntimeError(
+                f"diagnostics upload-url request failed: HTTP {status}"
+            )
+        body = resp.read().decode("utf-8")
+
+    try:
+        result = json.loads(body)
+    except (ValueError, json.JSONDecodeError) as exc:
+        raise RuntimeError(
+            f"diagnostics upload-url returned non-JSON response: {body[:200]}"
+        ) from exc
+
+    if not isinstance(result, dict) or not result.get("uploadUrl"):
+        raise RuntimeError(
+            "diagnostics upload-url response missing 'uploadUrl': "
+            f"{body[:200]}"
+        )
+    return result
+
+
+def put_bundle(
+    upload_url: str,
+    data: bytes,
+    content_type: str = "application/gzip",
+) -> None:
+    """PUT the gzipped *data* bundle to a presigned *upload_url*.
+
+    Sets the ``Content-Type`` header (must match what NAS pinned when signing
+    the URL, otherwise S3 rejects the signature). Raises on non-2xx.
+    """
+    req = urllib.request.Request(
+        upload_url,
+        data=data,
+        method="PUT",
+        headers={
+            "Content-Type": content_type,
+            "User-Agent": _USER_AGENT,
+        },
+    )
+    with urllib.request.urlopen(req, timeout=_UPLOAD_TIMEOUT) as resp:
+        status = getattr(resp, "status", None)
+        if status is None:
+            status = resp.getcode()
+        if not (200 <= status < 300):
+            raise RuntimeError(f"diagnostics bundle PUT failed: HTTP {status}")
+
+
+def share_to_nous(report_bundle: bytes) -> dict:
+    """Orchestrate the full Nous-S3 upload of a gzipped *report_bundle*.
+
+    Two steps: mint a presigned PUT URL (sending the exact ``sizeBytes`` NAS
+    signs into the URL's ``ContentLength``), then PUT the bundle. NAS is
+    stateless — the object's existence in S3 is the only state, so there is no
+    confirm/callback step. Returns the dict from :func:`request_upload_url`
+    (which carries ``viewUrl`` / ``id`` / expiry metadata) so the caller can
+    print the viewer link. Raises on any failure of either step.
+    """
+    size_bytes = len(report_bundle)
+    info = request_upload_url(
+        content_type="application/gzip", size_bytes=size_bytes
+    )
+    put_bundle(info["uploadUrl"], report_bundle, content_type="application/gzip")
+
+    return info

hermes_cli/subcommands/debug.py

@@ -29,6 +29,7 @@ Examples:
     hermes debug share --expire 30  Keep paste for 30 days
     hermes debug share --local      Print report locally (no upload)
     hermes debug share --no-redact  Disable upload-time secret redaction
+    hermes debug share --nous       Upload to Nous-internal storage (private)
     hermes debug delete <url>       Delete a previously uploaded paste
 """,
     )
@@ -64,6 +65,17 @@ Examples:
             "into the public paste service."
         ),
     )
+    share_parser.add_argument(
+        "--nous",
+        action="store_true",
+        help=(
+            "Upload the debug bundle to Nous-internal storage (AWS S3) instead "
+            "of a public paste service. The bundle is private — viewable only "
+            "by Nous staff (and allowlisted Discord mods) via a Google-login-"
+            "gated viewer — and auto-deletes after 14 days. Still force-redacts "
+            "secrets unless --no-redact is also passed."
+        ),
+    )
     delete_parser = debug_sub.add_parser(
         "delete",
         help="Delete a paste uploaded by 'hermes debug share'",

plugins/platforms/photon/README.md

@@ -1,136 +1,123 @@
 # Photon iMessage platform plugin
 
-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.
+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.
 
-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.
+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.
 
 ## 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.
-
 ```
-                         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) │
-                                              └──────────────────────┘
+┌─────────────────────────┐    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   │
+└─────────────────────────┘                           └──────────────────┘
 ```
 
-- **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`.
+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.
 
 ## First-time setup
 
 ```bash
-# One-shot setup: device login (opens browser) + project + user + sidecar deps
+# 1. One-shot setup: device login (opens browser) + project + user + sidecar deps
 hermes photon setup --phone +15551234567
 
-# Start the gateway
+# 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
 hermes gateway start --platform photon
 ```
 
-`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.
+`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.
 
 ## Credentials
 
-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`:
+Stored in `~/.hermes/auth.json` under `credential_pool`:
 
 ```jsonc
 {
   "credential_pool": {
     "photon": [
-      { "access_token": "<device-bearer>", "issued_at": ... }
+      { "access_token": "<dashboard-bearer>", "issued_at": ... }
     ],
     "photon_project": [
-      {
-        "dashboard_project_id": "<dashboard id>",
-        "spectrum_project_id": "<spectrumProjectId>",
-        "project_secret": "<projectSecret>",
-        "name": "Hermes Agent"
-      }
+      { "project_id": "...", "project_secret": "...", "name": "Hermes Agent" }
     ]
   }
 }
 ```
 
-> **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.
+The per-URL webhook signing secret is treated like an API key and
+lives in `~/.hermes/.env` as `PHOTON_WEBHOOK_SECRET`.
 
 ## Configuration knobs
 
-All env vars are documented in `plugin.yaml`. The most important:
+All env vars are documented in `plugin.yaml`. The most important are:
 
-| 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 |
+| 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        |
 
-## Attachments & limitations
+## Limitations (current Photon API)
 
-- **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(...))` /
+- **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(...))` /
   `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** — supported by `spectrum-ts` but not
-  yet exposed; the sidecar is the natural place to add them.
+  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.
 
 [photon]: https://photon.codes/

plugins/platforms/photon/adapter.py

@@ -1,30 +1,32 @@
 """
 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:
-    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.
+    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``.
 
 Outbound:
-    ``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.
+    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.
 """
 from __future__ import annotations
 
 import asyncio
-import base64
+import hashlib
+import hmac
 import json
 import logging
 import os
@@ -37,21 +39,21 @@ import sys
 import time
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from typing import Any, Dict, Optional
 
-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).
+try:
     import httpx
     HTTPX_AVAILABLE = True
-else:
-    try:
-        import httpx
-        HTTPX_AVAILABLE = True
-    except ImportError:  # pragma: no cover - httpx is already a Hermes dep
-        HTTPX_AVAILABLE = False
-        httpx = None
+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]
 
 from gateway.config import Platform, PlatformConfig
 from gateway.platforms.base import (
@@ -61,13 +63,21 @@ from gateway.platforms.base import (
     SendResult,
 )
 
-from .auth import load_project_credentials
+from .auth import (
+    DEFAULT_SPECTRUM_HOST,
+    load_project_credentials,
+    _spectrum_host,
+)
 
 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"
 
@@ -76,8 +86,11 @@ _DEFAULT_SIDECAR_BIND = "127.0.0.1"
 # size to ~16 KB.  Keep a conservative cap that matches BlueBubbles.
 _MAX_MESSAGE_LENGTH = 8000
 
-# Dedup parameters — the gRPC stream is at-least-once, and a sidecar
-# reconnect can replay, so keep at least 1k ids for ~48h.
+# 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_MAX_SIZE = 4000
 _DEDUP_WINDOW_SECONDS = 48 * 3600
 
@@ -105,7 +118,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:
+    if not HTTPX_AVAILABLE or not AIOHTTP_AVAILABLE:
         return False
     if not shutil.which(os.getenv("PHOTON_NODE_BIN") or "node"):
         return False
@@ -133,33 +146,61 @@ def is_connected(cfg: PlatformConfig) -> bool:
 
 
 def _env_enablement() -> Optional[dict]:
-    """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``.
-    """
+    """Seed PlatformConfig.extra from env so env-only setups appear in status."""
     project_id, project_secret = load_project_credentials()
     if not (project_id and project_secret):
         return None
-    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
+    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:])
 
 
 # ---------------------------------------------------------------------------
 # Adapter
 
 class PhotonAdapter(BasePlatformAdapter):
-    """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.
-    """
+    """Inbound: signed webhook on aiohttp. Outbound: Node sidecar via loopback HTTP."""
 
     MAX_MESSAGE_LENGTH = _MAX_MESSAGE_LENGTH
 
@@ -168,8 +209,6 @@ 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")
@@ -184,6 +223,27 @@ 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"),
@@ -199,13 +259,12 @@ 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. The gRPC stream is at-least-once, so we
-        # may see the same messageId more than once (e.g. after a reconnect).
+        # Lightweight in-memory dedup. Photon's at-least-once guarantee
+        # means we WILL see the same message.id more than once.
         self._seen_messages: Dict[str, float] = {}
 
         # Group-chat mention gating (parity with BlueBubbles). When enabled,
@@ -286,6 +345,13 @@ 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
@@ -300,11 +366,19 @@ class PhotonAdapter(BasePlatformAdapter):
             )
             return False
 
-        client = httpx.AsyncClient(timeout=30.0)
-        self._http_client = client
+        # 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
 
-        # The sidecar holds the gRPC stream for BOTH directions, so it is
-        # required now (not just for outbound).
+        # Spin up the Node sidecar (required for outbound).
         if self._autostart_sidecar:
             try:
                 await self._start_sidecar()
@@ -314,39 +388,23 @@ class PhotonAdapter(BasePlatformAdapter):
                     f"failed to start Photon sidecar: {e}",
                     retryable=True,
                 )
-                await client.aclose()
-                self._http_client = None
+                await self._stop_webhook_server()
                 return False
         else:
-            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()
-        )
+            logger.info("[photon] sidecar autostart disabled — outbound will fail")
 
+        self._http_client = httpx.AsyncClient(timeout=30.0)
         self._mark_connected()
         logger.info(
-            "[photon] connected — sidecar on %s:%d, streaming inbound over gRPC",
+            "[photon] connected — webhook at %s:%d%s, sidecar on %s:%d",
+            self._webhook_bind, self._webhook_port, self._webhook_path,
             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()
@@ -355,61 +413,68 @@ class PhotonAdapter(BasePlatformAdapter):
             self._http_client = None
         self._mark_disconnected()
 
-    # -- Inbound stream consumer ------------------------------------------
+    # -- Webhook server ----------------------------------------------------
 
-    async def _inbound_loop(self) -> None:
-        """Consume the sidecar's ``/inbound`` NDJSON stream, with reconnect.
+    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()
 
-        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:
+    async def _stop_webhook_server(self) -> None:
+        if self._runner is not None:
             try:
-                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)
+                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 def _on_inbound_line(self, line: str) -> None:
         try:
-            event = json.loads(line)
+            payload = json.loads(body or b"{}")
         except json.JSONDecodeError:
-            logger.debug("[photon] skipping non-JSON inbound line")
-            return
-        msg_id = event.get("messageId")
-        if msg_id and self._is_duplicate(msg_id):
-            return
+            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)")
+
         try:
-            await self._dispatch_inbound(event)
+            await self._dispatch_inbound(payload)
         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()
@@ -423,77 +488,44 @@ class PhotonAdapter(BasePlatformAdapter):
         self._seen_messages[msg_id] = now
         return False
 
-    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 {}
+    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 {}
 
         space_id = space.get("id") or ""
+        sender_id = sender.get("id") or ""
         if not space_id:
             logger.warning("[photon] inbound missing space.id")
             return
 
-        # 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
+        # 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"
 
-        ts_str = event.get("timestamp") or ""
+        # Timestamp — ISO 8601 from the platform.
+        ts_str = msg.get("timestamp") or ""
         try:
-            timestamp = (
-                datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
-                if ts_str
-                else datetime.now(tz=timezone.utc)
-            )
+            timestamp = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
         except ValueError:
             timestamp = datetime.now(tz=timezone.utc)
 
-        # 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":
+        # 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":
             text = content.get("text") or ""
             mtype = MessageType.TEXT
-        elif ctype == "attachment":
+        elif content.get("type") == "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: {ctype}]"
+            text = f"[Photon content type not handled: {content.get('type')}]"
             mtype = MessageType.TEXT
 
         # Group-mention gating (parity with BlueBubbles). In group chats with
@@ -513,20 +545,18 @@ class PhotonAdapter(BasePlatformAdapter):
             chat_id=space_id,
             chat_name=space_id,
             chat_type=chat_type,
-            user_id=sender_id,
+            user_id=sender_id or space_id,
             user_name=sender_id or None,
         )
-        message_event = MessageEvent(
+        event = MessageEvent(
             text=text,
             message_type=mtype,
             source=source,
-            message_id=event.get("messageId"),
-            raw_message=event,
+            message_id=msg.get("id"),
+            raw_message=payload,
             timestamp=timestamp,
-            media_urls=media_urls,
-            media_types=media_types,
         )
-        await self.handle_message(message_event)
+        await self.handle_message(event)
 
     # -- Sidecar lifecycle -------------------------------------------------
 
@@ -640,7 +670,7 @@ class PhotonAdapter(BasePlatformAdapter):
         reply_to: Optional[str] = None,
         metadata: Optional[Dict[str, Any]] = None,
     ) -> SendResult:
-        return await self._sidecar_send(chat_id, content)
+        return await self._sidecar_send(chat_id, content, reply_to=reply_to)
 
     # -- Outbound media (parity with the BlueBubbles iMessage channel) -----
     #
@@ -666,7 +696,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,
+            chat_id, local_path, caption=caption, reply_to=reply_to,
         )
 
     async def send_image_file(
@@ -679,7 +709,7 @@ class PhotonAdapter(BasePlatformAdapter):
         **kwargs,
     ) -> SendResult:
         return await self._sidecar_send_attachment(
-            chat_id, image_path, caption=caption,
+            chat_id, image_path, caption=caption, reply_to=reply_to,
         )
 
     async def send_voice(
@@ -692,7 +722,7 @@ class PhotonAdapter(BasePlatformAdapter):
         **kwargs,
     ) -> SendResult:
         return await self._sidecar_send_attachment(
-            chat_id, audio_path, caption=caption, kind="voice",
+            chat_id, audio_path, caption=caption, reply_to=reply_to, kind="voice",
         )
 
     async def send_video(
@@ -705,7 +735,7 @@ class PhotonAdapter(BasePlatformAdapter):
         **kwargs,
     ) -> SendResult:
         return await self._sidecar_send_attachment(
-            chat_id, video_path, caption=caption,
+            chat_id, video_path, caption=caption, reply_to=reply_to,
         )
 
     async def send_document(
@@ -719,7 +749,7 @@ class PhotonAdapter(BasePlatformAdapter):
         **kwargs,
     ) -> SendResult:
         return await self._sidecar_send_attachment(
-            chat_id, file_path, name=file_name, caption=caption,
+            chat_id, file_path, name=file_name, caption=caption, reply_to=reply_to,
         )
 
     async def send_animation(
@@ -737,29 +767,23 @@ class PhotonAdapter(BasePlatformAdapter):
 
     async def send_typing(self, chat_id: str, metadata=None) -> None:
         try:
-            await self._sidecar_call(
-                "/typing", {"spaceId": chat_id, "state": "start"}
-            )
+            await self._sidecar_call("/typing", {"spaceId": chat_id})
         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; the inbound event also carries the
-        DM/group type, but here we only have the id, so infer conservatively.
+        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.
         """
-        return {"name": chat_id, "type": "dm", "id": chat_id}
+        chat_type = "group" if ";+;" in chat_id else "dm"
+        return {"name": chat_id, "type": chat_type, "id": chat_id}
 
-    async def _sidecar_send(self, space_id: str, text: str) -> SendResult:
+    async def _sidecar_send(
+        self, space_id: str, text: str, *, reply_to: Optional[str] = None,
+    ) -> SendResult:
         if len(text) > self.MAX_MESSAGE_LENGTH:
             logger.warning(
                 "[photon] truncating outbound from %d to %d chars",
@@ -767,6 +791,8 @@ 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:
@@ -781,6 +807,7 @@ 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.
@@ -815,6 +842,8 @@ 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:
@@ -858,81 +887,11 @@ 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.  Reuses a live sidecar already listening on the
-# configured port (cron processes cannot spawn the sidecar themselves).
+# 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.
 
 async def _standalone_send(
     pconfig: PlatformConfig,
@@ -1021,7 +980,7 @@ def register(ctx) -> None:
 
     ctx.register_platform(
         name="photon",
-        label="iMessage via Photon",
+        label="Photon iMessage",
         adapter_factory=lambda cfg: PhotonAdapter(cfg),
         check_fn=check_requirements,
         validate_config=validate_config,
@@ -1052,7 +1011,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."
+            "Attachments arrive as metadata only (no download URL yet)."
         ),
     )
 

plugins/platforms/photon/auth.py

@@ -1,37 +1,27 @@
 """
-Photon Dashboard API client + device-code login flow.
+Photon Dashboard + Spectrum API client and device-code login flow.
 
 This module is pure Python — it intentionally does not depend on
-``spectrum-ts``.  Every management-plane operation (login, find/create
-project, enable Spectrum, rotate the project secret, register a user,
-list the assigned iMessage line) talks to Photon's **Dashboard API** on a
-single host, exactly like the official Photon CLI (``photon-hq/cli``):
+``spectrum-ts``.  All management-plane operations (login, create
+project, create user, register webhook) talk to Photon's HTTP API
+directly:
 
     Dashboard API   https://app.photon.codes/api/...
-                    OAuth 2.0 device flow, Bearer access token
+                    OAuth bearer token from device flow
 
-A Photon project carries two distinct identifiers:
+    Spectrum API    https://spectrum.photon.codes/projects/{id}/...
+                    HTTP Basic with (projectId, projectSecret)
 
-    * ``id``                — the Dashboard project id (used in API paths)
-    * ``spectrumProjectId`` — the Spectrum Cloud project id, populated when
-                              Spectrum is enabled on the project
+The webhook receiver + Node sidecar in ``adapter.py`` consume the
+credentials this module persists to ``~/.hermes/auth.json``.
 
-The ``spectrum-ts`` SDK (run by the Node sidecar) authenticates to Spectrum
-Cloud with ``(spectrumProjectId, projectSecret)`` — so the value we persist
-as ``PHOTON_PROJECT_ID`` for the runtime is the **spectrumProjectId**, not
-the Dashboard ``id``.  The Dashboard ``id`` is kept only for management
-calls.
-
-Credential storage mirrors every other Hermes channel:
-
-    * runtime SDK creds  -> ``~/.hermes/.env``  (``PHOTON_PROJECT_ID`` =
-      spectrumProjectId, ``PHOTON_PROJECT_SECRET``) via ``save_env_value``
-    * management metadata -> ``~/.hermes/auth.json`` under
-      ``credential_pool.photon`` (device token) and
-      ``credential_pool.photon_project`` (dashboard id, spectrum id, name)
-
-Reference: https://github.com/photon-hq/cli and
-https://photon.codes/docs/api-reference/device-login/request-device-+-user-code
+Reference docs (read at integration time):
+  https://photon.codes/docs/api-reference/introduction
+  https://photon.codes/docs/api-reference/device-login/request-device-+-user-code
+  https://photon.codes/docs/api-reference/device-login/exchange-device-code-for-token
+  https://photon.codes/docs/api-reference/projects/create-project
+  https://photon.codes/docs/api-reference/users/create-user
+  https://photon.codes/docs/webhooks/overview
 """
 from __future__ import annotations
 
@@ -42,7 +32,7 @@ import re
 import time
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, Optional, Tuple
 
 try:
     import httpx
@@ -61,20 +51,17 @@ class PhotonDashboardAuthError(RuntimeError):
 # Hosted Photon allowlists registered device clients on the device-code
 # endpoint — an unregistered client_id is rejected with
 # `400 {"error":"invalid_client"}`.  Use Photon's published CLI device
-# client (matches `CLI_CLIENT_ID` in photon-hq/cli) until the dashboard API
-# registers Hermes as its own client_id.
+# client until the dashboard API registers Hermes as its own client_id.
 DEFAULT_CLIENT_ID = "photon-cli"
 DEFAULT_SCOPE = "openid profile email"
 
 DEFAULT_DASHBOARD_HOST = "https://app.photon.codes"
+DEFAULT_SPECTRUM_HOST = "https://spectrum.photon.codes"
 
-# Default name of the project Hermes provisions for the operator.
-DEFAULT_PROJECT_NAME = "Hermes Agent"
-
-# Polling defaults per RFC 8628.  Photon overrides via `interval` /
-# `expires_in` in the device-code response — those win.
+# Polling defaults per RFC 8628.  Photon may override via `interval` /
+# `expires_in` fields in the device-code response — those win.
 DEFAULT_POLL_INTERVAL = 5
-DEFAULT_POLL_TIMEOUT = 1800  # 30 min, matching the CLI's fallback
+DEFAULT_POLL_TIMEOUT = 900  # 15 minutes is conservative; Photon returns expires_in
 
 E164_RE = re.compile(r"^\+[1-9]\d{6,14}$")
 
@@ -117,7 +104,7 @@ def _save_auth(data: Dict[str, Any]) -> None:
 
 
 def load_photon_token() -> Optional[str]:
-    """Return the device-flow bearer token stored by ``login()`` or ``None``."""
+    """Return the bearer token stored by ``login()`` or ``None``."""
     auth = _load_auth()
     pool = auth.get("credential_pool", {}).get("photon") or []
     if isinstance(pool, list) and pool:
@@ -141,13 +128,7 @@ def store_photon_token(token: str) -> None:
 
 
 def load_project_credentials() -> Tuple[Optional[str], Optional[str]]:
-    """Return the runtime SDK creds ``(spectrum_project_id, project_secret)``.
-
-    Precedence: process env (``~/.hermes/.env`` is loaded into the gateway's
-    environment at startup) wins, then ``auth.json`` for offline / status
-    use.  This is the pair the Node sidecar feeds to ``spectrum-ts`` — the id
-    is the **spectrumProjectId**, not the Dashboard id.
-    """
+    """Return ``(project_id, project_secret)`` from auth.json + env override."""
     env_id = os.getenv("PHOTON_PROJECT_ID")
     env_sec = os.getenv("PHOTON_PROJECT_SECRET")
     if env_id and env_sec:
@@ -156,72 +137,24 @@ def load_project_credentials() -> Tuple[Optional[str], Optional[str]]:
     proj = auth.get("credential_pool", {}).get("photon_project") or []
     if isinstance(proj, list) and proj:
         entry = proj[0]
-        # back-compat: old records used "project_id" for the spectrum id
-        sid = entry.get("spectrum_project_id") or entry.get("project_id")
-        return (env_id or sid, env_sec or entry.get("project_secret"))
+        return (
+            env_id or entry.get("project_id"),
+            env_sec or entry.get("project_secret"),
+        )
     return env_id, env_sec
 
 
-def load_dashboard_project_id() -> Optional[str]:
-    """Return the Dashboard project id (for management API calls)."""
-    env_id = os.getenv("PHOTON_DASHBOARD_PROJECT_ID")
-    if env_id:
-        return env_id
+def store_project_credentials(project_id: str, project_secret: str, **extra: Any) -> None:
+    """Persist the Spectrum project's id+secret under ``credential_pool.photon_project``."""
     auth = _load_auth()
-    proj = auth.get("credential_pool", {}).get("photon_project") or []
-    if isinstance(proj, list) and proj:
-        return proj[0].get("dashboard_project_id") or proj[0].get("project_id")
-    return None
-
-
-def store_project_credentials(
-    *,
-    spectrum_project_id: str,
-    project_secret: str,
-    dashboard_project_id: Optional[str] = None,
-    name: Optional[str] = None,
-) -> None:
-    """Persist project credentials to both .env (runtime) and auth.json (mgmt).
-
-    The runtime SDK creds land in ``~/.hermes/.env`` via the same
-    ``save_env_value`` helper every other channel uses, so the gateway picks
-    them up from the environment with zero adapter changes.  A copy of the
-    non-secret ids (plus the secret, for offline ``status``) is written to
-    ``auth.json`` so management commands work even when ``.env`` hasn't been
-    loaded into the current process.
-    """
-    auth = _load_auth()
-    record: Dict[str, Any] = {
-        "spectrum_project_id": spectrum_project_id,
+    record = {
+        "project_id": project_id,
         "project_secret": project_secret,
         "issued_at": int(time.time()),
     }
-    if dashboard_project_id:
-        record["dashboard_project_id"] = dashboard_project_id
-    if name:
-        record["name"] = name
+    record.update(extra)
     auth.setdefault("credential_pool", {})["photon_project"] = [record]
     _save_auth(auth)
-    _persist_runtime_env(spectrum_project_id, project_secret)
-
-
-def _persist_runtime_env(spectrum_project_id: str, project_secret: str) -> None:
-    """Write the SDK creds to ``~/.hermes/.env`` (canonical runtime store).
-
-    Isolated in its own helper so the secret value flows straight into
-    ``save_env_value`` without ever being bound to a printable local in a
-    caller — same CodeQL-clean-flow rationale as the rest of this module.
-    """
-    try:
-        from hermes_cli.config import save_env_value
-    except ImportError:
-        logger.warning("photon: hermes_cli.config unavailable — skipping .env write")
-        return
-    try:
-        save_env_value("PHOTON_PROJECT_ID", spectrum_project_id)
-        save_env_value("PHOTON_PROJECT_SECRET", project_secret)
-    except Exception as e:  # pragma: no cover - defensive
-        logger.warning("photon: could not write project creds to .env: %s", e)
 
 
 # ---------------------------------------------------------------------------
@@ -248,8 +181,8 @@ def _dashboard_host() -> str:
     return (os.getenv("PHOTON_DASHBOARD_HOST") or DEFAULT_DASHBOARD_HOST).rstrip("/")
 
 
-def _bearer(token: str) -> Dict[str, str]:
-    return {"Authorization": f"Bearer {token}"}
+def _spectrum_host() -> str:
+    return (os.getenv("PHOTON_API_HOST") or DEFAULT_SPECTRUM_HOST).rstrip("/")
 
 
 def request_device_code(
@@ -285,22 +218,16 @@ def poll_for_token(
 ) -> str:
     """Poll ``/api/auth/device/token`` until the user approves.
 
-    Mirrors the official CLI's polling loop: sleep first, then poll;
-    ``authorization_pending`` keeps the interval, ``slow_down`` adds 5s,
-    HTTP 429 adds 10s, and ``access_denied`` / ``expired_token`` abort.
-
-    The bearer token comes from the response body's top-level
-    ``access_token`` (better-auth device-grant shape), with
-    ``session.access_token`` and the ``set-auth-token`` header kept as
-    fallbacks for API drift.
+    Returns the bearer token from the ``set-auth-token`` response header
+    (Photon's documented mechanism).  Falls back to ``session.access_token``
+    in the JSON body if the header is absent — see the API spec.
     """
     if httpx is None:
         raise RuntimeError("httpx is required for Photon device login")
     url = f"{_dashboard_host()}/api/auth/device/token"
     deadline = time.time() + (timeout or code.expires_in or DEFAULT_POLL_TIMEOUT)
-    sleep = interval if interval is not None else (code.interval or DEFAULT_POLL_INTERVAL)
+    sleep = interval or code.interval or DEFAULT_POLL_INTERVAL
     while time.time() < deadline:
-        time.sleep(sleep)
         try:
             resp = httpx.post(
                 url,
@@ -313,6 +240,7 @@ def poll_for_token(
             )
         except httpx.RequestError as e:
             logger.warning("photon: device-token poll failed: %s", e)
+            time.sleep(sleep)
             continue
         if resp.status_code == 200:
             body: Dict[str, Any] = {}
@@ -331,35 +259,34 @@ def poll_for_token(
                     "data.access_token, accessToken, or set-auth-token)."
                 )
             return candidates[0].token
-        if resp.status_code == 429:
-            # RFC 8628 §3.5 — treat 429 as slow_down.
-            sleep += 10
-            if on_pending:
-                _safe(on_pending)
-            continue
         if resp.status_code == 400:
-            body = {}
+            # RFC 8628 §3.5 — error codes are returned with 400.
+            body: Dict[str, Any] = {}
             try:
                 body = resp.json() or {}
             except json.JSONDecodeError:
                 pass
             err = body.get("error") or body.get("message") or ""
-            if err == "authorization_pending":
+            if err in ("authorization_pending", "slow_down"):
                 if on_pending:
-                    _safe(on_pending)
-                continue
-            if err == "slow_down":
-                sleep += 5
-                if on_pending:
-                    _safe(on_pending)
+                    try:
+                        on_pending()
+                    except Exception:
+                        pass
+                if err == "slow_down":
+                    sleep += 5
+                time.sleep(sleep)
                 continue
             if err in ("expired_token", "access_denied"):
                 raise RuntimeError(f"Photon login failed: {err}")
+            # Unknown error — surface it
             raise RuntimeError(f"Photon device token error: {err or resp.text}")
+        # Unexpected status; log and retry
         logger.warning(
             "photon: device-token unexpected status %s: %s",
             resp.status_code, resp.text[:200],
         )
+        time.sleep(sleep)
     raise TimeoutError("Photon device login timed out")
 
 
@@ -499,13 +426,6 @@ def _validated_dashboard_token(candidates: list) -> str:
     raise RuntimeError("Photon did not return a usable dashboard token")
 
 
-def _safe(fn: Callable[[], None]) -> None:
-    try:
-        fn()
-    except Exception:
-        pass
-
-
 def login_device_flow(
     *,
     client_id: str = DEFAULT_CLIENT_ID,
@@ -514,12 +434,15 @@ def login_device_flow(
 ) -> str:
     """Run the full device-code login flow and persist the token.
 
-    Returns the bearer token.  ``on_user_code`` receives the
-    :class:`DeviceCode` so callers can print it + optionally open a browser.
+    Returns the bearer token.  ``on_user_code`` is a callback receiving the
+    :class:`DeviceCode` so callers can print + optionally open the browser.
     """
     code = request_device_code(client_id=client_id)
     if on_user_code:
-        _safe(lambda: on_user_code(code))
+        try:
+            on_user_code(code)
+        except Exception:
+            pass
     if open_browser:
         try:
             import webbrowser
@@ -538,335 +461,280 @@ def login_device_flow(
     return token
 
 
-def get_session(token: str) -> Dict[str, Any]:
-    """GET ``/api/auth/get-session`` — confirm the token + fetch the user."""
-    if httpx is None:
-        raise RuntimeError("httpx is required for Photon")
-    url = f"{_dashboard_host()}/api/auth/get-session"
-    resp = httpx.get(url, headers=_bearer(token), timeout=30.0)
-    resp.raise_for_status()
-    return resp.json() or {}
-
-
 # ---------------------------------------------------------------------------
-# Dashboard API: projects
-
-def _unwrap_list(data: Any) -> List[Dict[str, Any]]:
-    if isinstance(data, list):
-        return data
-    if isinstance(data, dict):
-        for key in ("data", "projects", "users", "lines", "items"):
-            inner = data.get(key)
-            if isinstance(inner, list):
-                return inner
-    return []
-
-
-def list_projects(token: str) -> List[Dict[str, Any]]:
-    """GET ``/api/projects`` — return the caller's projects."""
-    if httpx is None:
-        raise RuntimeError("httpx is required for Photon")
-    url = f"{_dashboard_host()}/api/projects"
-    resp = httpx.get(url, headers=_bearer(token), timeout=30.0)
-    resp.raise_for_status()
-    return _unwrap_list(resp.json())
-
-
-def find_project_by_name(token: str, name: str) -> Optional[Dict[str, Any]]:
-    """Return the first project whose name matches (case-insensitive)."""
-    target = (name or "").strip().lower()
-    for proj in list_projects(token):
-        if (proj.get("name") or "").strip().lower() == target:
-            return proj
-    return None
-
-
-def get_project(token: str, project_id: str) -> Dict[str, Any]:
-    """GET ``/api/projects/{id}`` — includes ``spectrum`` + ``spectrumProjectId``."""
-    if httpx is None:
-        raise RuntimeError("httpx is required for Photon")
-    url = f"{_dashboard_host()}/api/projects/{project_id}"
-    resp = httpx.get(url, headers=_bearer(token), timeout=30.0)
-    resp.raise_for_status()
-    return resp.json() or {}
-
+# Dashboard API: create project
 
 def create_project(
     token: str,
     *,
-    name: str = DEFAULT_PROJECT_NAME,
+    name: str,
     location: str = "United States",
+    platforms: Optional[list] = None,
 ) -> Dict[str, Any]:
-    """POST ``/api/projects`` with ``spectrum: true`` and return ``{success, id}``."""
+    """POST ``/api/projects/`` with ``spectrum: true`` and return the response.
+
+    The response includes ``spectrumProjectId`` and ``projectSecret`` — those
+    are the HTTP Basic credentials for the Spectrum API.  Photon only
+    returns ``projectSecret`` to project owners at creation time.
+    """
     if httpx is None:
         raise RuntimeError("httpx is required for Photon project creation")
-    url = f"{_dashboard_host()}/api/projects"
+    url = f"{_dashboard_host()}/api/projects/"
     body: Dict[str, Any] = {
         "name": name,
         "location": location,
         "spectrum": True,
-        "template": False,
-        "observability": False,
+        "platforms": platforms or ["imessage"],
     }
-    resp = httpx.post(url, json=body, headers=_bearer(token), timeout=30.0)
+    resp = httpx.post(
+        url,
+        json=body,
+        headers={"Authorization": f"Bearer {token}"},
+        timeout=30.0,
+    )
     resp.raise_for_status()
-    data = resp.json() or {}
-    if data.get("error"):
-        raise RuntimeError(f"Photon create-project failed: {data['error']}")
-    if not data.get("id"):
-        raise RuntimeError("Photon create-project did not return a project id")
-    return data
-
-
-def ensure_spectrum_enabled(token: str, project_id: str) -> Dict[str, Any]:
-    """Enable Spectrum on the project if needed; return the project dict.
-
-    The dashboard exposes Spectrum as a toggle, so we only flip it when
-    ``spectrum`` is currently false, then re-fetch to pick up the freshly
-    populated ``spectrumProjectId``.
-    """
-    if httpx is None:
-        raise RuntimeError("httpx is required for Photon")
-    proj = get_project(token, project_id)
-    if not proj.get("spectrum"):
-        url = f"{_dashboard_host()}/api/projects/{project_id}/spectrum/toggle"
-        resp = httpx.post(url, json={}, headers=_bearer(token), timeout=30.0)
-        resp.raise_for_status()
-        proj = get_project(token, project_id)
-    if not proj.get("spectrumProjectId"):
-        raise RuntimeError(
-            "Spectrum is enabled but the project has no spectrumProjectId yet — "
-            "retry in a moment, or enable Spectrum from the dashboard."
-        )
-    return proj
-
-
-def regenerate_project_secret(token: str, project_id: str) -> str:
-    """POST ``/api/projects/{id}/regenerate-secret`` → the new project secret.
-
-    This is the only way to read a project secret (the dashboard shows it
-    exactly once), so callers should persist the returned value immediately.
-    """
-    if httpx is None:
-        raise RuntimeError("httpx is required for Photon")
-    url = f"{_dashboard_host()}/api/projects/{project_id}/regenerate-secret"
-    resp = httpx.post(url, json={}, headers=_bearer(token), timeout=30.0)
-    resp.raise_for_status()
-    data = resp.json() or {}
-    if data.get("error"):
-        raise RuntimeError(f"Photon regenerate-secret failed: {data['error']}")
-    secret = data.get("projectSecret")
-    if not secret:
-        raise RuntimeError("Photon regenerate-secret returned no projectSecret")
-    return str(secret)
+    return resp.json()
 
 
 # ---------------------------------------------------------------------------
-# Dashboard API: spectrum users
-
-def _normalize_phone(phone: str) -> str:
-    """Reduce a phone string to ``+`` and digits for dedup comparison."""
-    return re.sub(r"[^\d+]", "", phone or "")
-
-
-def list_users(token: str, project_id: str) -> List[Dict[str, Any]]:
-    """GET ``/api/projects/{id}/spectrum/users`` → ``SpectrumUser[]``."""
-    if httpx is None:
-        raise RuntimeError("httpx is required for Photon")
-    url = f"{_dashboard_host()}/api/projects/{project_id}/spectrum/users"
-    resp = httpx.get(url, headers=_bearer(token), timeout=30.0)
-    resp.raise_for_status()
-    return _unwrap_list(resp.json())
-
-
-def find_user_by_phone(
-    token: str, project_id: str, phone_number: str,
-) -> Optional[Dict[str, Any]]:
-    """Return an existing Spectrum user with the given phone number, or None."""
-    target = _normalize_phone(phone_number)
-    for user in list_users(token, project_id):
-        if _normalize_phone(user.get("phoneNumber") or "") == target:
-            return user
-    return None
-
+# Spectrum API: create user
 
 def create_user(
-    token: str,
     project_id: str,
+    project_secret: str,
     *,
     phone_number: str,
+    user_type: str = "shared",
     first_name: Optional[str] = None,
     last_name: Optional[str] = None,
     email: Optional[str] = None,
-    send_invite: bool = False,
+    assigned_phone_number: Optional[str] = None,
 ) -> Dict[str, Any]:
-    """POST ``/api/projects/{id}/spectrum/users`` and return the created user."""
+    """POST ``/projects/{id}/users/`` on the Spectrum API.
+
+    For free users we always pass ``type=shared``; Photon's Cosmos pool
+    assigns the iMessage line.  ``assigned_phone_number`` is only valid
+    for the paid ``dedicated`` mode.
+    """
     if httpx is None:
         raise RuntimeError("httpx is required for Photon user creation")
     if not E164_RE.match(phone_number):
         raise ValueError(
             f"phone_number must be E.164 (e.g. +15551234567); got {phone_number!r}"
         )
-    url = f"{_dashboard_host()}/api/projects/{project_id}/spectrum/users"
-    body: Dict[str, Any] = {"phoneNumber": phone_number, "sendInvite": send_invite}
+    url = f"{_spectrum_host()}/projects/{project_id}/users/"
+    body: Dict[str, Any] = {"type": user_type, "phoneNumber": phone_number}
     if first_name:
         body["firstName"] = first_name
     if last_name:
         body["lastName"] = last_name
     if email:
         body["email"] = email
-    resp = httpx.post(url, json=body, headers=_bearer(token), timeout=30.0)
-    resp.raise_for_status()
-    data = resp.json() or {}
-    if data.get("error"):
-        raise RuntimeError(f"Photon create-user failed: {data['error']}")
-    return data.get("user") or data
-
-
-def register_user_if_absent(
-    token: str,
-    project_id: str,
-    *,
-    phone_number: str,
-    first_name: Optional[str] = None,
-    last_name: Optional[str] = None,
-    email: Optional[str] = None,
-) -> Tuple[Dict[str, Any], bool]:
-    """Idempotently register a Spectrum user.
-
-    Returns ``(user, created)`` — ``created`` is False when a user with the
-    same phone number already exists (the official CLI does no dedup, so we
-    add it here to make ``setup`` safely re-runnable).
-    """
-    existing = find_user_by_phone(token, project_id, phone_number)
-    if existing is not None:
-        return existing, False
-    user = create_user(
-        token, project_id,
-        phone_number=phone_number,
-        first_name=first_name,
-        last_name=last_name,
-        email=email,
-    )
-    return user, True
-
-
-def user_assigned_line(user: Optional[Dict[str, Any]]) -> Optional[str]:
-    """Return the iMessage number a Spectrum user is assigned to text on.
-
-    This is the user's ``assignedPhoneNumber`` (the dashboard's "TEXTS ON"
-    column) — i.e. the number to text to reach the agent, as opposed to the
-    user's own ``phoneNumber``. On shared-number plans there is no dedicated
-    entry in ``/lines``, so this per-user field is the source of truth.
-    Returns ``None`` when unset (e.g. a freshly created, not-yet-assigned user).
-    """
-    if not user:
-        return None
-    val = user.get("assignedPhoneNumber")
-    return str(val) if val else None
-
-
-# ---------------------------------------------------------------------------
-# Dashboard API: iMessage lines (the assigned number inventory)
-
-def list_lines(token: str, project_id: str) -> List[Dict[str, Any]]:
-    """GET ``/api/projects/{id}/lines`` → ``[{id, platform, phoneNumber, status}]``."""
-    if httpx is None:
-        raise RuntimeError("httpx is required for Photon")
-    url = f"{_dashboard_host()}/api/projects/{project_id}/lines"
-    resp = httpx.get(url, headers=_bearer(token), timeout=30.0)
-    resp.raise_for_status()
-    return _unwrap_list(resp.json())
-
-
-def add_line(
-    token: str, project_id: str, *, platform: str = "imessage",
-) -> Dict[str, Any]:
-    """POST ``/api/projects/{id}/lines`` to provision a new line."""
-    if httpx is None:
-        raise RuntimeError("httpx is required for Photon")
-    url = f"{_dashboard_host()}/api/projects/{project_id}/lines"
+    if assigned_phone_number:
+        body["assignedPhoneNumber"] = assigned_phone_number
     resp = httpx.post(
-        url, json={"platform": platform}, headers=_bearer(token), timeout=30.0,
+        url,
+        json=body,
+        auth=(project_id, project_secret),
+        timeout=30.0,
     )
     resp.raise_for_status()
     data = resp.json() or {}
-    if data.get("error"):
-        raise RuntimeError(f"Photon add-line failed: {data['error']}")
-    return data.get("line") or data
-
-
-def get_imessage_line(
-    token: str, project_id: str, *, create_if_missing: bool = True,
-) -> Optional[Dict[str, Any]]:
-    """Return the project's iMessage line (the number to text the agent).
-
-    If none exists and ``create_if_missing`` is set, provision one.  Returns
-    ``None`` if there is no line and provisioning failed.
-    """
-    for line in list_lines(token, project_id):
-        if (line.get("platform") or "").lower() == "imessage":
-            return line
-    if create_if_missing:
-        try:
-            return add_line(token, project_id, platform="imessage")
-        except Exception as e:
-            logger.warning("photon: could not auto-provision iMessage line: %s", e)
-            return None
-    return None
+    if not data.get("succeed"):
+        raise RuntimeError(
+            f"Photon create-user failed: {data.get('message') or data}"
+        )
+    return data.get("data") or {}
 
 
 # ---------------------------------------------------------------------------
-# Credential status (display-only — never emits raw secret material)
+# Spectrum API: webhook registration
+#
+# Endpoints from https://photon.codes/docs/webhooks/overview:
+#   POST   /projects/{id}/webhooks/          register, returns signing secret ONCE
+#   GET    /projects/{id}/webhooks/          list
+#   DELETE /projects/{id}/webhooks/{wid}     remove
+
+def register_webhook(
+    project_id: str, project_secret: str, *, webhook_url: str,
+) -> Dict[str, Any]:
+    """Register a webhook URL with Photon and return the API response.
+
+    Photon returns the per-URL signing secret exactly once in this
+    response, so callers who need to persist it should hand the
+    response to :func:`persist_webhook_signing_secret` immediately —
+    that helper writes the value into ``~/.hermes/.env`` (mode 0o600,
+    existing entries preserved) without the secret value ever needing
+    to leave this module.
+    """
+    if httpx is None:
+        raise RuntimeError("httpx is required for Photon webhook registration")
+    url = f"{_spectrum_host()}/projects/{project_id}/webhooks/"
+    resp = httpx.post(
+        url,
+        json={"webhookUrl": webhook_url},
+        auth=(project_id, project_secret),
+        timeout=30.0,
+    )
+    resp.raise_for_status()
+    data = resp.json() or {}
+    if not data.get("succeed"):
+        raise RuntimeError(
+            f"Photon register-webhook failed: {data.get('message') or data}"
+        )
+    return data.get("data") or {}
+
 
 def print_credential_summary(emit: Any = print) -> None:
     """Pretty-print the credential status table via the *emit* callback.
 
-    Every secret-bearing read is reduced to a display literal inside this
-    function (``"✓ stored"`` / ``"✗ missing"`` / a non-secret id); the
-    callback only ever receives the assembled banner string, so no tainted
-    value escapes into the caller's scope.
+    Same isolation rationale as :func:`persist_webhook_signing_secret`:
+    all secret-bearing reads happen inside this function; the *emit*
+    callback only ever receives display literals like ``"✓ stored"``
+    or a project UUID. No tainted variable ever escapes into the
+    caller's scope. Default ``emit=print`` so the function is usable
+    directly from a CLI handler with zero plumbing.
     """
+    # Resolve every credential read into a plain display string FIRST,
+    # in a tight block. The intermediate `labels` dict only ever stores
+    # literals from a finite set ("✓ stored" / "✗ missing" / "✓ set" /
+    # "⚠ unset — verification disabled" / a project UUID) — never a
+    # credential's raw bytes. We then assemble the whole banner into
+    # one string and call emit() exactly once with that string, so the
+    # static taint analyzer sees a single sink that consumes only a
+    # joined literal blob.
     labels: Dict[str, str] = {}
-    labels["device_token"] = (
-        "✓ stored" if load_photon_token()
-        else "✗ missing (run `hermes photon setup`)"
-    )
-    sid, sec = load_project_credentials()
-    labels["spectrum_project_id"] = sid if sid else "✗ missing"
-    labels["dashboard_project_id"] = load_dashboard_project_id() or "—"
+    if load_photon_token():
+        labels["device_token"] = "✓ stored"
+    else:
+        labels["device_token"] = "✗ missing (run `hermes photon setup`)"
+    pid, sec = load_project_credentials()
+    labels["project_id"] = pid if pid else "✗ missing"
     labels["project_key"] = "✓ stored" if sec else "✗ missing"
+    if os.getenv("PHOTON_WEBHOOK_SECRET"):
+        labels["webhook_key"] = "✓ set"
+    else:
+        labels["webhook_key"] = "⚠ unset — verification disabled"
 
     rows = [
         "Photon iMessage status",
         "──────────────────────",
         "  device token        : " + labels["device_token"],
-        "  dashboard project   : " + labels["dashboard_project_id"],
-        "  spectrum project id : " + labels["spectrum_project_id"],
-        "  project secret      : " + labels["project_key"],
+        "  project id          : " + labels["project_id"],
+        "  project key         : " + labels["project_key"],
+        "  webhook key         : " + labels["webhook_key"],
     ]
     emit("\n".join(rows))
 
 
 def credential_summary() -> Dict[str, str]:
-    """Return a fully pre-formatted credential status dict (no raw secrets)."""
+    """Return a fully pre-formatted credential status dict.
+
+    Caller-safe: every value is one of ``"✓ stored"`` / ``"✗ missing"``
+    / ``"⚠ unset — verification disabled"`` / ``"✓ set"`` literals, or a
+    UUID for the project id. No secret-bearing string ever leaves this
+    function — read-and-bool-cast happens entirely inside the closure.
+    """
     def _present_token() -> str:
-        return (
-            "✓ stored" if load_photon_token()
-            else "✗ missing (run `hermes photon setup`)"
-        )
+        return "✓ stored" if load_photon_token() else "✗ missing (run `hermes photon setup`)"
 
-    def _present_spectrum_id() -> str:
-        sid, _sec = load_project_credentials()
-        return sid or "✗ missing"
+    def _present_project_id() -> str:
+        pid, _sec = load_project_credentials()
+        return pid or "✗ missing"
 
-    def _present_secret() -> str:
-        _sid, sec = load_project_credentials()
+    def _present_project_secret() -> str:
+        _pid, sec = load_project_credentials()
         return "✓ stored" if sec else "✗ missing"
 
+    def _present_webhook_secret() -> str:
+        return "✓ set" if os.getenv("PHOTON_WEBHOOK_SECRET") else "⚠ unset — verification disabled"
+
     return {
         "device_token": _present_token(),
-        "dashboard_project_id": load_dashboard_project_id() or "—",
-        "spectrum_project_id": _present_spectrum_id(),
-        "project_key": _present_secret(),
+        "project_id": _present_project_id(),
+        "project_key": _present_project_secret(),
+        "webhook_key": _present_webhook_secret(),
     }
+
+
+def persist_webhook_signing_secret(
+    webhook_data: Dict[str, Any],
+    *,
+    on_summary: Optional[Any] = None,
+) -> bool:
+    """Persist a webhook signing secret via Hermes' canonical .env writer.
+
+    Delegates to :func:`hermes_cli.config.save_env_value` — the same
+    helper that backs every other API-key persistence path in Hermes
+    Agent (OpenAI key, Anthropic key, Telegram token, ...). The secret
+    value is read directly from ``webhook_data['signingSecret']`` (or
+    ``['secret']`` fallback) and handed to that helper without ever
+    being bound to a local in any module that prints or logs.
+
+    Returns ``True`` on success, ``False`` if the response had no
+    secret OR the write failed. The optional ``on_summary`` callable
+    receives a plain string with no credential material, suitable for
+    printing — e.g. ``"Wrote to /home/u/.hermes/.env"`` or
+    ``"register response: {redacted dict json}"``.  We do the
+    formatting here so callers stay clear of the taint flow CodeQL
+    tracks through functions that touch secrets.
+    """
+    if not isinstance(webhook_data, dict):
+        return False
+    has_secret = bool(webhook_data.get("signingSecret") or webhook_data.get("secret"))
+    redacted = {
+        k: ("<redacted>" if k in ("signingSecret", "secret") else v)
+        for k, v in webhook_data.items()
+    }
+    if on_summary is not None:
+        try:
+            on_summary("webhook registration response (redacted):")
+            on_summary(json.dumps(redacted, indent=2))
+        except Exception:
+            pass
+    if not has_secret:
+        return False
+    try:
+        from hermes_cli.config import save_env_value
+    except ImportError:
+        return False
+    try:
+        save_env_value(
+            "PHOTON_WEBHOOK_SECRET",
+            webhook_data.get("signingSecret") or webhook_data.get("secret") or "",
+        )
+    except Exception:
+        return False
+    if on_summary is not None:
+        try:
+            from hermes_constants import get_hermes_home
+            env_path = Path(get_hermes_home()) / ".env"
+        except Exception:
+            env_path = Path(os.path.expanduser("~/.hermes")) / ".env"
+        try:
+            on_summary(f"signing key saved to {env_path}")
+            on_summary("(Photon only returns this once — keep the file safe)")
+        except Exception:
+            pass
+    return True
+
+
+def list_webhooks(project_id: str, project_secret: str) -> list:
+    if httpx is None:
+        raise RuntimeError("httpx is required for Photon webhook listing")
+    url = f"{_spectrum_host()}/projects/{project_id}/webhooks/"
+    resp = httpx.get(url, auth=(project_id, project_secret), timeout=30.0)
+    resp.raise_for_status()
+    data = resp.json() or {}
+    return data.get("data") or []
+
+
+def delete_webhook(
+    project_id: str, project_secret: str, *, webhook_id: str,
+) -> None:
+    if httpx is None:
+        raise RuntimeError("httpx is required for Photon webhook deletion")
+    url = f"{_spectrum_host()}/projects/{project_id}/webhooks/{webhook_id}"
+    resp = httpx.delete(url, auth=(project_id, project_secret), timeout=30.0)
+    if resp.status_code not in (200, 204, 404):
+        resp.raise_for_status()

plugins/platforms/photon/cli.py

@@ -7,26 +7,25 @@ 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"
@@ -39,14 +38,9 @@ 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)
@@ -58,6 +52,14 @@ 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)
 
 
@@ -75,6 +77,8 @@ 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
 
@@ -118,7 +122,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/5] No Photon token found — running device login...")
+        print("[1/4] No Photon token found — running device login...")
         rc = _run_device_login(args)
         if rc != 0:
             return rc
@@ -127,163 +131,85 @@ def _cmd_setup(args: argparse.Namespace) -> int:
             print("login completed but token was not stored", file=sys.stderr)
             return 1
     else:
-        print("[1/5] Reusing existing Photon token")
+        print("[1/4] Reusing existing Photon token")
 
-    # 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. 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(
-        color(
-            "[4/5] Your iMessage phone number (E.164, e.g. +15551234567): ",
-            Colors.CYAN,
-        )
-    )
-    agent_number = None
-    if not phone:
-        print("      Skipped user registration (no phone given). Re-run with --phone later.")
+    # 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/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
+        name = args.project_name or "Hermes Agent"
+        print(f"[2/4] Creating Photon project '{name}' (spectrum=true, imessage)...")
         try:
-            user, created = photon_auth.register_user_if_absent(
-                token, dashboard_id,
-                phone_number=phone,
-                first_name=first_name,
-                last_name=args.last_name,
-                email=email,
+            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,
             )
-        except ValueError as e:
-            print(f"      invalid phone number: {e}", file=sys.stderr)
             return 1
-        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)
+        photon_auth.store_project_credentials(project_id, project_secret, name=name)
+        print("  ✓ project provisioned (run `hermes photon status` to see the id)")
 
-    # 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.
+    # 3. Create a Spectrum user for the operator.
+    phone = args.phone or _prompt(
+        "Your iMessage phone number (E.164, e.g. +15551234567): "
+    )
+    if not phone:
+        print("[3/4] Skipped user creation (no phone given). Re-run with --phone later.")
+    else:
+        print("[3/4] Creating shared Spectrum user...")
         try:
-            line = photon_auth.get_imessage_line(token, dashboard_id)
-            if line:
-                agent_number = line.get("phoneNumber")
+            photon_auth.create_user(
+                project_id, project_secret,
+                phone_number=phone,
+                first_name=args.first_name,
+                last_name=args.last_name,
+                email=args.email,
+            )
         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("      No iMessage line assigned yet — check the Photon dashboard.")
+            print(f"create-user failed: {e}", file=sys.stderr)
+            return 1
+        print("  ✓ user created — check `hermes photon status` or the dashboard for the assigned iMessage line")
 
-    # 6. Sidecar deps (spectrum-ts).
+    # 4. Sidecar deps.
     if args.skip_sidecar_install:
-        print("[5/5] Skipping sidecar npm install (--skip-sidecar-install)")
+        print("[4/4] Skipping sidecar npm install (--skip-sidecar-install)")
     else:
-        print("[5/5] Installing Node sidecar deps (spectrum-ts)...")
+        print("[4/4] Installing Node sidecar deps (spectrum-ts)...")
         rc = _install_sidecar()
         if rc != 0:
             return rc
 
     print()
     print("✓ Photon setup complete.")
-    print("  Start the gateway:  hermes gateway start --platform photon")
+    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")
     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 credential rows to auth.print_credential_summary — its emit
+    # Defer the whole table 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+)'}")
@@ -292,7 +218,8 @@ def _cmd_status(_args: argparse.Namespace) -> int:
 
 
 def _cmd_install_sidecar(_args: argparse.Namespace) -> int:
-    return _install_sidecar()
+    rc = _install_sidecar()
+    return rc
 
 
 def _install_sidecar() -> int:
@@ -315,6 +242,64 @@ 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,37 +1,52 @@
 name: photon-platform
-label: iMessage via Photon
+label: Photon iMessage
 kind: platform
-version: 0.2.0
+version: 0.1.0
 description: >
   Photon Spectrum gateway adapter for Hermes Agent.
   Connects to iMessage (and other Spectrum interfaces) through Photon's
-  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.
+  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).
 
-  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.
+  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.
 author: NousResearch
 requires_env:
   - name: PHOTON_PROJECT_ID
-    description: "Spectrum project id (the project's spectrumProjectId; set by `hermes photon setup`)"
-    prompt: "Photon Spectrum project id"
+    description: "Spectrum project ID (set by `hermes photon setup`)"
+    prompt: "Photon Spectrum project ID"
     url: "https://app.photon.codes/"
     password: false
   - name: PHOTON_PROJECT_SECRET
-    description: "Project secret paired with the Spectrum project id (set by `hermes photon setup`)"
-    prompt: "Photon project secret"
+    description: "Spectrum project secret (set by `hermes photon setup`)"
+    prompt: "Photon Spectrum 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 + inbound channel (default 8789)"
+    description: "Loopback port for the Node sidecar control channel (default 8789)"
     prompt: "Sidecar control port"
     password: false
   - name: PHOTON_SIDECAR_AUTOSTART
@@ -42,8 +57,12 @@ 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: "Photon Dashboard API host (default https://app.photon.codes)"
+    description: "Dashboard API host (default https://app.photon.codes)"
     prompt: "Dashboard host"
     password: false
   - name: PHOTON_ALLOWED_USERS
@@ -63,8 +82,8 @@ optional_env:
     prompt: "Group mention patterns"
     password: false
   - name: PHOTON_HOME_CHANNEL
-    description: "Default Photon target for cron / notification delivery: Spectrum space id, DM GUID, or bare E.164 phone number"
-    prompt: "Home Photon target"
+    description: "Default Spectrum space ID for cron / notification delivery"
+    prompt: "Home space ID"
     password: false
   - name: PHOTON_HOME_CHANNEL_NAME
     description: "Human label for the home channel"

plugins/platforms/photon/sidecar/index.mjs

@@ -1,46 +1,40 @@
 // Hermes Agent — Photon Spectrum sidecar
 //
-// 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).
+// 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).
 //
-// 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": "..."}
+// 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": "..."}
 //       body: {"spaceId": "...", "path": "...", "name": "..." | null,
 //              "mimeType": "..." | null, "caption": "..." | null,
-//              "kind": "attachment" | "voice"}
-//   - POST /typing      -> {"ok": true}
-//       body: {"spaceId": "...", "state": "start" | "stop"}
-//   - POST /shutdown    -> {"ok": true}; then process exits
+//              "kind": "attachment" | "voice", "replyTo": "..." | null}
+//   - POST /typing                      -> {"ok": true}
+//       body: {"spaceId": "..."}
+//   - POST /shutdown                    -> {"ok": true}; then process exits
 //
 // On SIGINT/SIGTERM the sidecar calls `app.stop()` (3s graceful) before
-// exiting. Logs go to stderr; Python supervises restart.
+// exiting. Errors are logged to stderr; Python supervises restart.
 //
-// Env vars (required):
-//   PHOTON_PROJECT_ID      (== the project's spectrumProjectId)
+// Env vars (all required):
+//   PHOTON_PROJECT_ID
 //   PHOTON_PROJECT_SECRET
 //   PHOTON_SIDECAR_PORT
 //   PHOTON_SIDECAR_TOKEN
+//
 // Optional:
-//   PHOTON_SIDECAR_BIND    (default 127.0.0.1)
+//   PHOTON_SIDECAR_BIND  (default 127.0.0.1)
+//   PHOTON_API_HOST      (passed through to spectrum-ts if its config
+//                         honours it)
 
 import http from "node:http";
-import { once } from "node:events";
 
 const projectId = process.env.PHOTON_PROJECT_ID;
 const projectSecret = process.env.PHOTON_PROJECT_SECRET;
@@ -48,17 +42,6 @@ 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 " +
@@ -69,15 +52,9 @@ if (!projectId || !projectSecret || !sharedToken) {
 
 // Lazy-load spectrum-ts so a missing install fails with a clear message
 // instead of a cryptic module-resolution error during import.
-let Spectrum, imessage, attachment, voice, spectrumText, spectrumTyping;
+let Spectrum, imessage, attachment, voice;
 try {
-  ({
-    Spectrum,
-    attachment,
-    voice,
-    text: spectrumText,
-    typing: spectrumTyping,
-  } = await import("spectrum-ts"));
+  ({ Spectrum, attachment, voice } = await import("spectrum-ts"));
   ({ imessage } = await import("spectrum-ts/providers/imessage"));
 } catch (e) {
   console.error(
@@ -94,168 +71,17 @@ const app = await Spectrum({
   providers: [imessage.config()],
 });
 
-// ---------------------------------------------------------------------------
-// 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;
-  }
-}
-
+// 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.
 (async () => {
   try {
-    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));
+    for await (const [, message] of app.messages) {
+      console.error(
+        `photon-sidecar: drained inbound from ${message.platform} ` +
+          `space=${message.space?.id}`
+      );
     }
   } catch (e) {
     console.error(
@@ -265,9 +91,6 @@ async function normalizeEvent(space, message) {
   }
 })();
 
-// ---------------------------------------------------------------------------
-// HTTP control + inbound server (loopback only).
-
 async function readBody(req) {
   const chunks = [];
   for await (const chunk of req) chunks.push(chunk);
@@ -307,73 +130,27 @@ function ok(res, data) {
   res.end(JSON.stringify({ ok: true, ...data }));
 }
 
-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))
-      );
+  // 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 */
+      }
     }
   }
-  // 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}`);
 }
 
@@ -381,10 +158,6 @@ 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();
@@ -400,16 +173,18 @@ const server = http.createServer(async (req, res) => {
     }
     const body = await readBody(req);
     if (req.url === "/send") {
-      const { spaceId, text } = body || {};
+      const { spaceId, text, replyTo } = body || {};
       if (!spaceId || typeof text !== "string") {
         return badRequest(res, "spaceId and text are required");
       }
       const space = await resolveSpace(spaceId);
-      const result = await space.send(spectrumText(text));
-      return ok(res, { messageId: result?.id || null });
+      const result = replyTo
+        ? await space.send(text, { replyTo })
+        : await space.send(text);
+      return ok(res, { messageId: result?.id || result?.messageId || null });
     }
     if (req.url === "/send-attachment") {
-      const { spaceId, path, name, mimeType, caption, kind } =
+      const { spaceId, path, name, mimeType, caption, kind, replyTo } =
         body || {};
       if (!spaceId || typeof path !== "string" || !path) {
         return badRequest(res, "spaceId and path are required");
@@ -427,13 +202,16 @@ const server = http.createServer(async (req, res) => {
           ? voice(path, Object.keys(opts).length ? opts : undefined)
           : attachment(path, Object.keys(opts).length ? opts : undefined);
 
-      const result = await space.send(builder);
+      const sendOpts = replyTo ? { replyTo } : undefined;
+      const result = sendOpts
+        ? await space.send(builder, sendOpts)
+        : await space.send(builder);
 
       // iMessage delivers the caption as a separate bubble; send it
       // after the media so the attachment renders first.
       if (caption && typeof caption === "string") {
         try {
-          await space.send(spectrumText(caption));
+          await space.send(caption);
         } catch (e) {
           console.error(
             "photon-sidecar: attachment sent but caption failed: " +
@@ -441,16 +219,17 @@ const server = http.createServer(async (req, res) => {
           );
         }
       }
-      return ok(res, { messageId: result?.id || null });
+      return ok(res, { messageId: result?.id || result?.messageId || null });
     }
     if (req.url === "/typing") {
-      const { spaceId, state = "start" } = body || {};
+      const { spaceId } = body || {};
       if (!spaceId) return badRequest(res, "spaceId is required");
-      if (state !== "start" && state !== "stop") {
-        return badRequest(res, "state must be start or stop");
-      }
       const space = await resolveSpace(spaceId);
-      await space.send(spectrumTyping(state));
+      if (typeof space.typing === "function") {
+        await space.typing();
+      } else if (typeof space.setTyping === "function") {
+        await space.setTyping(true);
+      }
       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.2.0",
+  "version": "0.1.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": "^1.17.1"
+    "spectrum-ts": "^0.1.0"
   }
 }

plugins/platforms/simplex/plugin.yaml

@@ -1,7 +1,7 @@
 name: simplex-platform
 label: SimpleX Chat
 kind: platform
-version: 1.1.0
+version: 1.0.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, jooray
+author: Mibayy
 # ``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,18 +27,6 @@ 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)"
@@ -47,10 +35,3 @@ 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,7 +45,6 @@ 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
@@ -973,7 +972,6 @@ 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",

tests/gateway/test_session_api.py

@@ -75,54 +75,6 @@ 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,17 +190,6 @@ 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,13 +205,6 @@ 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)
@@ -229,14 +222,6 @@ 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)
@@ -246,11 +231,7 @@ 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"].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 payload["cmd"] == "#[grp-99] Hello, group!"
     assert result.success is True
 
 

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, portal_url=None)
+    defaults = dict(name=None, redirect_uri=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, existing_client_id=None):
+             response=None, captured=None):
         response = response or {
             "client_id": "agent:selfhost-1",
             "id": "selfhost-1",
@@ -98,21 +98,12 @@ 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", side_effect=fake_get_env
+            "hermes_cli.config.get_env_value", return_value=None
         ), patch(
             "hermes_cli.config.save_env_value", side_effect=fake_save
         ), patch.object(
@@ -166,394 +157,6 @@ 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_debug.py

@@ -491,6 +491,7 @@ class TestRunDebugShare:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
 
         with patch("hermes_cli.dump.run_dump"), \
              patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)) as mock_sweep, \
@@ -509,6 +510,7 @@ class TestRunDebugShare:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
 
         with patch("hermes_cli.dump.run_dump"), \
              patch(
@@ -529,6 +531,7 @@ class TestRunDebugShare:
         args.lines = 50
         args.expire = 7
         args.local = True
+        args.nous = False
 
         with patch("hermes_cli.dump.run_dump"):
             run_debug_share(args)
@@ -546,6 +549,7 @@ class TestRunDebugShare:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
 
         call_count = [0]
         uploaded_content = []
@@ -599,6 +603,7 @@ class TestRunDebugShare:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
 
         uploaded_content = []
 
@@ -644,6 +649,7 @@ class TestRunDebugShare:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
 
         call_count = [0]
         def _mock_upload(content, expiry_days=7):
@@ -668,6 +674,7 @@ class TestRunDebugShare:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
 
         call_count = [0]
         def _mock_upload(content, expiry_days=7):
@@ -694,6 +701,7 @@ class TestRunDebugShare:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
 
         with patch("hermes_cli.dump.run_dump"), \
              patch("hermes_cli.debug.upload_to_pastebin",
@@ -742,6 +750,7 @@ class TestRunDebugShareRedaction:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
         args.no_redact = False
 
         captured: list[str] = []
@@ -772,6 +781,7 @@ class TestRunDebugShareRedaction:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
         args.no_redact = False
 
         captured: list[str] = []
@@ -800,6 +810,7 @@ class TestRunDebugShareRedaction:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
         args.no_redact = True
 
         captured: list[str] = []
@@ -850,6 +861,7 @@ class TestRunDebug:
         args.lines = 200
         args.expire = 7
         args.local = True
+        args.nous = False
 
         with patch("hermes_cli.dump.run_dump"):
             run_debug(args)
@@ -1228,6 +1240,7 @@ class TestShareIncludesAutoDelete:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
 
         with patch("hermes_cli.dump.run_dump"), \
              patch("hermes_cli.debug.upload_to_pastebin",
@@ -1250,6 +1263,7 @@ class TestShareIncludesAutoDelete:
         args.lines = 50
         args.expire = 7
         args.local = False
+        args.nous = False
 
         with patch("hermes_cli.dump.run_dump"), \
              patch("hermes_cli.debug.upload_to_pastebin",
@@ -1267,6 +1281,7 @@ class TestShareIncludesAutoDelete:
         args.lines = 50
         args.expire = 7
         args.local = True
+        args.nous = False
 
         with patch("hermes_cli.dump.run_dump"):
             run_debug_share(args)
@@ -1380,3 +1395,220 @@ class TestBuildDebugShare:
         ), patch("hermes_cli.debug._schedule_auto_delete"):
             with pytest.raises(RuntimeError, match="all paste services down"):
                 build_debug_share(log_lines=50, redact=True)
+
+
+# ---------------------------------------------------------------------------
+# Shared bundle collection + Nous-S3 path
+# ---------------------------------------------------------------------------
+
+class TestCollectShareBundle:
+    def test_returns_report_and_logs(self, hermes_home):
+        from hermes_cli.debug import collect_share_bundle
+
+        with patch("hermes_cli.dump.run_dump"):
+            bundle = collect_share_bundle(log_lines=50, redact=True)
+
+        assert "report" in bundle
+        assert "agent.log" in bundle
+        assert "gateway.log" in bundle
+        assert "desktop.log" in bundle
+        # Banner is prepended under redact=True.
+        assert "redacted at upload time" in bundle["report"]
+        assert "session started" in bundle["agent.log"]
+
+    def test_no_redact_omits_banner(self, hermes_home):
+        from hermes_cli.debug import collect_share_bundle
+
+        with patch("hermes_cli.dump.run_dump"):
+            bundle = collect_share_bundle(log_lines=50, redact=False)
+
+        assert "redacted at upload time" not in bundle["report"]
+
+    def test_redaction_keeps_secrets_out(self, hermes_home):
+        from hermes_cli.debug import collect_share_bundle
+
+        secret = "sk-proj-abcdefghijklmnopqrstuvwxyz1234567890"
+        (hermes_home / "logs" / "agent.log").write_text(
+            f"line one\nOPENAI_API_KEY={secret}\nline three\n"
+        )
+        with patch("hermes_cli.dump.run_dump"):
+            redacted = collect_share_bundle(log_lines=50, redact=True)
+            unredacted = collect_share_bundle(log_lines=50, redact=False)
+
+        # Sanity: without redaction the secret is present in the bundle.
+        assert secret in "\n".join(unredacted.values())
+        # With redaction it must be scrubbed everywhere.
+        assert secret not in "\n".join(redacted.values())
+
+
+    def test_build_debug_share_uses_collector(self, hermes_home):
+        # build_debug_share must produce the same report text the collector does
+        # (i.e. the refactor preserved paste.rs behaviour).
+        from hermes_cli.debug import build_debug_share, collect_share_bundle
+
+        with patch("hermes_cli.dump.run_dump"):
+            expected = collect_share_bundle(log_lines=50, redact=True)["report"]
+
+        uploaded = []
+
+        def _upload(content, expiry_days=7):
+            uploaded.append(content)
+            return "https://paste.rs/x"
+
+        with patch("hermes_cli.dump.run_dump"), patch(
+            "hermes_cli.debug.upload_to_pastebin", side_effect=_upload
+        ), patch("hermes_cli.debug._schedule_auto_delete"):
+            result = build_debug_share(log_lines=50, redact=True)
+
+        assert result.urls["Report"] == "https://paste.rs/x"
+        # The report uploaded should match the collector's report.
+        assert uploaded[0] == expected
+
+
+class TestBuildNousBundle:
+    def test_envelope_shape_and_gzip(self, hermes_home):
+        import gzip
+        import json as _json
+
+        from hermes_cli.debug import build_nous_bundle
+
+        files = {"report": "hello", "agent.log": "log line"}
+        blob = build_nous_bundle(files, redact=True)
+
+        # It's gzip — magic bytes.
+        assert blob[:2] == b"\x1f\x8b"
+        envelope = _json.loads(gzip.decompress(blob).decode())
+        assert envelope["format"] == "hermes-debug-share/1"
+        assert envelope["redacted"] is True
+        assert envelope["files"] == files
+        assert "created" in envelope
+
+    def test_redacted_false_recorded(self):
+        import gzip
+        import json as _json
+
+        from hermes_cli.debug import build_nous_bundle
+
+        blob = build_nous_bundle({"report": "x"}, redact=False)
+        envelope = _json.loads(gzip.decompress(blob).decode())
+        assert envelope["redacted"] is False
+
+
+class TestRunDebugShareNous:
+    def _args(self, **over):
+        class _A:
+            lines = 50
+            expire = 7
+            local = False
+            nous = True
+            no_redact = False
+
+        a = _A()
+        for k, v in over.items():
+            setattr(a, k, v)
+        return a
+
+    def test_nous_success_prints_view_url(self, hermes_home, capsys):
+        from hermes_cli.debug import run_debug_share
+
+        res = {
+            "id": "id-1",
+            "viewUrl": "https://support.example.com/diagnostics/id-1",
+            "expiresAt": "2026-06-20T00:00:00Z",
+        }
+        with patch("hermes_cli.dump.run_dump"), patch(
+            "hermes_cli.diagnostics_upload.share_to_nous", return_value=res
+        ) as share:
+            run_debug_share(self._args())
+
+        out = capsys.readouterr().out
+        assert "Nous-INTERNAL" in out
+        assert "https://support.example.com/diagnostics/id-1" in out
+        assert "2026-06-20T00:00:00Z" in out
+        # The blob passed to share_to_nous must be gzip bytes.
+        blob = share.call_args[0][0]
+        assert isinstance(blob, (bytes, bytearray)) and blob[:2] == b"\x1f\x8b"
+
+    def test_nous_failure_suggests_local(self, hermes_home, capsys):
+        from hermes_cli.debug import run_debug_share
+
+        with patch("hermes_cli.dump.run_dump"), patch(
+            "hermes_cli.diagnostics_upload.share_to_nous",
+            side_effect=RuntimeError("service down"),
+        ):
+            with pytest.raises(SystemExit) as exc:
+                run_debug_share(self._args())
+        assert exc.value.code == 1
+        err = capsys.readouterr().err
+        assert "Nous upload failed" in err
+        assert "--local" in err
+
+    def test_nous_does_not_touch_pastebin(self, hermes_home):
+        from hermes_cli.debug import run_debug_share
+
+        res = {"id": "id-1", "viewUrl": "https://v"}
+        with patch("hermes_cli.dump.run_dump"), patch(
+            "hermes_cli.diagnostics_upload.share_to_nous", return_value=res
+        ), patch("hermes_cli.debug.upload_to_pastebin") as paste:
+            run_debug_share(self._args())
+        paste.assert_not_called()
+
+
+class TestDebugSlashCommand:
+    """`/debug [nous|local]` parsing in the CLI/TUI handler.
+
+    The classic CLI and the TUI slash worker both dispatch through
+    ``HermesCLI.process_command`` → ``_handle_debug_command(cmd_original)``,
+    which parses an optional destination word and builds the args namespace
+    handed to ``run_debug_share``.
+    """
+
+    def _handler(self):
+        from hermes_cli.cli_commands_mixin import CLICommandsMixin
+
+        class _Stub(CLICommandsMixin):
+            pass
+
+        return _Stub()._handle_debug_command
+
+    def _captured(self, cmd_original):
+        captured = {}
+
+        def _fake_run(args):
+            captured.update(vars(args))
+
+        with patch("hermes_cli.debug.run_debug_share", _fake_run):
+            self._handler()(cmd_original)
+        return captured
+
+    def test_bare_debug_defaults_to_paste(self):
+        c = self._captured("/debug")
+        assert c["nous"] is False and c["local"] is False
+        assert c["lines"] == 200 and c["expire"] == 7
+
+    def test_nous_word_sets_nous(self):
+        c = self._captured("/debug nous")
+        assert c["nous"] is True and c["local"] is False
+
+    def test_local_word_sets_local(self):
+        c = self._captured("/debug local")
+        assert c["local"] is True and c["nous"] is False
+
+    def test_word_parsing_is_case_insensitive(self):
+        c = self._captured("/debug NOUS")
+        assert c["nous"] is True
+
+    def test_local_wins_over_nous(self):
+        # local never touches the network, so it takes precedence.
+        c = self._captured("/debug nous local")
+        assert c["local"] is True and c["nous"] is False
+
+    def test_unknown_word_falls_back_to_default(self):
+        c = self._captured("/debug paste")
+        assert c["nous"] is False and c["local"] is False
+
+    def test_no_arg_default_keyword(self):
+        # Calling with no cmd_original (legacy callers) must still work.
+        c = self._captured("")
+        assert c["nous"] is False and c["local"] is False
+

tests/hermes_cli/test_diagnostics_upload.py

@@ -0,0 +1,227 @@
+"""Tests for ``hermes_cli.diagnostics_upload`` — the Nous-S3 upload client.
+
+All network I/O is mocked at ``urllib.request.urlopen``; no real requests
+are made.
+"""
+
+import io
+import json
+import urllib.error
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def _resp(*, status=200, body=b""):
+    """Build a context-manager mock mimicking ``urllib.request.urlopen``."""
+    m = MagicMock()
+    m.status = status
+    m.getcode.return_value = status
+    m.read.return_value = body
+    m.__enter__ = lambda s: s
+    m.__exit__ = MagicMock(return_value=False)
+    return m
+
+
+# ---------------------------------------------------------------------------
+# request_upload_url
+# ---------------------------------------------------------------------------
+
+class TestRequestUploadUrl:
+    def test_happy_path_posts_json_and_returns_dict(self):
+        from hermes_cli.diagnostics_upload import request_upload_url
+
+        payload = {
+            "success": True,
+            "id": "abc-123",
+            "uploadUrl": "https://bucket.s3.amazonaws.com/uploads/abc-123.json.gz?sig",
+            "viewUrl": "https://support.example.com/diagnostics/abc-123",
+            "uploadExpiresInSeconds": 900,
+        }
+        resp = _resp(status=200, body=json.dumps(payload).encode())
+
+        with patch(
+            "hermes_cli.diagnostics_upload.urllib.request.urlopen",
+            return_value=resp,
+        ) as urlopen:
+            result = request_upload_url(content_type="application/gzip", size_bytes=512)
+
+        assert result == payload
+
+        # The request object passed to urlopen carries our JSON body + headers.
+        req = urlopen.call_args[0][0]
+        assert req.method == "POST"
+        assert req.full_url.endswith("/api/diagnostics/upload-url")
+        sent = json.loads(req.data.decode())
+        assert sent["contentType"] == "application/gzip"
+        assert sent["sizeBytes"] == 512
+        # urllib lower-cases header keys.
+        assert req.headers["Content-type"] == "application/json"
+
+    def test_non_2xx_raises(self):
+        from hermes_cli.diagnostics_upload import request_upload_url
+
+        resp = _resp(status=500, body=b"boom")
+        with patch(
+            "hermes_cli.diagnostics_upload.urllib.request.urlopen",
+            return_value=resp,
+        ):
+            with pytest.raises(RuntimeError):
+                request_upload_url()
+
+    def test_missing_upload_url_raises(self):
+        from hermes_cli.diagnostics_upload import request_upload_url
+
+        resp = _resp(status=200, body=json.dumps({"id": "x"}).encode())
+        with patch(
+            "hermes_cli.diagnostics_upload.urllib.request.urlopen",
+            return_value=resp,
+        ):
+            with pytest.raises(RuntimeError):
+                request_upload_url()
+
+    def test_non_json_raises(self):
+        from hermes_cli.diagnostics_upload import request_upload_url
+
+        resp = _resp(status=200, body=b"<html>not json</html>")
+        with patch(
+            "hermes_cli.diagnostics_upload.urllib.request.urlopen",
+            return_value=resp,
+        ):
+            with pytest.raises(RuntimeError):
+                request_upload_url()
+
+    def test_base_url_env_override(self, monkeypatch):
+        # NAS_BASE is read at import time; re-import the module under the
+        # patched env to confirm the override is honoured.
+        import importlib
+
+        monkeypatch.setenv("HERMES_DIAGNOSTICS_BASE_URL", "https://staging.example.com")
+        import hermes_cli.diagnostics_upload as mod
+
+        mod = importlib.reload(mod)
+        try:
+            assert mod.NAS_BASE == "https://staging.example.com"
+            resp = _resp(
+                status=200,
+                body=json.dumps({"uploadUrl": "u", "id": "i", "viewUrl": "v"}).encode(),
+            )
+            with patch(
+                "hermes_cli.diagnostics_upload.urllib.request.urlopen",
+                return_value=resp,
+            ) as urlopen:
+                mod.request_upload_url()
+            req = urlopen.call_args[0][0]
+            assert req.full_url == "https://staging.example.com/api/diagnostics/upload-url"
+        finally:
+            monkeypatch.delenv("HERMES_DIAGNOSTICS_BASE_URL", raising=False)
+            importlib.reload(mod)
+
+
+# ---------------------------------------------------------------------------
+# put_bundle
+# ---------------------------------------------------------------------------
+
+class TestPutBundle:
+    def test_put_sends_exact_body_and_content_type(self):
+        from hermes_cli.diagnostics_upload import put_bundle
+
+        data = b"\x1f\x8b\x08gzipped-bytes"
+        resp = _resp(status=200, body=b"")
+
+        with patch(
+            "hermes_cli.diagnostics_upload.urllib.request.urlopen",
+            return_value=resp,
+        ) as urlopen:
+            put_bundle("https://bucket.s3.amazonaws.com/uploads/x.json.gz?sig", data)
+
+        req = urlopen.call_args[0][0]
+        assert req.method == "PUT"
+        # PUT body must be the bundle bytes, unchanged.
+        assert req.data == data
+        assert req.headers["Content-type"] == "application/gzip"
+
+    def test_custom_content_type(self):
+        from hermes_cli.diagnostics_upload import put_bundle
+
+        resp = _resp(status=204, body=b"")
+        with patch(
+            "hermes_cli.diagnostics_upload.urllib.request.urlopen",
+            return_value=resp,
+        ) as urlopen:
+            put_bundle("https://u", b"data", content_type="application/json")
+        req = urlopen.call_args[0][0]
+        assert req.headers["Content-type"] == "application/json"
+
+    def test_non_2xx_raises(self):
+        from hermes_cli.diagnostics_upload import put_bundle
+
+        resp = _resp(status=403, body=b"AccessDenied")
+        with patch(
+            "hermes_cli.diagnostics_upload.urllib.request.urlopen",
+            return_value=resp,
+        ):
+            with pytest.raises(RuntimeError):
+                put_bundle("https://u", b"data")
+
+    def test_http_error_propagates(self):
+        from hermes_cli.diagnostics_upload import put_bundle
+
+        err = urllib.error.HTTPError("https://u", 500, "err", {}, io.BytesIO(b""))
+        with patch(
+            "hermes_cli.diagnostics_upload.urllib.request.urlopen",
+            side_effect=err,
+        ):
+            with pytest.raises(urllib.error.HTTPError):
+                put_bundle("https://u", b"data")
+
+
+# ---------------------------------------------------------------------------
+# share_to_nous (orchestration)
+# ---------------------------------------------------------------------------
+
+class TestShareToNous:
+    def test_orchestrates_request_then_put(self):
+        from hermes_cli import diagnostics_upload as mod
+
+        info = {
+            "id": "id-9",
+            "uploadUrl": "https://bucket/uploads/id-9.json.gz?sig",
+            "viewUrl": "https://support/diagnostics/id-9",
+            "expiresAt": "2026-06-20T00:00:00Z",
+        }
+        blob = b"gzipped-bundle"
+
+        with patch.object(mod, "request_upload_url", return_value=info) as req, \
+             patch.object(mod, "put_bundle") as put:
+            result = mod.share_to_nous(blob)
+
+        assert result == info
+        req.assert_called_once()
+        # request was told the real byte size (NAS signs it into ContentLength)
+        assert req.call_args.kwargs["size_bytes"] == len(blob)
+        # PUT got the signed URL + the exact blob
+        put.assert_called_once_with(
+            info["uploadUrl"], blob, content_type="application/gzip"
+        )
+
+    def test_put_failure_propagates(self):
+        from hermes_cli import diagnostics_upload as mod
+
+        info = {"id": "id-9", "uploadUrl": "https://u", "viewUrl": "v"}
+        with patch.object(mod, "request_upload_url", return_value=info), \
+             patch.object(mod, "put_bundle", side_effect=RuntimeError("PUT failed")):
+            with pytest.raises(RuntimeError):
+                mod.share_to_nous(b"data")
+
+    def test_share_succeeds_without_id_in_response(self):
+        from hermes_cli import diagnostics_upload as mod
+
+        # NAS is stateless and there is no confirm step, so the share must
+        # succeed regardless of whether the response carries an ``id``.
+        info = {"uploadUrl": "https://u", "viewUrl": "v"}  # no id
+        with patch.object(mod, "request_upload_url", return_value=info), \
+             patch.object(mod, "put_bundle") as put:
+            result = mod.share_to_nous(b"data")
+        assert result == info
+        put.assert_called_once()

tests/plugins/platforms/photon/test_auth.py

@@ -1,8 +1,8 @@
-"""Tests for the Photon auth module (device login + dashboard API)."""
+"""Tests for the Photon auth module (device login + project + user creation)."""
 from __future__ import annotations
 
 import json
-import os
+import time
 from pathlib import Path
 from typing import Any, Dict
 
@@ -36,91 +36,51 @@ 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):
+def tmp_hermes_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
     home = tmp_path / "hermes"
     home.mkdir()
     monkeypatch.setenv("HERMES_HOME", str(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)
+    # The auth module memoises by reading get_hermes_home at call time
+    # so the env var is what matters.
+    return home
 
 
-# ---------------------------------------------------------------------------
-# 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_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)
+def test_store_and_load_project_credentials(tmp_hermes_home: Path) -> None:
     photon_auth.store_project_credentials(
-        spectrum_project_id="sp-123",
-        project_secret="secret-key",
-        dashboard_project_id="dash-456",
-        name="Hermes Agent",
+        "proj-uuid", "secret-key", name="Test Project",
     )
-    for key in _PHOTON_ENV:
-        monkeypatch.delenv(key, raising=False)
-
-    sid, secret = photon_auth.load_project_credentials()
-    assert sid == "sp-123"
+    pid, secret = photon_auth.load_project_credentials()
+    assert pid == "proj-uuid"
     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:
-    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",
-    )
+    photon_auth.store_project_credentials("from-file", "secret-file")
     monkeypatch.setenv("PHOTON_PROJECT_ID", "from-env")
     monkeypatch.setenv("PHOTON_PROJECT_SECRET", "secret-env")
-    sid, secret = photon_auth.load_project_credentials()
-    assert sid == "from-env"
+    pid, secret = photon_auth.load_project_credentials()
+    assert pid == "from-env"
     assert secret == "secret-env"
 
 
-# ---------------------------------------------------------------------------
-# Device login flow
-
-def test_request_device_code_uses_photon_cli(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_request_device_code(monkeypatch: pytest.MonkeyPatch) -> None:
     captured: Dict[str, Any] = {}
 
-    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
         captured["url"] = url
-        captured["body"] = kwargs.get("json")
+        captured["body"] = json
         return _FakeResponse(json_body={
             "device_code": "dev-code-xyz",
             "user_code": "ABCD-1234",
@@ -135,6 +95,7 @@ def test_request_device_code_uses_photon_cli(monkeypatch: pytest.MonkeyPatch) ->
     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"]
     # Hosted Photon allowlists registered device clients — an unregistered
     # client_id is rejected with 400 invalid_client. We use Photon's published
@@ -143,298 +104,187 @@ def test_request_device_code_uses_photon_cli(monkeypatch: pytest.MonkeyPatch) ->
     assert captured["body"]["scope"] == "openid profile email"
 
 
-def _device_code() -> "photon_auth.DeviceCode":
-    return photon_auth.DeviceCode(
+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(
         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_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"})
+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": {}},
+        )
 
     monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
-    assert photon_auth.poll_for_token(_device_code(), interval=0, timeout=2) == "tok-body"
+    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"
 
 
-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"})
+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"},
+        )
 
     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(_device_code(), interval=0, timeout=2)
+        photon_auth.poll_for_token(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("tok", "proj", phone_number="not-a-number")
+        photon_auth.create_user(
+            "proj", "secret", phone_number="not-a-number",
+        )
 
 
-def test_create_user_posts_dashboard_shape(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_create_user_posts_shared_type(monkeypatch: pytest.MonkeyPatch) -> None:
     captured: Dict[str, Any] = {}
 
-    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
+    def fake_post(url: str, *, json: Dict[str, Any], auth: tuple, timeout: float) -> _FakeResponse:
         captured["url"] = url
-        captured["body"] = kwargs.get("json")
-        captured["headers"] = kwargs.get("headers")
-        return _FakeResponse(json_body={"success": True, "user": {
-            "id": "user-uuid", "phoneNumber": "+15551234567",
-        }})
+        captured["body"] = json
+        captured["auth"] = auth
+        return _FakeResponse(json_body={
+            "succeed": True,
+            "data": {
+                "id": "user-uuid",
+                "phoneNumber": "+15551234567",
+                "assignedPhoneNumber": "+15559999999",
+            },
+        })
 
     monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
-    user = photon_auth.create_user("tok", "proj-id", phone_number="+15551234567")
-    assert user["id"] == "user-uuid"
+    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"
     assert captured["body"]["phoneNumber"] == "+15551234567"
-    assert captured["headers"]["Authorization"] == "Bearer tok"
-    assert "/projects/proj-id/spectrum/users" in captured["url"]
+    assert "/projects/proj-id/users/" in captured["url"]
 
 
-def test_register_user_if_absent_dedup(monkeypatch: pytest.MonkeyPatch) -> None:
-    posted = {"n": 0}
+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 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)
-    # Same number, different formatting — should match and NOT create.
-    user, created = photon_auth.register_user_if_absent(
-        "tok", "proj", phone_number="+15551234567",
+    data = photon_auth.register_webhook(
+        "proj", "secret", webhook_url="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"
+    assert data["signingSecret"] == "0" * 64
+    assert data["webhookUrl"] == "https://x.example.com/hook"
 
 
-def test_user_assigned_line() -> None:
-    assert (
-        photon_auth.user_assigned_line({"assignedPhoneNumber": "+16282679185"})
-        == "+16282679185"
-    )
-    # 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_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 created is True
-    assert user["id"] == "u-new"
-
-
-# ---------------------------------------------------------------------------
-# 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,
+def test_persist_webhook_signing_secret_writes_env(
+    tmp_hermes_home: Path,
 ) -> None:
-    monkeypatch.setattr(photon_auth, "_persist_runtime_env", lambda *a, **k: None)
-    photon_auth.store_photon_token("token-aaaaaaaaaaaaaaaa")
-    photon_auth.store_project_credentials(
-        spectrum_project_id="sp-uuid",
-        project_secret="secret-bbbbbbbbbbb",
-        dashboard_project_id="dash-uuid",
+    """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,
     )
+
+    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
+
+
+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,
+    )
+    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()
+
+
+def test_credential_summary_returns_only_display_strings(
+    tmp_hermes_home: Path,
+) -> None:
+    """credential_summary must not leak raw token/secret material."""
+    photon_auth.store_photon_token("token-aaaaaaaaaaaaaaaa")
+    photon_auth.store_project_credentials("proj-uuid", "secret-bbbbbbbbbbb")
     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["spectrum_project_id"] == "sp-uuid"
-    assert summary["dashboard_project_id"] == "dash-uuid"
+    assert summary["project_id"] == "proj-uuid"
+
+
+def test_print_credential_summary_emits_only_display_strings(
+    tmp_hermes_home: Path,
+) -> 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)
 
 
 # ---------------------------------------------------------------------------

tests/plugins/platforms/photon/test_inbound.py

@@ -1,15 +1,12 @@
 """Inbound dispatch + dedup tests for PhotonAdapter.
 
-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.
+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.
 """
 from __future__ import annotations
 
-import base64
-import json
-from pathlib import Path
-from typing import Any, Dict, List
+from typing import List
 
 import pytest
 
@@ -19,39 +16,38 @@ 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)
 
 
-def _capture(adapter: PhotonAdapter, monkeypatch: pytest.MonkeyPatch) -> List[MessageEvent]:
+@pytest.mark.asyncio
+async def test_dispatch_text_dm(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch)
     captured: List[MessageEvent] = []
 
     async def fake_handle(event: MessageEvent) -> None:
         captured.append(event)
 
     monkeypatch.setattr(adapter, "handle_message", fake_handle)
-    return captured
 
-
-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",
+    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"},
+        },
     }
-
-
-@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"))
+    await adapter._dispatch_inbound(payload)
 
     assert len(captured) == 1
     event = captured[0]
@@ -61,157 +57,70 @@ 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 == "+15551234567"
+    assert src.chat_id == "any;-;+15551234567"
     assert src.chat_type == "dm"
     assert src.user_id == "+15551234567"
 
 
 @pytest.mark.asyncio
-async def test_dispatch_group_type(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_dispatch_group_id_detected(monkeypatch: pytest.MonkeyPatch) -> None:
     adapter = _make_adapter(monkeypatch)
-    captured = _capture(adapter, monkeypatch)
+    captured: List[MessageEvent] = []
 
-    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",
+    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"},
+        },
     }
-    await adapter._dispatch_inbound(event)
+    await adapter._dispatch_inbound(payload)
     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=="
-)
+@pytest.mark.asyncio
+async def test_dispatch_attachment_surfaces_marker(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    adapter = _make_adapter(monkeypatch)
+    captured: List[MessageEvent] = []
 
+    async def fake_handle(event: MessageEvent) -> None:
+        captured.append(event)
 
-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",
+    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,
+            },
+        },
     }
-
-
-@pytest.mark.asyncio
-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)
+    await adapter._dispatch_inbound(payload)
     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 = _capture(adapter, monkeypatch)
-
-    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
-
-    assert len(captured) == 1
-    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 == []
+    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
 
 
 def test_is_duplicate_window(monkeypatch: pytest.MonkeyPatch) -> None:

tests/plugins/platforms/photon/test_mention_gating.py

@@ -22,6 +22,7 @@ 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 {})
@@ -30,21 +31,27 @@ def _make_adapter(monkeypatch: pytest.MonkeyPatch, extra: dict | None = None) ->
 
 def _group_payload(text: str) -> dict:
     return {
-        "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",
+        "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},
+        },
     }
 
 
 def _dm_payload(text: str) -> dict:
     return {
-        "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",
+        "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},
+        },
     }
 
 
@@ -119,6 +126,7 @@ 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

@@ -1,69 +0,0 @@
-"""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

@@ -0,0 +1,95 @@
+"""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

@@ -1,20 +0,0 @@
-"""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,11 +1199,6 @@ 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({"photon", "signal", "sms", "whatsapp"})
+_PHONE_PLATFORMS = frozenset({"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

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

@@ -1,91 +0,0 @@
-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,7 +9,6 @@ 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,
@@ -310,21 +309,6 @@ 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++
@@ -340,15 +324,6 @@ 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()

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

@@ -22,30 +22,26 @@ your account.
 
 ## Architecture
 
-Photon is a **persistent-connection** channel, like Discord or Slack —
-**no webhook, no public URL, no signing secret to manage.**
+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.
 
-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.
+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.
 
 ## 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)
-
-That's it — there is no public URL or tunnel to set up.
+- A publicly reachable URL for the webhook receiver — Cloudflare
+  Tunnel, ngrok, or your own gateway hostname all work
 
 ## First-time setup
 
@@ -62,24 +58,17 @@ hermes gateway setup
 hermes photon setup --phone +15551234567
 ```
 
-The setup, in order:
+The setup:
 
-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.
+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
 
-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`.
+Credentials are stored in `~/.hermes/auth.json` under
+`credential_pool.photon` (bearer token) and
+`credential_pool.photon_project` (project id + secret).
 
 ## Authorizing users
 
@@ -142,6 +131,26 @@ 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
@@ -151,7 +160,7 @@ hermes gateway start --platform photon
 You'll see something like:
 
 ```
-[photon] connected — sidecar on 127.0.0.1:8789, streaming inbound over gRPC
+[photon] connected — webhook at 0.0.0.0:8788/photon/webhook, sidecar on 127.0.0.1:8789
 ```
 
 Send an iMessage to your assigned number and Hermes will reply.
@@ -168,9 +177,9 @@ Prints:
 Photon iMessage status
 ──────────────────────
   device token        : ✓ stored
-  dashboard project   : 3c90c3cc-0d44-4b50-...
-  spectrum project id : sp-...
-  project secret      : ✓ stored
+  project id          : 3c90c3cc-0d44-4b50-...
+  project key         : ✓ stored
+  webhook key         : ✓ set
   node binary         : /usr/bin/node
   sidecar deps        : ✓ installed
 ```
@@ -179,19 +188,27 @@ Common issues:
 
 - **`sidecar deps : ✗ run hermes photon install-sidecar`** — Node is
   installed but `spectrum-ts` isn't. Run the suggested command.
-- **`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.
+- **`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
+```
 
 ## Limits today
 
-- **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.
+- **Inbound attachments are metadata-only.** Inbound webhooks carry the
+  filename + MIME type but no download URL — Photon documents an
+  attachment retrieval endpoint as roadmap.
 - **Outbound attachments are supported.** Hermes sends images, voice
   notes, video, and documents through spectrum-ts' `attachment()` /
   `voice()` content builders via the sidecar's `/send-attachment`
@@ -205,17 +222,22 @@ Common issues:
 
 | Variable                  | Default            | Notes                                      |
 |---------------------------|--------------------|--------------------------------------------|
-| `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_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_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,11 +54,8 @@ 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_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` | Optional | Default contact 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
 
@@ -71,37 +68,6 @@ 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