chore(deps): bump python-multipart from 0.0.20 to 0.0.31

Bumps [python-multipart](https://github.com/Kludex/python-multipart) from 0.0.20 to 0.0.31. - [Release notes](https://github.com/Kludex/python-multipart/releases) - [Changelog](https://github.com/Kludex/python-multipart/blob/main/CHANGELOG.md) - [Commits](https://github.com/Kludex/python-multipart/compare/0.0.20...0.0.31) --- updated-dependencies: - dependency-name: python-multipart dependency-version: 0.0.31 dependency-type: indirect ... Signed-off-by: dependabot[bot] <support@github.com>
fix(dashboard): stream file uploads via multipart instead of base64 JSON (NS-501) (#47663 )
2026-06-18 05:57:04 +00:00 · 2026-06-18 15:54:32 +10:00 · 2026-06-18 15:27:45 +10:00 · 2026-06-18 15:25:29 +10:00
29 changed files with 826 additions and 1505 deletions
@@ -15,7 +15,6 @@ from typing import Any, Dict, List, Optional, Set, Tuple

 from hermes_constants import OPENROUTER_BASE_URL
 from hermes_cli.config import load_env
-from agent.secret_scope import get_secret as _get_secret
 from agent.credential_persistence import (
    is_borrowed_credential_source,
    sanitize_borrowed_credential_payload,
@@ -1667,7 +1666,7 @@ def _seed_from_singletons(provider: str, entries: List[PooledCredential]) -> Tup
        _env_file = load_env()

        def _env_val(key: str) -> str:
-            return (_env_file.get(key) or _get_secret(key, "") or "").strip()
+            return (_env_file.get(key) or os.environ.get(key) or "").strip()

        anthropic_api_key = _env_val("ANTHROPIC_API_KEY")
        anthropic_oauth_env = (
@@ -1953,7 +1952,7 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool
    # changes to the .env file.
    def _get_env_prefer_dotenv(key: str) -> str:
        env_file = load_env()
-        val = env_file.get(key) or _get_secret(key, "") or ""
+        val = env_file.get(key) or os.environ.get(key) or ""
        return val.strip()

    # Honour user suppression — `hermes auth remove <provider> <N>` for an
@@ -1,205 +0,0 @@
-"""Profile-scoped credential resolution for multi-profile gateway multiplexing.
-
-The multiplexing gateway serves many profiles from one process. Each profile
-has its own ``.env`` with its own provider keys and platform tokens, so we
-**cannot** union them into the process-global ``os.environ`` (that would leak
-profile A's keys to profile B's turns, and to every subprocess spawned with
-``env=dict(os.environ)``).
-
-This module provides a fail-closed, context-local secret scope:
-
- ``set_secret_scope(mapping)`` installs the active profile's secrets for the
-  current task (a contextvar, so it propagates into the agent's worker thread
-  via ``copy_context()`` exactly like the HERMES_HOME override).
- ``get_secret(name)`` reads from that scope. When multiplexing is **active**
-  and no scope is set, it RAISES rather than silently falling back to
-  ``os.environ`` — an un-migrated or newly-added call site fails loud at that
-  exact line instead of leaking another profile's value. When multiplexing is
-  **off** (the default), it transparently reads ``os.environ`` so the
-  single-profile gateway and every non-gateway caller behave exactly as before.
-
-Design rationale lives in ``docs/design/multiplexing-gateway.md`` (Workstream A).
-"""
-from __future__ import annotations
-
-import os
-from contextvars import ContextVar, Token
-from pathlib import Path
-from typing import Dict, Mapping, Optional
-
-
-# ── multiplex-active flag ────────────────────────────────────────────────
-# Process-global: set once at gateway startup when gateway.multiplex_profiles
-# is true. Governs whether get_secret() fails closed on an unscoped read.
-# A plain module global (not a contextvar): it describes the deployment mode,
-# not a per-task value.
-_MULTIPLEX_ACTIVE: bool = False
-
-
-def set_multiplex_active(active: bool) -> None:
-    """Mark whether the process is running as a profile multiplexer.
-
-    Called once at gateway startup. When True, ``get_secret`` fails closed on
-    an unscoped read instead of falling back to ``os.environ``.
-    """
-    global _MULTIPLEX_ACTIVE
-    _MULTIPLEX_ACTIVE = bool(active)
-
-
-def is_multiplex_active() -> bool:
-    """Return whether the process is running as a profile multiplexer."""
-    return _MULTIPLEX_ACTIVE
-
-
-# ── the secret scope contextvar ──────────────────────────────────────────
-_SECRET_SCOPE: ContextVar[Optional[Mapping[str, str]]] = ContextVar(
-    "_SECRET_SCOPE", default=None
-)
-
-
-class UnscopedSecretError(RuntimeError):
-    """Raised when a secret is read in multiplex mode with no scope installed.
-
-    This is the fail-closed signal: it means a credential read reached
-    ``get_secret`` without a profile scope active, which in a multiplexer would
-    otherwise leak whichever profile's value happened to be in ``os.environ``.
-    The fix is to wrap the call path in ``set_secret_scope(...)`` (the per-turn
-    / per-adapter profile scope), not to widen the allowlist.
-    """
-
-
-def set_secret_scope(secrets: Optional[Mapping[str, str]]) -> Token:
-    """Install the active profile's secret mapping for the current context.
-
-    Returns a token for ``reset_secret_scope``. Pass ``None`` to clear.
-    """
-    return _SECRET_SCOPE.set(secrets)
-
-
-def reset_secret_scope(token: Token) -> None:
-    """Restore the previous secret scope."""
-    _SECRET_SCOPE.reset(token)
-
-
-def current_secret_scope() -> Optional[Mapping[str, str]]:
-    """Return the active secret mapping, or None when no scope is installed."""
-    return _SECRET_SCOPE.get()
-
-
-# ── genuinely-global env vars (NOT per-profile secrets) ──────────────────
-# These are process/deployment-level settings, not profile credentials. They
-# legitimately live in os.environ and must keep reading from it even in
-# multiplex mode — routing them through the fail-closed path would wrongly
-# crash. Anything matching is read from os.environ regardless of scope.
-#
-# Membership test is by exact name OR prefix (see _is_global_env). Keep this
-# list tight: when in doubt a value is a profile secret, not a global.
-_GLOBAL_ENV_EXACT = frozenset({
-    # Hermes runtime / deployment
-    "HERMES_HOME", "HERMES_PROFILE", "HERMES_GATEWAY_LOCK_DIR",
-    "HERMES_MAX_ITERATIONS", "HERMES_MAX_TOKENS", "HERMES_API_TIMEOUT",
-    "HERMES_REDACT_SECRETS", "HERMES_NOUS_TIMEOUT_SECONDS",
-    "_HERMES_GATEWAY",
-    # OS / interpreter
-    "PATH", "HOME", "USER", "LANG", "LC_ALL", "TZ", "PWD", "SHELL", "TMPDIR",
-    "VIRTUAL_ENV", "PYTHONPATH", "SSL_CERT_FILE",
-    # Kanban paths (per-board, not per-profile-secret)
-    "HERMES_KANBAN_DB", "HERMES_KANBAN_WORKSPACES_ROOT", "HERMES_KANBAN_BOARD",
-})
-_GLOBAL_ENV_PREFIXES = (
-    "HERMES_KANBAN_",
-    "HERMES_TELEGRAM_",   # tuning knobs (batch delays, fallback toggles) — NOT the token
-    "TERMINAL_",          # terminal/sandbox backend settings
-)
-
-
-def _is_global_env(name: str) -> bool:
-    """Return True for genuinely process-global (non-profile-secret) env vars."""
-    if name in _GLOBAL_ENV_EXACT:
-        return True
-    return any(name.startswith(p) for p in _GLOBAL_ENV_PREFIXES)
-
-
-def get_secret(name: str, default: Optional[str] = None) -> Optional[str]:
-    """Resolve a credential by env-var name, honoring the active profile scope.
-
-    Resolution order:
-
-    1. Genuinely-global vars (``_is_global_env``) always read ``os.environ`` —
-       they are deployment settings, not profile secrets.
-    2. When a secret scope is installed (multiplexed turn), read from it; an
-       absent key returns ``default``. The scope is authoritative — we do NOT
-       fall through to ``os.environ``, because in a multiplexer ``os.environ``
-       may hold another profile's value.
-    3. No scope installed:
-       - multiplex INACTIVE (default deployment): read ``os.environ`` —
-         identical to the legacy ``os.getenv`` behavior every caller had before.
-       - multiplex ACTIVE: FAIL CLOSED. Raise ``UnscopedSecretError`` so the
-         missing scope is caught loudly instead of leaking a cross-profile value.
-    """
-    if _is_global_env(name):
-        val = os.environ.get(name)
-        return val if val is not None else default
-
-    scope = _SECRET_SCOPE.get()
-    if scope is not None:
-        val = scope.get(name)
-        return val if val is not None else default
-
-    if _MULTIPLEX_ACTIVE:
-        raise UnscopedSecretError(
-            f"get_secret({name!r}) called with no profile secret scope active "
-            f"while multiplexing is on. This credential read must run inside a "
-            f"set_secret_scope(...) block (the per-turn / per-adapter profile "
-            f"scope). Reading os.environ here would risk leaking another "
-            f"profile's value. See docs/design/multiplexing-gateway.md "
-            f"(Workstream A)."
-        )
-
-    val = os.environ.get(name)
-    return val if val is not None else default
-
-
-def load_env_file(env_path: Path) -> Dict[str, str]:
-    """Parse a ``.env`` file into a plain dict WITHOUT touching ``os.environ``.
-
-    Used to load a profile's secrets into an isolated mapping for
-    ``set_secret_scope``. Mirrors python-dotenv's basic parsing (KEY=VALUE,
-    ``export`` prefix, ``#`` comments, optional matching quotes) but never
-    mutates the process environment — that isolation is the whole point.
-    """
-    secrets: Dict[str, str] = {}
-    try:
-        text = env_path.read_text(encoding="utf-8")
-    except (FileNotFoundError, OSError, UnicodeDecodeError):
-        return secrets
-
-    for raw in text.splitlines():
-        line = raw.strip()
-        if not line or line.startswith("#"):
-            continue
-        if line.startswith("export "):
-            line = line[len("export "):].lstrip()
-        if "=" not in line:
-            continue
-        key, _, value = line.partition("=")
-        key = key.strip()
-        if not key:
-            continue
-        value = value.strip()
-        if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
-            value = value[1:-1]
-        secrets[key] = value
-
-    return secrets
-
-
-def build_profile_secret_scope(hermes_home: Path) -> Dict[str, str]:
-    """Build a profile's secret mapping from its ``<home>/.env``.
-
-    Returns a fresh dict (safe to install via ``set_secret_scope``). Genuinely
-    global vars are intentionally NOT copied in — ``get_secret`` reads those
-    from ``os.environ`` directly, so the scope holds only profile secrets.
-    """
-    return load_env_file(Path(hermes_home) / ".env")
-
@@ -545,13 +545,6 @@ class GatewayConfig:
    thread_sessions_per_user: bool = False  # When False (default), threads are shared across all participants
    max_concurrent_sessions: Optional[int] = None  # Positive int caps simultaneous active chat sessions

-    # Multi-profile multiplexing (opt-in; default off preserves one-gateway-per-profile).
-    # When True, the default profile's gateway serves inbound messages for every
-    # profile on the host: profiles are stamped into session keys and (in later
-    # phases) per-profile adapters/credentials are resolved. When False, the
-    # gateway behaves exactly as before — single HERMES_HOME, no profile stamping.
-    multiplex_profiles: bool = False
-
    # Unauthorized DM policy
    unauthorized_dm_behavior: str = "pair"  # "pair" or "ignore"

@@ -657,7 +650,6 @@ class GatewayConfig:
            "group_sessions_per_user": self.group_sessions_per_user,
            "thread_sessions_per_user": self.thread_sessions_per_user,
            "max_concurrent_sessions": self.max_concurrent_sessions,
-            "multiplex_profiles": self.multiplex_profiles,
            "unauthorized_dm_behavior": self.unauthorized_dm_behavior,
            "streaming": self.streaming.to_dict(),
            "session_store_max_age_days": self.session_store_max_age_days,
@@ -703,12 +695,7 @@ class GatewayConfig:

        group_sessions_per_user = data.get("group_sessions_per_user")
        thread_sessions_per_user = data.get("thread_sessions_per_user")
-        multiplex_profiles = data.get("multiplex_profiles")
        nested_gateway = data.get("gateway") if isinstance(data.get("gateway"), dict) else {}
-        if multiplex_profiles is None and isinstance(nested_gateway, dict):
-            # Also honor gateway.multiplex_profiles written by
-            # ``hermes config set gateway.multiplex_profiles true``.
-            multiplex_profiles = nested_gateway.get("multiplex_profiles")
        if "max_concurrent_sessions" in data:
            max_concurrent_raw = data.get("max_concurrent_sessions")
            max_concurrent_key = "max_concurrent_sessions"
@@ -745,7 +732,6 @@ class GatewayConfig:
            stt_enabled=_coerce_bool(stt_enabled, True),
            group_sessions_per_user=_coerce_bool(group_sessions_per_user, True),
            thread_sessions_per_user=_coerce_bool(thread_sessions_per_user, False),
-            multiplex_profiles=_coerce_bool(multiplex_profiles, False),
            max_concurrent_sessions=max_concurrent_sessions,
            unauthorized_dm_behavior=unauthorized_dm_behavior,
            streaming=StreamingConfig.from_dict(data.get("streaming", {})),
@@ -837,13 +823,6 @@ def load_gateway_config() -> GatewayConfig:
            if "thread_sessions_per_user" in yaml_cfg:
                gw_data["thread_sessions_per_user"] = yaml_cfg["thread_sessions_per_user"]

-            # Multiplexing flag: accept both the top-level key and the nested
-            # gateway.multiplex_profiles form (from_dict resolves the nested
-            # fallback, but surface the top-level key here for parity with the
-            # other session-scope flags above).
-            if "multiplex_profiles" in yaml_cfg:
-                gw_data["multiplex_profiles"] = yaml_cfg["multiplex_profiles"]
-
            gateway_section = yaml_cfg.get("gateway")
            if isinstance(gateway_section, dict) and "max_concurrent_sessions" in gateway_section:
                gw_data["max_concurrent_sessions"] = gateway_section["max_concurrent_sessions"]
@@ -57,11 +57,6 @@ from gateway.platforms.base import (

 logger = logging.getLogger(__name__)

-# Sentinel returned by _resolve_request_profile when a /p/<profile>/ prefix
-# names a profile this gateway does not serve (→ 404). Distinct from None
-# (no prefix / multiplexing off → handle as the default profile).
-_PROFILE_REJECTED = object()
-
 _BUILTIN_DELIVER_PLATFORMS = {
    "telegram", "discord", "slack", "signal", "sms", "whatsapp",
    "matrix", "mattermost", "homeassistant", "email", "dingtalk",
@@ -194,14 +189,6 @@ class WebhookAdapter(BasePlatformAdapter):
        app = web.Application()
        app.router.add_get("/health", self._handle_health)
        app.router.add_post("/webhooks/{route_name}", self._handle_webhook)
-        # Multi-profile multiplexing: a /p/<profile>/webhooks/<route> prefix
-        # routes the inbound event to that profile. Same handler; the profile is
-        # captured from the path and stamped onto the SessionSource so the agent
-        # turn resolves that profile's config/skills/credentials. Only honored
-        # when gateway.multiplex_profiles is on (the handler validates).
-        app.router.add_post(
-            "/p/{profile}/webhooks/{route_name}", self._handle_webhook
-        )

        # Port conflict detection — fail fast if port is already in use
        import socket as _socket
@@ -410,35 +397,6 @@ class WebhookAdapter(BasePlatformAdapter):
        except Exception as e:
            logger.error("[webhook] Failed to reload dynamic routes: %s", e)

-    def _resolve_request_profile(self, request: "web.Request"):
-        """Resolve + validate the /p/<profile>/ URL prefix on a webhook request.
-
-        Returns:
-          - ``None`` when no profile prefix is present, or multiplexing is off
-            (the prefix is ignored, request handled as the default profile).
-          - the profile name (str) when present, multiplexing is on, and the
-            profile is one this gateway serves.
-          - ``_PROFILE_REJECTED`` when a prefix is present but the profile is
-            unknown/unconfigured (handler returns 404).
-        """
-        profile = (request.match_info.get("profile") or "").strip()
-        if not profile:
-            return None
-        runner = self.gateway_runner
-        cfg = getattr(runner, "config", None)
-        if not getattr(cfg, "multiplex_profiles", False):
-            # Prefix supplied but multiplexing is off — ignore it, behave as
-            # the single-profile gateway (don't 404 a would-be valid route).
-            return None
-        try:
-            from hermes_cli.profiles import profiles_to_serve
-            served = {name for name, _ in profiles_to_serve(multiplex=True)}
-        except Exception:
-            return _PROFILE_REJECTED
-        if profile not in served:
-            return _PROFILE_REJECTED
-        return profile
-
    async def _handle_webhook(self, request: "web.Request") -> "web.Response":
        """POST /webhooks/{route_name} — receive and process a webhook event."""
        # Hot-reload dynamic subscriptions on each request (mtime-gated, cheap)
@@ -447,13 +405,6 @@ class WebhookAdapter(BasePlatformAdapter):
        route_name = request.match_info.get("route_name", "")
        route_config = self._routes.get(route_name)

-        # Multi-profile: resolve + validate the /p/<profile>/ prefix if present.
-        profile = self._resolve_request_profile(request)
-        if profile is _PROFILE_REJECTED:
-            return web.json_response(
-                {"error": "Unknown or unconfigured profile"}, status=404
-            )
-
        if not route_config:
            return web.json_response(
                {"error": f"Unknown route: {route_name}"}, status=404
@@ -690,8 +641,6 @@ class WebhookAdapter(BasePlatformAdapter):
            user_id=f"webhook:{route_name}",
            user_name=route_name,
        )
-        if profile and isinstance(profile, str):
-            source.profile = profile
        event = MessageEvent(
            text=prompt,
            message_type=MessageType.TEXT,
@@ -113,6 +113,227 @@ def relay_inbound_config() -> tuple[Optional[str], Optional[str], int]:
    return (key or None, host or "0.0.0.0", port)


+def relay_endpoint() -> Optional[str]:
+    """The gateway's own PUBLIC inbound URL, asserted to the connector at provision.
+
+    The connector delivers signed inbound POSTs to this URL and stores it on the
+    tenant's route rows. It is gateway-asserted (the connector scopes it to the
+    verified tenant, so a dishonest gateway can only misdirect its OWN inbound).
+    The *source* of the value differs by deployment but the code path is uniform:
+    a self-hosted operator sets ``GATEWAY_RELAY_ENDPOINT`` (mirrors how they set
+    ``HERMES_DASHBOARD_PUBLIC_URL``); a hosted/NAS container has the same var
+    stamped in (NAS knows the public URL only in that case). Absent -> the
+    gateway provisions outbound-only (no inbound routes written).
+
+    Env first (Docker), then ``gateway.relay_endpoint`` in config.yaml.
+    """
+    url = os.environ.get("GATEWAY_RELAY_ENDPOINT", "").strip()
+    if not url:
+        try:
+            from gateway.run import _load_gateway_config  # late import to avoid cycle
+
+            cfg = (_load_gateway_config().get("gateway") or {})
+            url = str(cfg.get("relay_endpoint", "") or "").strip()
+        except Exception:  # noqa: BLE001 - config absence/parse must never crash boot
+            url = ""
+    return url.rstrip("/") or None
+
+
+def relay_route_keys() -> list[str]:
+    """Discriminators (guild_ids / chat_ids / paths) this gateway's tenant owns.
+
+    Gateway-provided config, paired with ``relay_endpoint()``: the connector
+    writes one route row per (routeKey -> tenant, endpoint), so route keys only
+    take effect alongside an endpoint. Empty -> outbound-only provisioning (the
+    connector accepts an empty set and writes no route rows).
+
+    ``GATEWAY_RELAY_ROUTE_KEYS`` is comma-separated; config.yaml
+    ``gateway.relay_route_keys`` may be a list or a comma string.
+    """
+    raw = os.environ.get("GATEWAY_RELAY_ROUTE_KEYS", "").strip()
+    if not raw:
+        try:
+            from gateway.run import _load_gateway_config  # late import to avoid cycle
+
+            cfg = (_load_gateway_config().get("gateway") or {})
+            val = cfg.get("relay_route_keys", "")
+            if isinstance(val, (list, tuple)):
+                return [str(k).strip() for k in val if str(k).strip()]
+            raw = str(val or "").strip()
+        except Exception:  # noqa: BLE001
+            raw = ""
+    return [k.strip() for k in raw.split(",") if k.strip()]
+
+
+def _provision_url(relay_dial_url: str) -> str:
+    """Map the ``ws(s)://…/relay`` dial URL to the ``http(s)://…/relay/provision`` POST URL."""
+    raw = relay_dial_url.rstrip("/")
+    if raw.startswith("ws://"):
+        raw = "http://" + raw[len("ws://"):]
+    elif raw.startswith("wss://"):
+        raw = "https://" + raw[len("wss://"):]
+    if raw.endswith("/relay"):
+        raw = raw[: -len("/relay")]
+    return f"{raw}/relay/provision"
+
+
+def _post_provision(
+    *,
+    provision_url: str,
+    access_token: str,
+    gateway_id: str,
+    platform: str,
+    bot_id: str,
+    gateway_endpoint: Optional[str],
+    route_keys: list[str],
+    timeout: float = 15.0,
+) -> dict:
+    """POST to the connector's ``/relay/provision`` and return the JSON body.
+
+    The connector validates ``access_token`` against NAS, derives the
+    authoritative tenant, mints the per-gateway secret + per-tenant delivery key,
+    upserts the tenant's route rows, and returns
+    ``{secret, deliveryKey, tenant, gatewayId, routeKeys}``. Raises RuntimeError
+    with a user-facing message on any non-2xx / transport failure.
+    """
+    import json
+    import urllib.error
+    import urllib.request
+
+    body: dict = {
+        "gatewayId": gateway_id,
+        "platform": platform,
+        "botId": bot_id,
+        "gatewayEndpoint": gateway_endpoint or "",
+        "routeKeys": route_keys,
+    }
+    data = json.dumps(body).encode("utf-8")
+    req = urllib.request.Request(
+        provision_url,
+        data=data,
+        method="POST",
+        headers={
+            "Authorization": f"Bearer {access_token}",
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        },
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            payload = json.loads(resp.read().decode())
+    except urllib.error.HTTPError as exc:
+        detail = ""
+        try:
+            detail = (json.loads(exc.read().decode()) or {}).get("error", "")
+        except Exception:
+            pass
+        raise RuntimeError(
+            f"connector returned HTTP {exc.code}" + (f": {detail}" if detail else "")
+        ) from exc
+    except urllib.error.URLError as exc:
+        raise RuntimeError(f"could not reach connector: {exc.reason}") from exc
+
+    if not isinstance(payload, dict) or not payload.get("secret"):
+        raise RuntimeError("connector returned an unexpected response (no secret)")
+    return payload
+
+
+def self_provision_if_managed() -> bool:
+    """Managed-boot self-provision: mint relay creds in-process, no human, no disk.
+
+    Fires only on a MANAGED boot (``is_managed()``) with relay configured
+    (``relay_url()`` set) and NO per-gateway secret already present. In that case
+    the runtime resolves the agent's own Nous access token (the same
+    ``resolve_nous_access_token()`` the enroll CLI / dashboard register use),
+    POSTs ``/relay/provision`` asserting its own endpoint + route keys, and sets
+    ``GATEWAY_RELAY_ID`` / ``GATEWAY_RELAY_SECRET`` / ``GATEWAY_RELAY_DELIVERY_KEY``
+    into ``os.environ`` so the subsequent ``register_relay_adapter()`` picks them
+    up. The creds live ONLY in process memory — never written to ``~/.hermes/.env``
+    (``save_env_value`` refuses under managed anyway, and keeping the secret off
+    any volume is the stronger posture).
+
+    Stateless: process-env creds don't survive a restart, so a managed container
+    re-provisions every boot; the connector's rotation window covers a still-
+    connected prior instance. An explicitly-pinned ``GATEWAY_RELAY_SECRET`` (env
+    or config) is RESPECTED — self-provision skips so an operator pin isn't
+    stomped.
+
+    Returns True if it provisioned, False otherwise. NEVER raises: a provision
+    failure logs and returns False so the gateway still boots (and
+    ``register_relay_adapter`` will simply dial unauthenticated / be rejected,
+    rather than the whole gateway crashing).
+    """
+    import logging
+
+    logger = logging.getLogger("gateway.relay")
+
+    try:
+        from hermes_cli.config import is_managed
+    except Exception:  # noqa: BLE001
+        return False
+
+    if not is_managed():
+        return False
+    dial_url = relay_url()
+    if not dial_url:
+        return False
+
+    # Respect an already-present (pinned/stamped) secret — don't stomp it.
+    existing_id, existing_secret = relay_connection_auth()
+    if existing_id and existing_secret:
+        logger.info("relay self-provision skipped: GATEWAY_RELAY_SECRET already set")
+        return False
+
+    try:
+        from hermes_cli.auth import resolve_nous_access_token
+
+        access_token = resolve_nous_access_token()
+    except Exception as exc:  # noqa: BLE001 - boot must survive a token failure
+        logger.warning("relay self-provision skipped: could not resolve Nous token (%s)", exc)
+        return False
+
+    platform, bot_id = relay_platform_identity()
+    # gatewayId default mirrors the enroll CLI's hostname-based slug.
+    import socket
+
+    try:
+        host = socket.gethostname().strip()
+    except Exception:  # noqa: BLE001
+        host = ""
+    gateway_id = os.environ.get("GATEWAY_RELAY_ID", "").strip() or f"gw-{host or 'hermes'}"
+    endpoint = relay_endpoint()
+    route_keys = relay_route_keys()
+
+    try:
+        result = _post_provision(
+            provision_url=_provision_url(dial_url),
+            access_token=access_token,
+            gateway_id=gateway_id,
+            platform=platform,
+            bot_id=bot_id,
+            gateway_endpoint=endpoint,
+            route_keys=route_keys,
+        )
+    except RuntimeError as exc:
+        logger.warning("relay self-provision failed (%s); gateway will boot without relay auth", exc)
+        return False
+
+    # Set creds in-process so register_relay_adapter() + relay_inbound_config()
+    # read them from os.environ. Never logged.
+    os.environ["GATEWAY_RELAY_ID"] = str(result.get("gatewayId") or gateway_id)
+    os.environ["GATEWAY_RELAY_SECRET"] = str(result.get("secret") or "")
+    os.environ["GATEWAY_RELAY_DELIVERY_KEY"] = str(result.get("deliveryKey") or "")
+    tenant = str(result.get("tenant") or "")
+    logger.info(
+        "relay self-provisioned (gateway_id=%s tenant=%s routes=%d inbound=%s)",
+        os.environ["GATEWAY_RELAY_ID"],
+        tenant or "?",
+        len(route_keys),
+        "yes" if endpoint else "outbound-only",
+    )
+    return True
+
+
 def register_relay_adapter(force: bool = False, url: Optional[str] = None) -> bool:
    """Register the generic ``relay`` platform via the platform registry.

@@ -1173,31 +1173,13 @@ def _reload_runtime_env_preserving_config_authority() -> None:
    pick up rotated API keys. config.yaml remains authoritative for agent budget
    settings such as agent.max_turns; otherwise a stale HERMES_MAX_ITERATIONS in
    .env can replace the startup bridge on later turns.
-
-    In multiplex mode this is a NO-OP for the credential reload: secrets come
-    from the per-turn ``set_secret_scope`` (installed by ``_profile_runtime_scope``)
-    which loads the routed profile's ``.env`` into an isolated mapping. Mutating
-    the process-global ``os.environ`` here would defeat that isolation and leak
-    the default profile's keys to every profile's turns and subprocesses.
    """
-    from agent.secret_scope import is_multiplex_active
-    if is_multiplex_active():
-        # Credentials are resolved from the active profile's secret scope, not
-        # os.environ. Still honor config.yaml's agent.max_turns bridge below
-        # using the scoped home, but never reload .env into global env.
-        _bridge_max_turns_from_config(_hermes_home)
-        return
-
    load_hermes_dotenv(
        hermes_home=_hermes_home,
        project_env=Path(__file__).resolve().parents[1] / '.env',
    )
-    _bridge_max_turns_from_config(_hermes_home)

-
-def _bridge_max_turns_from_config(home: "Path") -> None:
-    """Bridge config.yaml agent.max_turns into HERMES_MAX_ITERATIONS (a global)."""
-    config_path = home / 'config.yaml'
+    config_path = _hermes_home / 'config.yaml'
    if not config_path.exists():
        return
    try:
@@ -1214,44 +1196,6 @@ def _bridge_max_turns_from_config(home: "Path") -> None:
        os.environ["HERMES_MAX_ITERATIONS"] = str(agent_cfg["max_turns"])


-from contextlib import contextmanager as _contextmanager
-
-
-@_contextmanager
-def _profile_runtime_scope(profile_home: "Path"):
-    """Scope config/skills/memory AND credentials to a profile for one turn.
-
-    Combines the two seams the multiplexer needs:
-      1. ``set_hermes_home_override`` — redirects ``get_hermes_home()`` (config,
-         skills, memory, SOUL, sessions) to the profile's home. Contextvar, so
-         it propagates into the agent worker thread via ``copy_context()``.
-      2. ``set_secret_scope`` — installs the profile's ``.env`` secrets as the
-         authoritative credential source, so ``get_secret`` reads this profile's
-         keys and never the process-global ``os.environ`` (which in a
-         multiplexer may hold another profile's values).
-
-    Only used on the multiplexed inbound path. Single-profile gateways never
-    enter this scope, so their behavior is unchanged. Loading the profile's
-    ``.env`` here does NOT mutate ``os.environ`` — ``build_profile_secret_scope``
-    returns an isolated dict — which is what keeps subprocesses (MCP, kanban)
-    from inheriting cross-profile secrets.
-    """
-    from hermes_constants import set_hermes_home_override, reset_hermes_home_override
-    from agent.secret_scope import (
-        build_profile_secret_scope,
-        set_secret_scope,
-        reset_secret_scope,
-    )
-
-    home_token = set_hermes_home_override(str(profile_home))
-    secret_token = set_secret_scope(build_profile_secret_scope(Path(profile_home)))
-    try:
-        yield
-    finally:
-        reset_secret_scope(secret_token)
-        reset_hermes_home_override(home_token)
-
-
 _DOCKER_VOLUME_SPEC_RE = re.compile(r"^(?P<host>.+):(?P<container>/[^:]+?)(?::(?P<options>[^:]+))?$")
 _DOCKER_MEDIA_OUTPUT_CONTAINER_PATHS = {"/output", "/outputs"}

@@ -2296,22 +2240,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
    def __init__(self, config: Optional[GatewayConfig] = None):
        global _gateway_runner_ref
        self.config = config or load_gateway_config()
-        # Mark the process as a profile multiplexer when configured. This flips
-        # agent.secret_scope.get_secret() to fail-closed on any unscoped
-        # credential read, so a missed migration crashes loudly instead of
-        # leaking a cross-profile value (Workstream A). Inert when off.
-        try:
-            from agent.secret_scope import set_multiplex_active
-            set_multiplex_active(bool(getattr(self.config, "multiplex_profiles", False)))
-        except Exception:
-            logger.debug("could not set multiplex-active flag", exc_info=True)
        self.adapters: Dict[Platform, BasePlatformAdapter] = {}
-        # Multi-profile multiplexing: adapters for NON-default profiles live
-        # here, keyed by profile name then Platform. self.adapters stays the
-        # default/active profile's map so the ~93 existing self.adapters[...]
-        # sites are untouched when multiplexing is off (this dict is empty).
-        # Populated by _start_secondary_profile_adapters().
-        self._profile_adapters: Dict[str, Dict[Platform, BasePlatformAdapter]] = {}
        self._warn_if_docker_media_delivery_is_risky()
        _gateway_runner_ref = _weakref.ref(self)

@@ -2863,24 +2792,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
            except Exception:
                pass
        config = getattr(self, "config", None)
-        # Mirror SessionStore._resolve_profile_for_key so this fallback path
-        # produces the same namespace as the primary path: None (legacy
-        # agent:main) unless multiplexing is on, then the active profile.
-        _profile = None
-        if getattr(config, "multiplex_profiles", False):
-            if source.profile:
-                _profile = source.profile
-            else:
-                try:
-                    from hermes_cli.profiles import get_active_profile_name
-                    _profile = get_active_profile_name() or "default"
-                except Exception:
-                    _profile = None
        return build_session_key(
            source,
            group_sessions_per_user=getattr(config, "group_sessions_per_user", True),
            thread_sessions_per_user=getattr(config, "thread_sessions_per_user", False),
-            profile=_profile,
        )

    def _telegram_topic_mode_enabled(self, source: SessionSource) -> bool:
@@ -5201,7 +5116,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
        # adapter dials the connector over a WebSocket, negotiates its capability
        # descriptor at handshake, and bridges inbound/outbound like any platform.
        try:
-            from gateway.relay import register_relay_adapter, relay_url
+            from gateway.relay import (
+                register_relay_adapter,
+                relay_url,
+                self_provision_if_managed,
+            )
+
+            # Managed boot: self-provision relay creds in-process (resolve the
+            # agent's NAS token -> POST /relay/provision -> set GATEWAY_RELAY_* in
+            # os.environ) BEFORE registration reads them. No-op when not managed,
+            # relay unconfigured, or a secret is already pinned. Never raises.
+            self_provision_if_managed()

            if register_relay_adapter():
                logger.info("relay adapter registered (connector at %s)", relay_url())
@@ -5409,17 +5334,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                    "attempts": 1,
                    "next_retry": time.monotonic() + 30,
                }
-
-        # Multi-profile multiplexing: bring up adapters for every OTHER profile
-        # this gateway serves. Each profile's adapters connect under that
-        # profile's home + credential scope and stamp their inbound events with
-        # the profile so the agent turn resolves correctly. No-op when off.
-        try:
-            _secondary_connected = await self._start_secondary_profile_adapters()
-            connected_count += _secondary_connected
-        except Exception as e:
-            logger.error("Secondary-profile adapter startup failed: %s", e, exc_info=True)
-
+        
        if connected_count == 0:
            if startup_nonretryable_errors:
                reason = "; ".join(startup_nonretryable_errors)
@@ -6426,22 +6341,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                        time.monotonic() - _adapter_started_at,
                        e,
                    )
-
-            # Disconnect secondary-profile adapters (multiplex mode).
-            for _prof, _amap in list(getattr(self, "_profile_adapters", {}).items()):
-                for platform, adapter in list(_amap.items()):
-                    try:
-                        await adapter.cancel_background_tasks()
-                    except Exception as e:
-                        logger.debug("✗ %s bg-cancel error (profile %s): %s", platform.value, _prof, e)
-                    try:
-                        await adapter.disconnect()
-                        logger.info("✓ %s disconnected (profile: %s)", platform.value, _prof)
-                    except Exception as e:
-                        logger.error("✗ %s disconnect error (profile %s): %s", platform.value, _prof, e)
-                _amap.clear()
-            if hasattr(self, "_profile_adapters"):
-                self._profile_adapters.clear()
            logger.info(
                "Shutdown phase: all adapters disconnected at +%.2fs",
                _phase_elapsed(),
@@ -6611,155 +6510,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
        """Wait for shutdown signal."""
        await self._shutdown_event.wait()

-    async def _start_secondary_profile_adapters(self) -> int:
-        """Bring up adapters for every non-active profile this gateway serves.
-
-        Returns the number of secondary adapters that connected. No-op (returns
-        0) unless ``gateway.multiplex_profiles`` is on.
-
-        Each profile's adapters are created and connected under that profile's
-        HERMES_HOME + secret scope (``_profile_runtime_scope``), stored in
-        ``self._profile_adapters[profile]``, and given a message handler that
-        stamps ``source.profile`` before delegating to the shared
-        ``_handle_message`` — so the agent turn resolves that profile's config,
-        skills, and credentials. Same-platform credential collisions (two
-        profiles polling the same bot token) are detected and refused here, the
-        only point that sees every profile's resolved credentials together.
-        """
-        if not getattr(self.config, "multiplex_profiles", False):
-            return 0
-
-        try:
-            from hermes_cli.profiles import profiles_to_serve, get_active_profile_name
-        except Exception:
-            return 0
-
-        active = get_active_profile_name() or "default"
-        connected = 0
-        # (platform, token-fingerprint) -> profile that claimed it. Detects two
-        # profiles trying to poll the same bot credential (impossible to do
-        # concurrently). Seed with the active profile's adapters.
-        claimed: Dict[tuple, str] = {}
-        for _plat, _ad in self.adapters.items():
-            fp = self._adapter_credential_fingerprint(_ad)
-            if fp is not None:
-                claimed[(_plat, fp)] = active
-
-        for profile_name, profile_home in profiles_to_serve(multiplex=True):
-            if profile_name == active:
-                continue  # handled by the primary startup loop
-            try:
-                connected += await self._start_one_profile_adapters(
-                    profile_name, profile_home, claimed
-                )
-            except Exception as e:
-                logger.error(
-                    "Failed to start adapters for profile '%s': %s",
-                    profile_name, e, exc_info=True,
-                )
-
-        # Record served profiles in runtime status for `hermes status`.
-        try:
-            from gateway.status import write_runtime_status
-            served = [active] + sorted(self._profile_adapters.keys())
-            write_runtime_status(served_profiles=served)
-        except Exception:
-            logger.debug("could not record served_profiles", exc_info=True)
-
-        return connected
-
-    async def _start_one_profile_adapters(
-        self, profile_name: str, profile_home: "Path", claimed: Dict[tuple, str]
-    ) -> int:
-        """Create+connect one profile's adapters under its runtime scope."""
-        from gateway.config import load_gateway_config
-
-        with _profile_runtime_scope(profile_home):
-            profile_cfg = load_gateway_config()
-
-        profile_map = self._profile_adapters.setdefault(profile_name, {})
-        connected = 0
-        for platform, platform_config in profile_cfg.platforms.items():
-            if not platform_config.enabled:
-                continue
-            with _profile_runtime_scope(profile_home):
-                adapter = self._create_adapter(platform, platform_config)
-            if not adapter:
-                continue
-
-            # Same-token conflict detection — refuse a duplicate poll.
-            fp = self._adapter_credential_fingerprint(adapter)
-            if fp is not None:
-                owner = claimed.get((platform, fp))
-                if owner is not None:
-                    logger.error(
-                        "Profile '%s' and '%s' both configure %s with the same "
-                        "credential — refusing to start the duplicate (a single "
-                        "bot token cannot be polled twice). Give each profile its "
-                        "own %s credential.",
-                        owner, profile_name, platform.value, platform.value,
-                    )
-                    await self._safe_adapter_disconnect(adapter, platform)
-                    continue
-                claimed[(platform, fp)] = profile_name
-
-            # Stamp every inbound event from this adapter with its profile so
-            # the agent turn (and session key) resolve to the right home.
-            adapter.set_message_handler(
-                self._make_profile_message_handler(profile_name)
-            )
-            adapter.set_fatal_error_handler(self._handle_adapter_fatal_error)
-            adapter.set_session_store(self.session_store)
-            adapter.set_busy_session_handler(self._handle_active_session_busy_message)
-            adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id)
-            adapter._busy_text_mode = self._busy_text_mode
-
-            try:
-                with _profile_runtime_scope(profile_home):
-                    success = await self._connect_adapter_with_timeout(adapter, platform)
-                if success:
-                    profile_map[platform] = adapter
-                    connected += 1
-                    logger.info("✓ %s connected (profile: %s)", platform.value, profile_name)
-                else:
-                    logger.warning("✗ %s failed to connect (profile: %s)", platform.value, profile_name)
-                    await self._safe_adapter_disconnect(adapter, platform)
-            except Exception as e:
-                logger.error("✗ %s error (profile: %s): %s", platform.value, profile_name, e)
-                await self._safe_adapter_disconnect(adapter, platform)
-        return connected
-
-    def _make_profile_message_handler(self, profile_name: str):
-        """Return a message handler that stamps source.profile then delegates."""
-        async def _handler(event):
-            try:
-                if getattr(event, "source", None) is not None and not event.source.profile:
-                    event.source.profile = profile_name
-            except Exception:
-                pass
-            return await self._handle_message(event)
-        return _handler
-
-    @staticmethod
-    def _adapter_credential_fingerprint(adapter: Any) -> Optional[str]:
-        """Return a stable, log-safe fingerprint of an adapter's credential.
-
-        Used only to detect two profiles claiming the same bot token. Returns a
-        salted hash (never the token itself) of the adapter's primary
-        credential, or None when no credential is discoverable (in which case
-        we don't attempt conflict detection for it).
-        """
-        token = None
-        for attr in ("token", "bot_token", "_token", "api_token", "_bot_token"):
-            val = getattr(adapter, attr, None)
-            if isinstance(val, str) and val.strip():
-                token = val.strip()
-                break
-        if not token:
-            return None
-        import hashlib
-        return hashlib.sha256(("hermes-mux:" + token).encode("utf-8")).hexdigest()[:16]
-
    def _create_adapter(
        self, 
        platform: Platform, 
@@ -13989,64 +13739,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
        channel_prompt: Optional[str] = None,
        persist_user_message: Optional[str] = None,
        persist_user_timestamp: Optional[float] = None,
-    ) -> Dict[str, Any]:
-        """Profile-scoping wrapper around the agent run.
-
-        When multiplexing is active, resolve the inbound source's profile and
-        run the whole turn inside ``_profile_runtime_scope`` so config/skills/
-        memory resolve to that profile's home AND credentials resolve from that
-        profile's secret scope (never the process-global ``os.environ``). When
-        multiplexing is off this is a transparent pass-through — zero behavior
-        change for single-profile gateways.
-        """
-        if not getattr(getattr(self, "config", None), "multiplex_profiles", False):
-            return await self._run_agent_inner(
-                message, context_prompt, history, source, session_id,
-                session_key=session_key, run_generation=run_generation,
-                _interrupt_depth=_interrupt_depth, event_message_id=event_message_id,
-                channel_prompt=channel_prompt, persist_user_message=persist_user_message,
-                persist_user_timestamp=persist_user_timestamp,
-            )
-
-        profile_home = self._resolve_profile_home_for_source(source)
-        with _profile_runtime_scope(profile_home):
-            return await self._run_agent_inner(
-                message, context_prompt, history, source, session_id,
-                session_key=session_key, run_generation=run_generation,
-                _interrupt_depth=_interrupt_depth, event_message_id=event_message_id,
-                channel_prompt=channel_prompt, persist_user_message=persist_user_message,
-                persist_user_timestamp=persist_user_timestamp,
-            )
-
-    def _resolve_profile_home_for_source(self, source: SessionSource) -> "Path":
-        """Resolve which profile's HERMES_HOME should serve this inbound source.
-
-        Prefers the profile the source was routed to (``source.profile`` — set
-        by the /p/<profile>/ URL prefix or a per-credential adapter), falling
-        back to the active profile (the multiplexer's own home).
-        """
-        from hermes_cli.profiles import get_active_profile_name, get_profile_dir
-        try:
-            name = (source.profile or "").strip() or get_active_profile_name() or "default"
-            return get_profile_dir(name)
-        except Exception:
-            from hermes_constants import get_hermes_home
-            return get_hermes_home()
-
-    async def _run_agent_inner(
-        self,
-        message: str,
-        context_prompt: str,
-        history: List[Dict[str, Any]],
-        source: SessionSource,
-        session_id: str,
-        session_key: str = None,
-        run_generation: Optional[int] = None,
-        _interrupt_depth: int = 0,
-        event_message_id: Optional[str] = None,
-        channel_prompt: Optional[str] = None,
-        persist_user_message: Optional[str] = None,
-        persist_user_timestamp: Optional[float] = None,
    ) -> Dict[str, Any]:
        """
        Run the agent with the given message and context.
@@ -92,11 +92,6 @@ class SessionSource:
    parent_chat_id: Optional[str] = None  # Parent channel when chat_id refers to a thread
    message_id: Optional[str] = None  # ID of the triggering message (for pin/reply/react)
    role_authorized: bool = False  # True when adapter granted access via role (not user ID)
-    # Profile this inbound message is routed to in a multiplexing gateway
-    # (from the /p/<profile>/ URL prefix or per-credential adapter ownership).
-    # None => the gateway's active/default profile. Drives both session-key
-    # namespacing and the per-turn config/credential scope.
-    profile: Optional[str] = None
    
    @property
    def description(self) -> str:
@@ -140,8 +135,6 @@ class SessionSource:
            d["parent_chat_id"] = self.parent_chat_id
        if self.message_id:
            d["message_id"] = self.message_id
-        if self.profile:
-            d["profile"] = self.profile
        return d

    @classmethod
@@ -160,7 +153,6 @@ class SessionSource:
            guild_id=data.get("guild_id"),
            parent_chat_id=data.get("parent_chat_id"),
            message_id=data.get("message_id"),
-            profile=data.get("profile"),
        )
    

@@ -623,41 +615,15 @@ def is_shared_multi_user_session(
    return not group_sessions_per_user


-def _session_key_namespace(profile: Optional[str]) -> str:
-    """Return the ``agent:<ns>`` namespace prefix for a session key.
-
-    The historical key format is ``agent:main:<platform>:<chat_type>:...`` where
-    ``main`` is a static namespace literal (NOT a branch name — branching keys
-    off ``session_id``, not this slot). Multi-profile multiplexing reuses this
-    slot to carry the profile:
-
-    - default profile (or ``None``/``""``/``"default"``) → ``agent:main`` —
-      BYTE-IDENTICAL to every key ever generated, so existing sessions and all
-      positional parsers (``parts[2]`` == platform, etc.) are unaffected.
-    - named profile ``coder`` → ``agent:coder`` — keeps the same positional
-      layout, just a different namespace, so two profiles serving the same
-      platform/chat never collide.
-    """
-    if not profile or profile == "default":
-        return "agent:main"
-    return f"agent:{profile}"
-
-
 def build_session_key(
    source: SessionSource,
    group_sessions_per_user: bool = True,
    thread_sessions_per_user: bool = False,
-    profile: Optional[str] = None,
 ) -> str:
    """Build a deterministic session key from a message source.

    This is the single source of truth for session key construction.

-    ``profile`` selects the key namespace (see :func:`_session_key_namespace`).
-    It defaults to ``None`` ⇒ the legacy ``agent:main`` namespace, so callers
-    that don't multiplex produce byte-identical keys to before. Only the
-    multiplexing gateway passes a non-default profile.
-
    DM rules:
      - DMs include chat_id when present, so each private conversation is isolated.
      - thread_id further differentiates threaded DMs within the same DM chat.
@@ -677,7 +643,6 @@ def build_session_key(
        shared session per chat.
      - Without identifiers, messages fall back to one session per platform/chat_type.
    """
-    ns = _session_key_namespace(profile)
    platform = source.platform.value
    if source.chat_type == "dm":
        dm_chat_id = source.chat_id
@@ -686,12 +651,12 @@ def build_session_key(

        if dm_chat_id:
            if source.thread_id:
-                return f"{ns}:{platform}:dm:{dm_chat_id}:{source.thread_id}"
-            return f"{ns}:{platform}:dm:{dm_chat_id}"
+                return f"agent:main:{platform}:dm:{dm_chat_id}:{source.thread_id}"
+            return f"agent:main:{platform}:dm:{dm_chat_id}"
        # No chat_id — fall back to the sender's own identifier before the
        # bare per-platform sink.  Without this, every DM from every user that
        # arrives without a chat_id (non-standard adapters / synthetic sources)
-        # collapses into one shared "<ns>:<platform>:dm" session, and a
+        # collapses into one shared "agent:main:<platform>:dm" session, and a
        # single cached agent ends up serving multiple people's conversations —
        # cross-user history bleed.  participant_id keeps DMs isolated per user.
        dm_participant_id = source.user_id_alt or source.user_id
@@ -702,11 +667,11 @@ def build_session_key(
            )
        if dm_participant_id:
            if source.thread_id:
-                return f"{ns}:{platform}:dm:{dm_participant_id}:{source.thread_id}"
-            return f"{ns}:{platform}:dm:{dm_participant_id}"
+                return f"agent:main:{platform}:dm:{dm_participant_id}:{source.thread_id}"
+            return f"agent:main:{platform}:dm:{dm_participant_id}"
        if source.thread_id:
-            return f"{ns}:{platform}:dm:{source.thread_id}"
-        return f"{ns}:{platform}:dm"
+            return f"agent:main:{platform}:dm:{source.thread_id}"
+        return f"agent:main:{platform}:dm"

    participant_id = source.user_id_alt or source.user_id
    if participant_id and source.platform == Platform.WHATSAPP:
@@ -714,7 +679,7 @@ def build_session_key(
        # single group member gets two isolated per-user sessions when the
        # bridge reshuffles alias forms.
        participant_id = canonical_whatsapp_identifier(str(participant_id)) or participant_id
-    key_parts = [ns, platform, source.chat_type]
+    key_parts = ["agent:main", platform, source.chat_type]

    if source.chat_id:
        key_parts.append(source.chat_id)
@@ -810,32 +775,12 @@ class SessionStore:
                logger.debug("Could not remove temp file %s: %s", tmp_path, e)
            raise
    
-    def _resolve_profile_for_key(self, source: Optional[SessionSource] = None) -> Optional[str]:
-        """Return the profile namespace for session keys, or None when off.
-
-        When ``multiplex_profiles`` is disabled (default), returns ``None`` so
-        keys stay in the legacy ``agent:main`` namespace — byte-identical to
-        before. When enabled, prefers the profile the inbound source was routed
-        to (``source.profile`` — set by the /p/<profile>/ URL prefix or
-        per-credential adapter), falling back to the active profile name.
-        """
-        if not getattr(self.config, "multiplex_profiles", False):
-            return None
-        if source is not None and source.profile:
-            return source.profile
-        try:
-            from hermes_cli.profiles import get_active_profile_name
-            return get_active_profile_name() or "default"
-        except Exception:
-            return None
-
    def _generate_session_key(self, source: SessionSource) -> str:
        """Generate a session key from a source."""
        return build_session_key(
            source,
            group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True),
            thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False),
-            profile=self._resolve_profile_for_key(source),
        )
    
    def _is_session_expired(self, entry: SessionEntry) -> bool:
@@ -515,7 +515,6 @@ def write_runtime_status(
    platform_state: Any = _UNSET,
    error_code: Any = _UNSET,
    error_message: Any = _UNSET,
-    served_profiles: Any = _UNSET,
 ) -> None:
    """Persist gateway runtime health information for diagnostics/status."""
    path = _get_runtime_status_path()
@@ -536,11 +535,6 @@ def write_runtime_status(
        payload["restart_requested"] = bool(restart_requested)
    if active_agents is not _UNSET:
        payload["active_agents"] = max(0, int(active_agents))
-    if served_profiles is not _UNSET:
-        # Profiles this gateway multiplexes (multi-profile mode). Absent/empty
-        # for a single-profile gateway. Lets `hermes status` show per-profile
-        # coverage without a second probe.
-        payload["served_profiles"] = list(served_profiles or [])

    if platform is not _UNSET:
        platform_payload = payload["platforms"].get(platform, {})
@@ -64,6 +64,39 @@ _EXCLUDED_NAMES = {
    "cron.pid",
 }

+# File names that ``hermes import`` must never overwrite, matched by basename so
+# they're caught for the root profile (``gateway_state.json``) and for named
+# profiles alike (``profiles/<name>/gateway_state.json``).
+#
+# These hold *volatile gateway/process runtime state that is namespaced to the
+# machine or container the backup was taken on* — PIDs in a dead process
+# namespace, a runtime lock, the process registry, and the gateway's last
+# recorded run/desired state. Restoring them onto a different host (or a hosted
+# container) is at best meaningless and at worst actively harmful:
+#
+#   - ``gateway_state.json`` drives the container-boot reconciler
+#     (``container_boot._read_desired_state``), which only auto-starts a
+#     gateway whose recorded state is ``running``. A backup taken from a
+#     machine where the gateway was stopped (or carrying a stale/foreign
+#     value) overwrites the container's own state and leaves the gateway
+#     stuck "starting"/"cooking", disconnecting it from the Nous portal
+#     (NS-508 / the second half of NS-501).
+#   - ``gateway.pid`` / ``cron.pid`` / ``gateway.lock`` / ``processes.json``
+#     reference PIDs and locks in the *source* machine's process namespace; a
+#     numerically-equal PID in the new environment is a different process.
+#     These mirror exactly what ``container_boot._STALE_RUNTIME_FILES`` already
+#     sweeps on every container boot.
+#
+# Older backups predate the backup-side exclusions, so we filter on import too
+# rather than trusting the archive's contents.
+_IMPORT_SKIP_NAMES = {
+    "gateway_state.json",
+    "gateway.pid",
+    "cron.pid",
+    "gateway.lock",
+    "processes.json",
+}
+
 # zipfile.open() drops Unix mode bits on extract; restore tightens these to 0600.
 _SECRET_FILE_NAMES = {".env", "auth.json", "state.db"}

@@ -385,6 +418,7 @@ def run_import(args) -> None:

        errors = []
        restored = 0
+        skipped_runtime: list[str] = []
        t0 = time.monotonic()

        for member in members:
@@ -397,6 +431,16 @@ def run_import(args) -> None:
            if not rel:
                continue

+            # Never overwrite volatile gateway/process runtime state. These are
+            # namespaced to the machine/container the backup was taken on;
+            # clobbering them (especially gateway_state.json) breaks the gateway
+            # reconciler on the target and disconnects hosted instances from the
+            # Nous portal. Matched by basename so both the root profile and
+            # named profiles (profiles/<name>/gateway_state.json) are covered.
+            if Path(rel).name in _IMPORT_SKIP_NAMES:
+                skipped_runtime.append(rel)
+                continue
+
            target = hermes_root / rel

            # Security: reject absolute paths and traversals
@@ -433,6 +477,16 @@ def run_import(args) -> None:
            if len(errors) > 10:
                print(f"  ... and {len(errors) - 10} more")

+        if skipped_runtime:
+            print(
+                f"\n  Preserved {len(skipped_runtime)} runtime state "
+                f"file(s) (kept this machine's, not the backup's):"
+            )
+            for rel in sorted(skipped_runtime)[:10]:
+                print(f"    {rel}")
+            if len(skipped_runtime) > 10:
+                print(f"    ... and {len(skipped_runtime) - 10} more")
+
        # Post-import: restore profile wrapper scripts
        profiles_dir = hermes_root / "profiles"
        restored_profiles = []
@@ -3865,86 +3865,6 @@ def _running_under_gateway_supervisor() -> bool:
    return False


-def _guard_named_profile_under_multiplexer(force: bool = False) -> None:
-    """Refuse a named-profile gateway when a multiplexer is already serving it.
-
-    When the default profile's gateway runs with gateway.multiplex_profiles=on,
-    it is the sole inbound process for EVERY profile on the host. Starting a
-    separate gateway for a named profile would double-bind that profile's
-    platforms (two pollers on one bot token, port fights). In that mode a
-    named-profile ``hermes gateway run`` is always a misconfiguration, so we
-    hard-error with a pointer to the multiplexer. ``--force`` overrides.
-
-    Inert unless ALL of: (a) this invocation is a named profile, (b) a default-
-    profile gateway is running, (c) that gateway's config has multiplexing on.
-    """
-    if force:
-        return
-    # (a) Are we a named profile? Default/custom-hash homes return "".
-    try:
-        suffix = _profile_suffix()
-    except Exception:
-        return
-    if not suffix:
-        return  # default profile (or unrecognized) — this guard doesn't apply
-
-    try:
-        from hermes_constants import get_default_hermes_root
-        default_root = get_default_hermes_root()
-        # (b) Is the default-profile gateway running?
-        from gateway.status import get_running_pid as _default_running_pid  # noqa
-    except Exception:
-        return
-
-    try:
-        import yaml as _yaml
-        from gateway.status import _read_pid_record  # type: ignore
-
-        # (b) default gateway PID file present + alive
-        default_pid_path = default_root / "gateway.pid"
-        rec = _read_pid_record(default_pid_path)
-        if not rec:
-            return
-        from gateway.status import _pid_exists, _pid_from_record
-        pid = _pid_from_record(rec)
-        if not pid or not _pid_exists(pid):
-            return
-
-        # (c) default config has multiplexing on
-        cfg_path = default_root / "config.yaml"
-        if not cfg_path.exists():
-            return
-        with open(cfg_path, encoding="utf-8") as f:
-            cfg = _yaml.safe_load(f) or {}
-        multiplex = bool(
-            cfg.get("multiplex_profiles")
-            or (cfg.get("gateway", {}) or {}).get("multiplex_profiles")
-        )
-        if not multiplex:
-            return
-    except Exception:
-        logger.debug("Multiplexer-conflict probe failed", exc_info=True)
-        return
-
-    print_error(
-        f"The default gateway is running as a profile multiplexer and already "
-        f"serves profile '{suffix}'."
-    )
-    print(
-        "  When gateway.multiplex_profiles is on, the default gateway is the\n"
-        "  single inbound process for every profile. Starting a separate\n"
-        "  gateway for this profile would double-bind its platforms (two\n"
-        "  pollers on one bot token, port conflicts).\n"
-    )
-    print("  Manage the multiplexer instead (from the default profile):")
-    print()
-    print("    hermes gateway restart")
-    print()
-    print("  Pass --force to start a separate profile gateway anyway (not")
-    print("  recommended while the multiplexer is running).")
-    sys.exit(1)
-
-
 def _guard_supervised_gateway_conflict(force: bool = False) -> None:
    """Refuse a foreground gateway when a service manager already supervises one.

@@ -4057,7 +3977,6 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False, fo
               systemd/launchd service is already supervising this profile.
    """
    _guard_official_docker_root_gateway()
-    _guard_named_profile_under_multiplexer(force=force)
    _guard_supervised_gateway_conflict(force=force)
    _guard_existing_gateway_process_conflict(replace=replace)
    sys.path.insert(0, str(PROJECT_ROOT))
@@ -29,7 +29,7 @@ import subprocess
 import sys
 from dataclasses import dataclass
 from pathlib import Path, PurePosixPath, PureWindowsPath
-from typing import List, Optional, Tuple
+from typing import List, Optional

 from agent.skill_utils import is_excluded_skill_path

@@ -781,47 +781,6 @@ def list_profiles() -> List[ProfileInfo]:
    return profiles


-def profiles_to_serve(multiplex: bool) -> List[Tuple[str, Path]]:
-    """Return the ``(profile_name, hermes_home)`` pairs a gateway should serve.
-
-    This is the single chokepoint for "which profiles does the inbound gateway
-    handle" so later multiplexing phases never re-derive the set.
-
-    - ``multiplex=False`` (default): returns exactly one entry for the *active*
-      profile — byte-for-byte the single-profile behavior the gateway has
-      always had. The name is ``"default"`` for the default profile or the
-      active named profile's id.
-    - ``multiplex=True``: returns the default profile plus every valid named
-      profile under ``profiles/``, each paired with its own HERMES_HOME.
-
-    Intentionally lightweight (a directory scan + name validation only): no
-    per-profile config reads, gateway-running probes, or skill counts like
-    :func:`list_profiles`. It runs on gateway startup and must stay cheap.
-
-    The returned ``hermes_home`` is the path to pass to
-    ``set_hermes_home_override`` when scoping a turn to that profile.
-    """
-    active = get_active_profile_name() or "default"
-    if not multiplex:
-        return [(active, get_profile_dir(active))]
-
-    serve: List[Tuple[str, Path]] = [("default", _get_default_hermes_home())]
-
-    profiles_root = _get_profiles_root()
-    if profiles_root.is_dir():
-        for entry in sorted(profiles_root.iterdir()):
-            if not entry.is_dir():
-                continue
-            name = entry.name
-            if name == "default":
-                continue  # default is the built-in entry already added above
-            if not _PROFILE_ID_RE.match(name):
-                continue
-            serve.append((name, entry))
-
-    return serve
-
-
 def create_profile(
    name: str,
    clone_from: Optional[str] = None,
@@ -12,7 +12,6 @@ logger = logging.getLogger(__name__)

 from hermes_cli import auth as auth_mod
 from agent.credential_pool import CredentialPool, PooledCredential, get_custom_provider_pool_key, load_pool
-from agent.secret_scope import get_secret as _get_secret
 from hermes_cli.auth import (
    AuthError,
    DEFAULT_CODEX_BASE_URL,
@@ -36,19 +35,6 @@ from hermes_constants import OPENROUTER_BASE_URL
 from utils import base_url_host_matches, base_url_hostname, env_int


-def _getenv(name: str, default: str = "") -> str:
-    """Profile-scoped replacement for ``os.getenv`` on credential/provider reads.
-
-    Routes through the secret scope (Workstream A): identical to ``os.getenv``
-    when multiplexing is off, scope-aware (and fail-closed on an unscoped read)
-    when on. Genuinely-global vars are handled inside ``get_secret`` and still
-    read ``os.environ``. Keeps the ``(name, default) -> str`` contract every
-    call site here already relies on.
-    """
-    val = _get_secret(name, default)
-    return val if val is not None else default
-
-
 def _normalize_custom_provider_name(value: str) -> str:
    return value.strip().lower().replace(" ", "-")

@@ -170,7 +156,7 @@ def _host_derived_api_key(base_url: str) -> str:
    if sanitized in ("OPENAI", "OPENROUTER", "OLLAMA"):
        return ""
    env_name = f"{sanitized}_API_KEY"
-    return (_getenv(env_name, "") or "").strip()
+    return (os.getenv(env_name, "") or "").strip()


 def _auto_detect_local_model(base_url: str) -> str:
@@ -451,7 +437,7 @@ def resolve_requested_provider(requested: Optional[str] = None) -> str:

    # Prefer the persisted config selection over any stale shell/.env
    # provider override so chat uses the endpoint the user last saved.
-    env_provider = _getenv("HERMES_INFERENCE_PROVIDER", "").strip().lower()
+    env_provider = os.getenv("HERMES_INFERENCE_PROVIDER", "").strip().lower()
    if env_provider:
        return env_provider

@@ -556,7 +542,7 @@ def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, An
            name_norm = _normalize_custom_provider_name(ep_name)
            # Resolve the API key from the env var name stored in key_env
            key_env = str(entry.get("key_env", "") or "").strip()
-            resolved_api_key = _getenv(key_env, "").strip() if key_env else ""
+            resolved_api_key = os.getenv(key_env, "").strip() if key_env else ""
            # Fall back to inline api_key when key_env is absent or unresolvable
            if not resolved_api_key:
                resolved_api_key = str(entry.get("api_key", "") or "").strip()
@@ -775,8 +761,8 @@ def _resolve_named_custom_runtime(
        api_key_candidates = [
            (explicit_api_key or "").strip(),
            # Gate env key fallbacks on authoritative hosts (#28660)
-            (_getenv("OPENAI_API_KEY", "").strip()     if _da_is_openai_url else ""),
-            (_getenv("OPENROUTER_API_KEY", "").strip() if _da_is_openrouter  else ""),
+            (os.getenv("OPENAI_API_KEY", "").strip()     if _da_is_openai_url else ""),
+            (os.getenv("OPENROUTER_API_KEY", "").strip() if _da_is_openrouter  else ""),
            # Bonus (#28660): derive `<VENDOR>_API_KEY` from the host so users
            # who set DEEPSEEK_API_KEY / GROQ_API_KEY / MISTRAL_API_KEY get the
            # intuitive match without configuring `custom_providers` first.
@@ -829,11 +815,11 @@ def _resolve_named_custom_runtime(
    api_key_candidates = [
        (explicit_api_key or "").strip(),
        str(custom_provider.get("api_key", "") or "").strip(),
-        _getenv(str(custom_provider.get("key_env", "") or "").strip(), "").strip(),
+        os.getenv(str(custom_provider.get("key_env", "") or "").strip(), "").strip(),
        # Gate provider env keys on their authoritative hosts — sending
        # OPENAI_API_KEY to a local-llm endpoint leaks credentials (#28660).
-        (_getenv("OPENAI_API_KEY", "").strip()     if _cp_is_openai_url  else ""),
-        (_getenv("OPENROUTER_API_KEY", "").strip() if _cp_is_openrouter  else ""),
+        (os.getenv("OPENAI_API_KEY", "").strip()     if _cp_is_openai_url  else ""),
+        (os.getenv("OPENROUTER_API_KEY", "").strip() if _cp_is_openrouter  else ""),
        # Bonus (#28660): derive `<VENDOR>_API_KEY` from the host as a final
        # fallback when key_env wasn't set explicitly.
        _host_derived_api_key(base_url),
@@ -892,8 +878,8 @@ def _resolve_openrouter_runtime(
        except Exception:
            pass

-    env_openrouter_base_url = _getenv("OPENROUTER_BASE_URL", "").strip()
-    env_custom_base_url = _getenv("CUSTOM_BASE_URL", "").strip()
+    env_openrouter_base_url = os.getenv("OPENROUTER_BASE_URL", "").strip()
+    env_custom_base_url = os.getenv("CUSTOM_BASE_URL", "").strip()

    # Use config base_url when available and the provider context matches.
    # OPENAI_BASE_URL env var is no longer consulted — config.yaml is
@@ -933,8 +919,8 @@ def _resolve_openrouter_runtime(
    if _is_openrouter_context:
        api_key_candidates = [
            explicit_api_key,
-            _getenv("OPENROUTER_API_KEY"),
-            _getenv("OPENAI_API_KEY"),
+            os.getenv("OPENROUTER_API_KEY"),
+            os.getenv("OPENAI_API_KEY"),
        ]
    else:
        # Custom endpoint: use api_key from config when using config base_url (#1760).
@@ -954,9 +940,9 @@ def _resolve_openrouter_runtime(
        api_key_candidates = [
            explicit_api_key,
            (cfg_api_key if use_config_base_url else ""),
-            (_getenv("OLLAMA_API_KEY")     if _is_ollama_url                       else ""),
-            (_getenv("OPENAI_API_KEY")     if (_is_openai_url or _is_openai_azure) else ""),
-            (_getenv("OPENROUTER_API_KEY") if _is_openrouter_url                   else ""),
+            (os.getenv("OLLAMA_API_KEY")     if _is_ollama_url                       else ""),
+            (os.getenv("OPENAI_API_KEY")     if (_is_openai_url or _is_openai_azure) else ""),
+            (os.getenv("OPENROUTER_API_KEY") if _is_openrouter_url                   else ""),
            # Bonus (#28660): derive `<VENDOR>_API_KEY` from the host so users
            # who set DEEPSEEK_API_KEY / GROQ_API_KEY / MISTRAL_API_KEY get the
            # intuitive match. Helper returns "" for IPs/loopback and for env
@@ -1059,7 +1045,7 @@ def _resolve_azure_foundry_runtime(
        if inferred:
            cfg_api_mode = inferred

-    env_base_url = _getenv("AZURE_FOUNDRY_BASE_URL", "").strip().rstrip("/")
+    env_base_url = os.getenv("AZURE_FOUNDRY_BASE_URL", "").strip().rstrip("/")
    base_url = explicit_base_url_clean or cfg_base_url or env_base_url
    if not base_url:
        raise AuthError(
@@ -1148,7 +1134,7 @@ def _resolve_azure_foundry_runtime(
        except Exception:
            api_key = ""
    if not api_key:
-        api_key = _getenv("AZURE_FOUNDRY_API_KEY", "").strip()
+        api_key = os.getenv("AZURE_FOUNDRY_API_KEY", "").strip()
    if not api_key:
        raise AuthError(
            "Azure Foundry requires an API key. Set AZURE_FOUNDRY_API_KEY in "
@@ -1248,7 +1234,7 @@ def _resolve_explicit_runtime(
        expires_at = state.get("agent_key_expires_at") or state.get("expires_at")
        if not api_key:
            creds = resolve_nous_runtime_credentials(
-                timeout_seconds=float(_getenv("HERMES_NOUS_TIMEOUT_SECONDS", "15")),
+                timeout_seconds=float(os.getenv("HERMES_NOUS_TIMEOUT_SECONDS", "15")),
            )
            api_key = creds.get("api_key", "")
            expires_at = creds.get("expires_at")
@@ -1277,7 +1263,7 @@ def _resolve_explicit_runtime(
    if pconfig and pconfig.auth_type == "api_key":
        env_url = ""
        if pconfig.base_url_env_var:
-            env_url = _getenv(pconfig.base_url_env_var, "").strip().rstrip("/")
+            env_url = os.getenv(pconfig.base_url_env_var, "").strip().rstrip("/")

        base_url = explicit_base_url
        if not base_url:
@@ -1349,8 +1335,8 @@ def resolve_runtime_provider(
    if requested_provider == "anthropic" and "azure.com" in _eff_base:
        _azure_key = (
            (explicit_api_key or "").strip()
-            or _getenv("AZURE_ANTHROPIC_KEY", "").strip()
-            or _getenv("ANTHROPIC_API_KEY", "").strip()
+            or os.getenv("AZURE_ANTHROPIC_KEY", "").strip()
+            or os.getenv("ANTHROPIC_API_KEY", "").strip()
        )
        return {
            "provider": "anthropic",
@@ -1405,8 +1391,8 @@ def resolve_runtime_provider(
    if provider == "openrouter":
        cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
        cfg_base_url = str(model_cfg.get("base_url") or "").strip()
-        env_openai_base_url = _getenv("OPENAI_BASE_URL", "").strip()
-        env_openrouter_base_url = _getenv("OPENROUTER_BASE_URL", "").strip()
+        env_openai_base_url = os.getenv("OPENAI_BASE_URL", "").strip()
+        env_openrouter_base_url = os.getenv("OPENROUTER_BASE_URL", "").strip()
        has_custom_endpoint = bool(
            explicit_base_url
            or env_openai_base_url
@@ -1462,7 +1448,7 @@ def resolve_runtime_provider(
    if provider == "nous":
        try:
            creds = resolve_nous_runtime_credentials(
-                timeout_seconds=float(_getenv("HERMES_NOUS_TIMEOUT_SECONDS", "15")),
+                timeout_seconds=float(os.getenv("HERMES_NOUS_TIMEOUT_SECONDS", "15")),
            )
            return {
                "provider": "nous",
@@ -1615,7 +1601,7 @@ def resolve_runtime_provider(
            for hint_key in ("key_env", "api_key_env"):
                env_var = str(model_cfg.get(hint_key) or "").strip()
                if env_var:
-                    token = _getenv(env_var, "").strip()
+                    token = os.getenv(env_var, "").strip()
                    if token:
                        break
            # Next: an inline api_key on the model config (useful in multi-profile
@@ -1625,8 +1611,8 @@ def resolve_runtime_provider(
            # Finally fall back to the historical fixed names.
            if not token:
                token = (
-                    _getenv("AZURE_ANTHROPIC_KEY", "").strip()
-                    or _getenv("ANTHROPIC_API_KEY", "").strip()
+                    os.getenv("AZURE_ANTHROPIC_KEY", "").strip()
+                    or os.getenv("ANTHROPIC_API_KEY", "").strip()
                )
            if not token:
                raise AuthError(
@@ -70,7 +70,10 @@ from gateway.status import (
 from utils import env_var_enabled

 try:
-    from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
+    from fastapi import (
+        FastAPI, File, Form, HTTPException, Request, UploadFile,
+        WebSocket, WebSocketDisconnect,
+    )
    from fastapi.middleware.cors import CORSMiddleware
    from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Response
    from fastapi.staticfiles import StaticFiles
@@ -82,7 +85,10 @@ except ImportError:
    try:
        from tools.lazy_deps import ensure as _lazy_ensure
        _lazy_ensure("tool.dashboard", prompt=False)
-        from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
+        from fastapi import (
+            FastAPI, File, Form, HTTPException, Request, UploadFile,
+            WebSocket, WebSocketDisconnect,
+        )
        from fastapi.middleware.cors import CORSMiddleware
        from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Response
        from fastapi.staticfiles import StaticFiles
@@ -1486,6 +1492,74 @@ async def upload_managed_file(payload: ManagedFileUpload, request: Request):
    }


+# Stream uploads to disk in fixed-size chunks. The legacy JSON endpoint above
+# buffers the whole file as a base64 data URL in a JSON body, which (a) inflates
+# the payload ~33%, (b) holds the entire file (plus its decoded copy) in memory,
+# and (c) reliably trips upstream proxy body-size/timeout limits with a 502 on
+# large backup archives (NS-501). This multipart endpoint reads the request body
+# in 1 MiB chunks straight to a temp file, enforces the size cap as it goes, and
+# atomically renames into place — constant memory, no base64 inflation.
+_UPLOAD_CHUNK_BYTES = 1024 * 1024
+
+
+@app.post("/api/files/upload-stream")
+async def upload_managed_file_stream(
+    request: Request,
+    file: UploadFile = File(...),
+    path: str = Form(...),
+    overwrite: bool = Form(True),
+):
+    policy, target, display_path = _resolve_managed_path(path, request, for_write=True)
+    if target.exists() and target.is_dir():
+        raise HTTPException(status_code=409, detail="A directory already exists at that path")
+    if target.exists() and not overwrite:
+        raise HTTPException(status_code=409, detail="File already exists")
+
+    try:
+        target.parent.mkdir(parents=True, exist_ok=True)
+    except PermissionError:
+        raise HTTPException(status_code=403, detail="File is not writable")
+    except OSError as exc:
+        raise HTTPException(status_code=500, detail=f"Could not create parent directory: {exc}")
+
+    # Write to a sibling temp file first so a partial/aborted upload never
+    # clobbers an existing file, then atomically rename into place.
+    tmp_fd, tmp_name = tempfile.mkstemp(
+        prefix=f".{target.name}.", suffix=".upload", dir=str(target.parent)
+    )
+    tmp_path = Path(tmp_name)
+    total = 0
+    try:
+        with os.fdopen(tmp_fd, "wb") as out:
+            while True:
+                chunk = await file.read(_UPLOAD_CHUNK_BYTES)
+                if not chunk:
+                    break
+                total += len(chunk)
+                if total > _MANAGED_FILE_MAX_BYTES:
+                    raise HTTPException(status_code=413, detail="File is too large")
+                out.write(chunk)
+        os.replace(tmp_path, target)
+    except HTTPException:
+        tmp_path.unlink(missing_ok=True)
+        raise
+    except PermissionError:
+        tmp_path.unlink(missing_ok=True)
+        raise HTTPException(status_code=403, detail="File is not writable")
+    except OSError as exc:
+        tmp_path.unlink(missing_ok=True)
+        raise HTTPException(status_code=500, detail=f"Could not write file: {exc}")
+    finally:
+        await file.close()
+
+    return {
+        "ok": True,
+        "entry": _managed_file_entry(policy, target),
+        "path": display_path,
+        **_managed_response_meta(policy),
+    }
+
+
@app.post("/api/files/mkdir")
 async def create_managed_directory(payload: ManagedDirectoryCreate, request: Request):
    policy, target, display_path = _resolve_managed_path(payload.path, request, for_write=True)
@@ -106,6 +106,11 @@ dependencies = [
  "pathspec==1.1.1",
  "fastapi>=0.104.0,<1",
  "uvicorn[standard]>=0.24.0,<1",
+  # Streaming multipart uploads for the dashboard file manager (NS-501).
+  # FastAPI's UploadFile/Form depend on python-multipart; it is NOT pulled in
+  # by fastapi itself, so the dashboard's multipart upload endpoint would 500
+  # without an explicit dependency here (and in the `web` extra below).
+  "python-multipart<1,>=0.0.9",
  "ptyprocess>=0.7.0,<1; sys_platform != 'win32'",
  "pywinpty>=2.0.0,<3; sys_platform == 'win32'",
  # Image resize recovery for the vision tools. Pillow shrinks oversized images
@@ -253,7 +258,7 @@ youtube = [
 # `hermes dashboard` (localhost SPA + API).  Not in core to keep the default install lean.
 # starlette==1.0.1 pinned for CVE-2026-48710 (BadHost) — fastapi pulls Starlette
 # transitively and pre-1.0.1 is the vulnerable range. See the mcp extra above.
-web = ["fastapi==0.133.1", "uvicorn[standard]==0.41.0", "starlette==1.0.1"]
+web = ["fastapi==0.133.1", "uvicorn[standard]==0.41.0", "starlette==1.0.1", "python-multipart==0.0.31"]
 all = [
  # Policy (2026-05-12): `[all]` includes only extras that genuinely
  # CAN'T be lazy-installed via `tools/lazy_deps.py` — i.e. things every
@@ -1,130 +0,0 @@
-"""Tests for the profile-scoped credential primitive (Workstream A / Phase 2)."""
-import pytest
-
-from agent import secret_scope as ss
-
-
-@pytest.fixture(autouse=True)
-def _reset_multiplex():
-    """Ensure each test starts and ends with multiplexing off (it's a global)."""
-    ss.set_multiplex_active(False)
-    yield
-    ss.set_multiplex_active(False)
-
-
-class TestMultiplexInactiveBackwardCompat:
-    """Default deployment: get_secret transparently reads os.environ."""
-
-    def test_reads_environ(self, monkeypatch):
-        monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test")
-        assert ss.get_secret("ANTHROPIC_API_KEY") == "sk-test"
-
-    def test_missing_returns_default(self, monkeypatch):
-        monkeypatch.delenv("NOPE_KEY", raising=False)
-        assert ss.get_secret("NOPE_KEY") is None
-        assert ss.get_secret("NOPE_KEY", "fallback") == "fallback"
-
-    def test_no_raise_without_scope(self, monkeypatch):
-        monkeypatch.delenv("SOME_KEY", raising=False)
-        # multiplex off => unscoped read is fine, returns default
-        assert ss.get_secret("SOME_KEY") is None
-
-
-class TestMultiplexActiveFailClosed:
-    """Multiplex on: an unscoped secret read raises instead of leaking."""
-
-    def test_unscoped_read_raises(self, monkeypatch):
-        monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-leaky")
-        ss.set_multiplex_active(True)
-        with pytest.raises(ss.UnscopedSecretError):
-            ss.get_secret("ANTHROPIC_API_KEY")
-
-    def test_scoped_read_uses_scope_not_environ(self, monkeypatch):
-        monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-from-environ")
-        ss.set_multiplex_active(True)
-        token = ss.set_secret_scope({"ANTHROPIC_API_KEY": "sk-from-scope"})
-        try:
-            assert ss.get_secret("ANTHROPIC_API_KEY") == "sk-from-scope"
-        finally:
-            ss.reset_secret_scope(token)
-
-    def test_scoped_missing_key_returns_default_not_environ(self, monkeypatch):
-        # Even though the value exists in os.environ, a scope is authoritative:
-        # an absent scope key must NOT fall through to the (cross-profile) env.
-        monkeypatch.setenv("OPENAI_API_KEY", "sk-other-profile")
-        ss.set_multiplex_active(True)
-        token = ss.set_secret_scope({"ANTHROPIC_API_KEY": "sk-mine"})
-        try:
-            assert ss.get_secret("OPENAI_API_KEY") is None
-            assert ss.get_secret("OPENAI_API_KEY", "d") == "d"
-        finally:
-            ss.reset_secret_scope(token)
-
-    def test_global_env_still_reads_environ_under_multiplex(self, monkeypatch):
-        monkeypatch.setenv("HERMES_HOME", "/opt/data")
-        ss.set_multiplex_active(True)
-        # No scope, multiplex on — but HERMES_HOME is global, so no raise.
-        assert ss.get_secret("HERMES_HOME") == "/opt/data"
-
-    def test_kanban_prefix_is_global(self, monkeypatch):
-        monkeypatch.setenv("HERMES_KANBAN_DB", "/x/kanban.db")
-        ss.set_multiplex_active(True)
-        assert ss.get_secret("HERMES_KANBAN_DB") == "/x/kanban.db"
-
-
-class TestScopeIsolation:
-    """Two scopes never see each other's secrets."""
-
-    def test_nested_scopes_restore(self):
-        ss.set_multiplex_active(True)
-        t1 = ss.set_secret_scope({"K": "a"})
-        try:
-            assert ss.get_secret("K") == "a"
-            t2 = ss.set_secret_scope({"K": "b"})
-            try:
-                assert ss.get_secret("K") == "b"
-            finally:
-                ss.reset_secret_scope(t2)
-            assert ss.get_secret("K") == "a"
-        finally:
-            ss.reset_secret_scope(t1)
-
-
-class TestEnvFileParsing:
-    """load_env_file parses without mutating os.environ."""
-
-    def test_parses_basic(self, tmp_path):
-        env = tmp_path / ".env"
-        env.write_text(
-            "# comment\n"
-            "ANTHROPIC_API_KEY=sk-abc\n"
-            "export OPENAI_API_KEY=sk-def\n"
-            'QUOTED="quoted-value"\n'
-            "SINGLE='single'\n"
-            "\n"
-            "BAD_LINE_NO_EQUALS\n"
-        )
-        out = ss.load_env_file(env)
-        assert out == {
-            "ANTHROPIC_API_KEY": "sk-abc",
-            "OPENAI_API_KEY": "sk-def",
-            "QUOTED": "quoted-value",
-            "SINGLE": "single",
-        }
-
-    def test_does_not_mutate_environ(self, tmp_path, monkeypatch):
-        monkeypatch.delenv("ZZZ_KEY", raising=False)
-        env = tmp_path / ".env"
-        env.write_text("ZZZ_KEY=secret\n")
-        ss.load_env_file(env)
-        import os
-        assert "ZZZ_KEY" not in os.environ
-
-    def test_missing_file_returns_empty(self, tmp_path):
-        assert ss.load_env_file(tmp_path / "nope.env") == {}
-
-    def test_build_profile_secret_scope(self, tmp_path):
-        (tmp_path / ".env").write_text("ANTHROPIC_API_KEY=sk-profile\n")
-        assert ss.build_profile_secret_scope(tmp_path) == {
-            "ANTHROPIC_API_KEY": "sk-profile"
-        }
@@ -0,0 +1,166 @@
+"""Unit tests for managed-boot relay self-provisioning.
+
+Covers gateway.relay.self_provision_if_managed() + the relay_endpoint() /
+relay_route_keys() config readers. The connector HTTP POST is monkeypatched
+(the cross-repo E2E exercises the real /relay/provision); these prove the
+TRIGGER logic, in-process env wiring, and fail-soft boot behaviour.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+import gateway.relay as relay
+
+
+@pytest.fixture(autouse=True)
+def _clean_env(monkeypatch):
+    for k in (
+        "GATEWAY_RELAY_URL",
+        "GATEWAY_RELAY_ID",
+        "GATEWAY_RELAY_SECRET",
+        "GATEWAY_RELAY_DELIVERY_KEY",
+        "GATEWAY_RELAY_ENDPOINT",
+        "GATEWAY_RELAY_ROUTE_KEYS",
+        "GATEWAY_RELAY_PLATFORM",
+        "GATEWAY_RELAY_BOT_ID",
+    ):
+        monkeypatch.delenv(k, raising=False)
+    # Never read config.yaml off disk in these tests.
+    monkeypatch.setattr("gateway.run._load_gateway_config", lambda: {}, raising=False)
+
+
+def _stub_post(captured: dict):
+    """A fake _post_provision that records its kwargs and returns creds."""
+
+    def _fake(**kwargs):
+        captured.update(kwargs)
+        return {
+            "secret": "a" * 64,
+            "deliveryKey": "b" * 64,
+            "tenant": "org-tenant-x",
+            "gatewayId": kwargs["gateway_id"],
+            "routeKeys": kwargs["route_keys"],
+        }
+
+    return _fake
+
+
+def _arm(monkeypatch, *, managed=True, url="wss://connector.example/relay", token="nas-token"):
+    monkeypatch.setattr("hermes_cli.config.is_managed", lambda: managed)
+    monkeypatch.setattr(relay, "relay_url", lambda: url)
+    monkeypatch.setattr("hermes_cli.auth.resolve_nous_access_token", lambda: token)
+
+
+# ─────────────────────────── config readers ───────────────────────────
+
+def test_relay_endpoint_from_env(monkeypatch):
+    monkeypatch.setenv("GATEWAY_RELAY_ENDPOINT", "https://gw.example.com/inbound/")
+    assert relay.relay_endpoint() == "https://gw.example.com/inbound"
+
+
+def test_relay_endpoint_absent_is_none():
+    assert relay.relay_endpoint() is None
+
+
+def test_relay_route_keys_csv(monkeypatch):
+    monkeypatch.setenv("GATEWAY_RELAY_ROUTE_KEYS", "guild-1, guild-2 ,, guild-3")
+    assert relay.relay_route_keys() == ["guild-1", "guild-2", "guild-3"]
+
+
+def test_relay_route_keys_empty():
+    assert relay.relay_route_keys() == []
+
+
+def test_provision_url_maps_ws_to_http():
+    assert relay._provision_url("wss://c.example/relay") == "https://c.example/relay/provision"
+    assert relay._provision_url("ws://c.example/relay") == "http://c.example/relay/provision"
+    assert relay._provision_url("https://c.example") == "https://c.example/relay/provision"
+
+
+# ─────────────────────────── trigger logic ───────────────────────────
+
+def test_skips_when_not_managed(monkeypatch):
+    _arm(monkeypatch, managed=False)
+    called = {"n": 0}
+    monkeypatch.setattr(relay, "_post_provision", lambda **k: called.__setitem__("n", called["n"] + 1) or {})
+    assert relay.self_provision_if_managed() is False
+    assert called["n"] == 0
+
+
+def test_skips_when_relay_not_configured(monkeypatch):
+    _arm(monkeypatch, url=None)
+    called = {"n": 0}
+    monkeypatch.setattr(relay, "_post_provision", lambda **k: called.__setitem__("n", called["n"] + 1) or {})
+    assert relay.self_provision_if_managed() is False
+    assert called["n"] == 0
+
+
+def test_skips_when_secret_already_pinned(monkeypatch):
+    _arm(monkeypatch)
+    monkeypatch.setenv("GATEWAY_RELAY_ID", "gw-pinned")
+    monkeypatch.setenv("GATEWAY_RELAY_SECRET", "deadbeef")
+    called = {"n": 0}
+    monkeypatch.setattr(relay, "_post_provision", lambda **k: called.__setitem__("n", called["n"] + 1) or {})
+    assert relay.self_provision_if_managed() is False
+    assert called["n"] == 0
+    # The pinned secret is untouched.
+    assert relay.relay_connection_auth() == ("gw-pinned", "deadbeef")
+
+
+# ─────────────────────────── happy path ───────────────────────────
+
+def test_provisions_and_sets_env_in_process(monkeypatch):
+    _arm(monkeypatch)
+    monkeypatch.setenv("GATEWAY_RELAY_ENDPOINT", "https://gw.example.com/inbound")
+    monkeypatch.setenv("GATEWAY_RELAY_ROUTE_KEYS", "guild-1,guild-2")
+    captured: dict = {}
+    monkeypatch.setattr(relay, "_post_provision", _stub_post(captured))
+
+    assert relay.self_provision_if_managed() is True
+    # The connector POST carried the gateway-asserted endpoint + route keys.
+    assert captured["provision_url"] == "https://connector.example/relay/provision"
+    assert captured["access_token"] == "nas-token"
+    assert captured["gateway_endpoint"] == "https://gw.example.com/inbound"
+    assert captured["route_keys"] == ["guild-1", "guild-2"]
+    # Creds landed in os.environ (in-process), so register_relay_adapter() reads them.
+    gid, secret = relay.relay_connection_auth()
+    assert gid and secret == "a" * 64
+    key, _host, _port = relay.relay_inbound_config()
+    assert key == "b" * 64
+
+
+def test_outbound_only_when_no_endpoint(monkeypatch):
+    _arm(monkeypatch)
+    captured: dict = {}
+    monkeypatch.setattr(relay, "_post_provision", _stub_post(captured))
+
+    assert relay.self_provision_if_managed() is True
+    assert captured["gateway_endpoint"] is None
+    assert captured["route_keys"] == []
+    assert relay.relay_connection_auth()[1] == "a" * 64
+
+
+# ─────────────────────────── fail-soft ───────────────────────────
+
+def test_token_failure_is_non_fatal(monkeypatch):
+    _arm(monkeypatch)
+
+    def _boom():
+        raise RuntimeError("no token")
+
+    monkeypatch.setattr("hermes_cli.auth.resolve_nous_access_token", _boom)
+    # Must not raise; returns False; no creds set.
+    assert relay.self_provision_if_managed() is False
+    assert relay.relay_connection_auth() == (None, None)
+
+
+def test_connector_failure_is_non_fatal(monkeypatch):
+    _arm(monkeypatch)
+
+    def _boom(**kwargs):
+        raise RuntimeError("connector returned HTTP 503")
+
+    monkeypatch.setattr(relay, "_post_provision", _boom)
+    assert relay.self_provision_if_managed() is False
+    assert relay.relay_connection_auth() == (None, None)
@@ -1,78 +0,0 @@
-"""Phase 3: secondary-profile adapter registry + same-token conflict detection."""
-import pytest
-
-from gateway.run import GatewayRunner
-
-
-class _FakeAdapter:
-    def __init__(self, token=None):
-        self.token = token
-
-
-class TestCredentialFingerprint:
-    def test_none_without_token(self):
-        assert GatewayRunner._adapter_credential_fingerprint(_FakeAdapter()) is None
-
-    def test_stable_and_log_safe(self):
-        a = _FakeAdapter(token="secret-bot-token")
-        fp1 = GatewayRunner._adapter_credential_fingerprint(a)
-        fp2 = GatewayRunner._adapter_credential_fingerprint(_FakeAdapter(token="secret-bot-token"))
-        assert fp1 == fp2  # stable
-        assert "secret-bot-token" not in (fp1 or "")  # never the raw token
-        assert len(fp1) == 16
-
-    def test_distinct_tokens_distinct_fp(self):
-        a = GatewayRunner._adapter_credential_fingerprint(_FakeAdapter(token="tok-A"))
-        b = GatewayRunner._adapter_credential_fingerprint(_FakeAdapter(token="tok-B"))
-        assert a != b
-
-    def test_reads_alt_attrs(self):
-        class _AltAdapter:
-            def __init__(self):
-                self.bot_token = "alt-token"
-        assert GatewayRunner._adapter_credential_fingerprint(_AltAdapter()) is not None
-
-
-class TestProfileMessageHandler:
-    @pytest.mark.asyncio
-    async def test_stamps_profile_on_unstamped_source(self):
-        runner = GatewayRunner.__new__(GatewayRunner)
-        seen = {}
-
-        async def _fake_handle(event):
-            seen["profile"] = event.source.profile
-            return "ok"
-
-        runner._handle_message = _fake_handle
-        handler = runner._make_profile_message_handler("coder")
-
-        class _Src:
-            profile = None
-
-        class _Evt:
-            source = _Src()
-
-        result = await handler(_Evt())
-        assert result == "ok"
-        assert seen["profile"] == "coder"
-
-    @pytest.mark.asyncio
-    async def test_does_not_override_existing_profile(self):
-        runner = GatewayRunner.__new__(GatewayRunner)
-        seen = {}
-
-        async def _fake_handle(event):
-            seen["profile"] = event.source.profile
-            return "ok"
-
-        runner._handle_message = _fake_handle
-        handler = runner._make_profile_message_handler("coder")
-
-        class _Src:
-            profile = "writer"  # already stamped (e.g. by URL prefix)
-
-        class _Evt:
-            source = _Src()
-
-        await handler(_Evt())
-        assert seen["profile"] == "writer"
@@ -1,88 +0,0 @@
-"""End-to-end credential isolation proof for multiplex mode (Workstream A).
-
-These exercise the REAL resolution path (runtime_provider, secret scope, MCP
-interpolation) rather than mocking it, proving the property that matters: two
-profiles with different keys never see each other's, and an unscoped read in
-multiplex mode fails closed instead of leaking.
-"""
-import pytest
-
-from agent import secret_scope as ss
-
-
-@pytest.fixture(autouse=True)
-def _reset(monkeypatch):
-    ss.set_multiplex_active(False)
-    yield
-    ss.set_multiplex_active(False)
-
-
-class TestRuntimeProviderUsesScope:
-    """hermes_cli.runtime_provider._getenv resolves through the secret scope."""
-
-    def test_getenv_reads_scope_under_multiplex(self, monkeypatch):
-        from hermes_cli.runtime_provider import _getenv
-        monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-global-leak")
-        ss.set_multiplex_active(True)
-        tok = ss.set_secret_scope({"ANTHROPIC_API_KEY": "sk-profileA"})
-        try:
-            assert _getenv("ANTHROPIC_API_KEY") == "sk-profileA"
-        finally:
-            ss.reset_secret_scope(tok)
-
-    def test_getenv_two_profiles_isolated(self, monkeypatch):
-        from hermes_cli.runtime_provider import _getenv
-        ss.set_multiplex_active(True)
-
-        tok_a = ss.set_secret_scope({"OPENAI_API_KEY": "sk-A"})
-        try:
-            assert _getenv("OPENAI_API_KEY") == "sk-A"
-        finally:
-            ss.reset_secret_scope(tok_a)
-
-        tok_b = ss.set_secret_scope({"OPENAI_API_KEY": "sk-B"})
-        try:
-            assert _getenv("OPENAI_API_KEY") == "sk-B"
-        finally:
-            ss.reset_secret_scope(tok_b)
-
-    def test_getenv_fails_closed_unscoped(self, monkeypatch):
-        from hermes_cli.runtime_provider import _getenv
-        monkeypatch.setenv("OPENROUTER_API_KEY", "sk-leak")
-        ss.set_multiplex_active(True)
-        with pytest.raises(ss.UnscopedSecretError):
-            _getenv("OPENROUTER_API_KEY")
-
-    def test_getenv_global_var_still_reads_environ(self, monkeypatch):
-        from hermes_cli.runtime_provider import _getenv
-        monkeypatch.setenv("HERMES_MAX_ITERATIONS", "42")
-        ss.set_multiplex_active(True)
-        # global var: no scope needed, no raise
-        assert _getenv("HERMES_MAX_ITERATIONS") == "42"
-
-
-class TestMcpInterpolationUsesScope:
-    """MCP config ${VAR} interpolation resolves through the secret scope."""
-
-    def test_interpolation_reads_scope(self, monkeypatch):
-        from tools.mcp_tool import _interpolate_env_vars
-        monkeypatch.setenv("MY_MCP_TOKEN", "global-token")
-        ss.set_multiplex_active(True)
-        tok = ss.set_secret_scope({"MY_MCP_TOKEN": "profile-token"})
-        try:
-            cfg = {"env": {"TOKEN": "${MY_MCP_TOKEN}"}}
-            assert _interpolate_env_vars(cfg) == {"env": {"TOKEN": "profile-token"}}
-        finally:
-            ss.reset_secret_scope(tok)
-
-    def test_interpolation_unset_keeps_placeholder(self, monkeypatch):
-        from tools.mcp_tool import _interpolate_env_vars
-        monkeypatch.delenv("UNSET_MCP_VAR", raising=False)
-        # multiplex off: unset var keeps literal placeholder (legacy behavior)
-        assert _interpolate_env_vars("${UNSET_MCP_VAR}") == "${UNSET_MCP_VAR}"
-
-    def test_interpolation_off_reads_environ(self, monkeypatch):
-        from tools.mcp_tool import _interpolate_env_vars
-        monkeypatch.setenv("MY_MCP_TOKEN", "env-token")
-        # multiplex off: legacy os.environ resolution
-        assert _interpolate_env_vars("${MY_MCP_TOKEN}") == "env-token"
@@ -1,73 +0,0 @@
-"""Phase 1: HTTP-inbound /p/<profile>/ routing for the webhook adapter."""
-import pytest
-
-from gateway.config import GatewayConfig, Platform
-from gateway.session import SessionSource, build_session_key
-
-
-class TestSessionSourceProfileField:
-    def test_profile_roundtrips(self):
-        s = SessionSource(
-            platform=Platform.WEBHOOK if hasattr(Platform, "WEBHOOK") else Platform.TELEGRAM,
-            chat_id="c1",
-            chat_type="webhook",
-            profile="coder",
-        )
-        restored = SessionSource.from_dict(s.to_dict())
-        assert restored.profile == "coder"
-
-    def test_profile_absent_not_serialized(self):
-        s = SessionSource(platform=Platform.TELEGRAM, chat_id="c1", chat_type="dm")
-        assert "profile" not in s.to_dict()
-
-    def test_source_profile_drives_session_key_namespace(self):
-        s = SessionSource(platform=Platform.TELEGRAM, chat_id="99", chat_type="dm")
-        # build_session_key takes profile explicitly; the adapter passes
-        # source.profile through. Verify the namespace follows it.
-        assert build_session_key(s, profile="coder") == "agent:coder:telegram:dm:99"
-
-
-class TestWebhookProfileResolution:
-    """_resolve_request_profile validates the /p/<profile>/ prefix."""
-
-    def _adapter(self, multiplex: bool, served=("default", "coder")):
-        from gateway.platforms.webhook import WebhookAdapter, _PROFILE_REJECTED
-
-        class _FakeReq:
-            def __init__(self, profile):
-                self.match_info = {"profile": profile} if profile is not None else {}
-
-        cfg = GatewayConfig(multiplex_profiles=multiplex)
-
-        class _Runner:
-            config = cfg
-
-        # Construct minimally; we only call _resolve_request_profile.
-        adapter = WebhookAdapter.__new__(WebhookAdapter)
-        adapter.gateway_runner = _Runner()
-        return adapter, _FakeReq, _PROFILE_REJECTED, served
-
-    def test_no_prefix_returns_none(self):
-        adapter, Req, _REJ, _ = self._adapter(multiplex=True)
-        assert adapter._resolve_request_profile(Req(None)) is None
-
-    def test_prefix_ignored_when_multiplex_off(self):
-        adapter, Req, _REJ, _ = self._adapter(multiplex=False)
-        # Even a bogus profile is ignored (not 404'd) when multiplexing is off.
-        assert adapter._resolve_request_profile(Req("anything")) is None
-
-    def test_known_profile_accepted(self, monkeypatch):
-        adapter, Req, _REJ, served = self._adapter(multiplex=True)
-        monkeypatch.setattr(
-            "hermes_cli.profiles.profiles_to_serve",
-            lambda multiplex: [(n, None) for n in served],
-        )
-        assert adapter._resolve_request_profile(Req("coder")) == "coder"
-
-    def test_unknown_profile_rejected(self, monkeypatch):
-        adapter, Req, REJ, served = self._adapter(multiplex=True)
-        monkeypatch.setattr(
-            "hermes_cli.profiles.profiles_to_serve",
-            lambda multiplex: [(n, None) for n in served],
-        )
-        assert adapter._resolve_request_profile(Req("ghost")) is REJ
@@ -1,55 +0,0 @@
-"""Phase 4: lifecycle guard + per-profile observability."""
-import pytest
-
-
-class TestServedProfilesStatus:
-    def test_write_and_read_served_profiles(self, tmp_path, monkeypatch):
-        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
-        import importlib
-        import gateway.status as status
-        importlib.reload(status)
-        try:
-            status.write_runtime_status(
-                gateway_state="running", served_profiles=["default", "coder"]
-            )
-            rec = status.read_runtime_status()
-            assert rec.get("served_profiles") == ["default", "coder"]
-        finally:
-            importlib.reload(status)
-
-    def test_served_profiles_absent_by_default(self, tmp_path, monkeypatch):
-        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
-        import importlib
-        import gateway.status as status
-        importlib.reload(status)
-        try:
-            status.write_runtime_status(gateway_state="running")
-            rec = status.read_runtime_status()
-            assert "served_profiles" not in rec
-        finally:
-            importlib.reload(status)
-
-
-class TestNamedProfileMultiplexerGuard:
-    """_guard_named_profile_under_multiplexer is inert unless all conditions hold."""
-
-    def test_inert_for_default_profile(self, monkeypatch):
-        from hermes_cli import gateway as gw
-        monkeypatch.setattr(gw, "_profile_suffix", lambda: "")
-        # Should return without raising (default profile => guard N/A).
-        gw._guard_named_profile_under_multiplexer(force=False)
-
-    def test_force_bypasses(self, monkeypatch):
-        from hermes_cli import gateway as gw
-        # Even if it looks like a named profile, force returns immediately.
-        monkeypatch.setattr(gw, "_profile_suffix", lambda: "coder")
-        gw._guard_named_profile_under_multiplexer(force=True)
-
-    def test_inert_when_no_default_gateway_running(self, monkeypatch, tmp_path):
-        from hermes_cli import gateway as gw
-        monkeypatch.setattr(gw, "_profile_suffix", lambda: "coder")
-        monkeypatch.setattr(
-            "hermes_constants.get_default_hermes_root", lambda: tmp_path
-        )
-        # No gateway.pid in tmp_path => no running default gateway => no raise.
-        gw._guard_named_profile_under_multiplexer(force=False)
@@ -1,165 +0,0 @@
-"""Phase 0 foundations for multi-profile gateway multiplexing.
-
-Covers the three Phase 0 deliverables:
-  1. ``gateway.multiplex_profiles`` config flag (default False, round-trips).
-  2. ``hermes_cli.profiles.profiles_to_serve`` enumeration.
-  3. Profile-stamped ``build_session_key`` that is BYTE-IDENTICAL when the
-     flag is off (the orphan-every-session guard) and namespace-segmented when
-     on, without disturbing the positional key layout downstream parsers rely
-     on.
-"""
-import pytest
-from unittest.mock import patch
-
-from gateway.config import GatewayConfig, Platform
-from gateway.session import SessionSource, SessionStore, build_session_key
-
-
-def _src(**kw) -> SessionSource:
-    kw.setdefault("platform", Platform.TELEGRAM)
-    kw.setdefault("chat_id", "99")
-    kw.setdefault("chat_type", "dm")
-    return SessionSource(**kw)
-
-
-class TestSessionKeyByteIdenticalWhenOff:
-    """The non-negotiable guard: with no profile (or 'default'), every key is
-    byte-for-byte what it was before Phase 0. A diff here orphans every
-    existing session on upgrade."""
-
-    @pytest.mark.parametrize("profile", [None, "default"])
-    def test_dm_with_chat_id(self, profile):
-        s = _src(chat_id="99", chat_type="dm")
-        assert build_session_key(s, profile=profile) == "agent:main:telegram:dm:99"
-
-    @pytest.mark.parametrize("profile", [None, "default"])
-    def test_dm_with_thread(self, profile):
-        s = _src(chat_id="99", chat_type="dm", thread_id="t1")
-        assert build_session_key(s, profile=profile) == "agent:main:telegram:dm:99:t1"
-
-    @pytest.mark.parametrize("profile", [None, "default"])
-    def test_dm_without_chat_id_falls_back_to_user(self, profile):
-        s = _src(chat_id="", chat_type="dm", user_id="jordan")
-        assert build_session_key(s, profile=profile) == "agent:main:telegram:dm:jordan"
-
-    @pytest.mark.parametrize("profile", [None, "default"])
-    def test_group_per_user(self, profile):
-        s = _src(platform=Platform.DISCORD, chat_id="g1", chat_type="group", user_id="alice")
-        assert (
-            build_session_key(s, profile=profile)
-            == "agent:main:discord:group:g1:alice"
-        )
-
-    @pytest.mark.parametrize("profile", [None, "default"])
-    def test_group_shared_when_disabled(self, profile):
-        s = _src(platform=Platform.DISCORD, chat_id="g1", chat_type="group", user_id="alice")
-        assert (
-            build_session_key(s, group_sessions_per_user=False, profile=profile)
-            == "agent:main:discord:group:g1"
-        )
-
-
-class TestSessionKeyNamespacedWhenOn:
-    """A named profile occupies the namespace slot, isolating its sessions."""
-
-    def test_named_profile_dm(self):
-        s = _src(chat_id="99", chat_type="dm")
-        assert build_session_key(s, profile="coder") == "agent:coder:telegram:dm:99"
-
-    def test_named_profile_group_per_user(self):
-        s = _src(platform=Platform.DISCORD, chat_id="g1", chat_type="group", user_id="alice")
-        assert (
-            build_session_key(s, profile="coder")
-            == "agent:coder:discord:group:g1:alice"
-        )
-
-    def test_two_profiles_same_chat_do_not_collide(self):
-        s = _src(chat_id="99", chat_type="dm")
-        a = build_session_key(s, profile="default")
-        b = build_session_key(s, profile="coder")
-        c = build_session_key(s, profile="writer")
-        assert a != b != c and a != c
-
-    def test_positional_layout_preserved_for_parsers(self):
-        """Downstream parsers split on ':' and read parts[2]=platform,
-        parts[3]=chat_type, parts[4]=chat_id (see qqbot adapter
-        _parse_gateway_session_key). The profile must occupy parts[1] only."""
-        s = _src(platform=Platform.DISCORD, chat_id="g1", chat_type="group", user_id="alice")
-        parts = build_session_key(s, profile="coder").split(":")
-        assert parts[0] == "agent"
-        assert parts[1] == "coder"  # namespace slot (was always 'main')
-        assert parts[2] == "discord"  # platform — unchanged offset
-        assert parts[3] == "group"  # chat_type — unchanged offset
-        assert parts[4] == "g1"  # chat_id — unchanged offset
-
-    def test_default_namespace_layout_matches_named(self):
-        """Default and named keys differ ONLY in parts[1]."""
-        s = _src(platform=Platform.SLACK, chat_id="c1", chat_type="channel", user_id="u1")
-        d = build_session_key(s, profile="default").split(":")
-        n = build_session_key(s, profile="coder").split(":")
-        assert d[0] == n[0] == "agent"
-        assert d[1] == "main" and n[1] == "coder"
-        assert d[2:] == n[2:]  # everything after the namespace is identical
-
-
-class TestMultiplexConfigFlag:
-    """gateway.multiplex_profiles defaults off and round-trips."""
-
-    def test_default_is_false(self):
-        assert GatewayConfig().multiplex_profiles is False
-
-    def test_to_dict_includes_flag(self):
-        assert GatewayConfig().to_dict()["multiplex_profiles"] is False
-
-    def test_from_dict_top_level(self):
-        cfg = GatewayConfig.from_dict({"multiplex_profiles": True})
-        assert cfg.multiplex_profiles is True
-
-    def test_from_dict_nested_gateway(self):
-        cfg = GatewayConfig.from_dict({"gateway": {"multiplex_profiles": True}})
-        assert cfg.multiplex_profiles is True
-
-    def test_from_dict_coerces_truthy_string(self):
-        cfg = GatewayConfig.from_dict({"multiplex_profiles": "true"})
-        assert cfg.multiplex_profiles is True
-
-    def test_roundtrip(self):
-        cfg = GatewayConfig.from_dict(GatewayConfig(multiplex_profiles=True).to_dict())
-        assert cfg.multiplex_profiles is True
-
-
-class TestSessionStoreProfileResolution:
-    """SessionStore._generate_session_key honors the flag: legacy namespace
-    when off, active-profile namespace when on."""
-
-    def _store(self, tmp_path, **cfg_kw):
-        config = GatewayConfig(**cfg_kw)
-        with patch("gateway.session.SessionStore._ensure_loaded"):
-            s = SessionStore(sessions_dir=tmp_path, config=config)
-        s._db = None
-        s._loaded = True
-        return s
-
-    def test_flag_off_uses_legacy_namespace(self, tmp_path):
-        store = self._store(tmp_path)  # multiplex_profiles defaults False
-        s = _src(chat_id="99", chat_type="dm")
-        assert store._generate_session_key(s) == "agent:main:telegram:dm:99"
-        assert store._generate_session_key(s) == build_session_key(s)
-
-    def test_flag_off_resolve_profile_is_none(self, tmp_path):
-        store = self._store(tmp_path)
-        assert store._resolve_profile_for_key() is None
-
-    def test_flag_on_uses_active_profile_namespace(self, tmp_path):
-        store = self._store(tmp_path, multiplex_profiles=True)
-        s = _src(chat_id="99", chat_type="dm")
-        with patch("hermes_cli.profiles.get_active_profile_name", return_value="coder"):
-            assert store._generate_session_key(s) == "agent:coder:telegram:dm:99"
-
-    def test_flag_on_default_profile_stays_legacy(self, tmp_path):
-        store = self._store(tmp_path, multiplex_profiles=True)
-        s = _src(chat_id="99", chat_type="dm")
-        with patch("hermes_cli.profiles.get_active_profile_name", return_value="default"):
-            assert store._generate_session_key(s) == "agent:main:telegram:dm:99"
-
-
@@ -543,6 +543,126 @@ class TestImport:
        # traversal file should NOT exist outside hermes home
        assert not (tmp_path / "etc" / "passwd").exists()

+    def test_preserves_live_gateway_state(self, tmp_path, monkeypatch):
+        """Import must not overwrite the target's gateway_state.json.
+
+        The backup carries the *source* machine's gateway run/desired state.
+        Restoring it onto a hosted container drives the boot reconciler off
+        stale/foreign state and leaves the gateway stuck "starting",
+        disconnecting it from the Nous portal (NS-508). The live file wins.
+        """
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+        monkeypatch.setattr(Path, "home", lambda: tmp_path)
+
+        # The target (e.g. hosted container) already has its own live state.
+        live_state = '{"gateway_state": "running", "desired_state": "running"}'
+        (hermes_home / "gateway_state.json").write_text(live_state)
+
+        zip_path = tmp_path / "backup.zip"
+        self._make_backup_zip(zip_path, {
+            "config.yaml": "model: test\n",
+            # A backup from a laptop where the gateway was stopped.
+            "gateway_state.json": '{"gateway_state": "stopped", "desired_state": "stopped"}',
+        })
+
+        args = Namespace(zipfile=str(zip_path), force=True)
+
+        from hermes_cli.backup import run_import
+        run_import(args)
+
+        # config.yaml is restored normally...
+        assert (hermes_home / "config.yaml").read_text() == "model: test\n"
+        # ...but the live gateway_state.json is untouched.
+        assert (hermes_home / "gateway_state.json").read_text() == live_state
+
+    def test_does_not_seed_gateway_state_when_absent(self, tmp_path, monkeypatch):
+        """A backup's gateway_state.json is dropped, not written, when the
+        target has none — a foreign state must never seed the reconciler."""
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+        monkeypatch.setattr(Path, "home", lambda: tmp_path)
+
+        zip_path = tmp_path / "backup.zip"
+        self._make_backup_zip(zip_path, {
+            "config.yaml": "model: test\n",
+            "gateway_state.json": '{"gateway_state": "stopped"}',
+        })
+
+        args = Namespace(zipfile=str(zip_path), force=True)
+
+        from hermes_cli.backup import run_import
+        run_import(args)
+
+        assert (hermes_home / "config.yaml").exists()
+        assert not (hermes_home / "gateway_state.json").exists()
+
+    def test_preserves_per_profile_gateway_state(self, tmp_path, monkeypatch):
+        """The skip is matched by basename, so a named profile's
+        gateway_state.json (profiles/<name>/gateway_state.json) is preserved
+        the same way the root profile's is."""
+        hermes_home = tmp_path / ".hermes"
+        (hermes_home / "profiles" / "coder").mkdir(parents=True)
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+        monkeypatch.setattr(Path, "home", lambda: tmp_path)
+
+        live_state = '{"gateway_state": "running"}'
+        (hermes_home / "profiles" / "coder" / "gateway_state.json").write_text(live_state)
+
+        zip_path = tmp_path / "backup.zip"
+        self._make_backup_zip(zip_path, {
+            "config.yaml": "model: test\n",
+            "profiles/coder/config.yaml": "model: anthropic\n",
+            "profiles/coder/gateway_state.json": '{"gateway_state": "stopped"}',
+        })
+
+        args = Namespace(zipfile=str(zip_path), force=True)
+
+        from hermes_cli.backup import run_import
+        run_import(args)
+
+        # Profile config is restored, but its live gateway state is preserved.
+        assert (hermes_home / "profiles" / "coder" / "config.yaml").read_text() == "model: anthropic\n"
+        assert (
+            hermes_home / "profiles" / "coder" / "gateway_state.json"
+        ).read_text() == live_state
+
+    def test_preserves_runtime_pid_and_process_files(self, tmp_path, monkeypatch):
+        """gateway.pid / cron.pid / gateway.lock / processes.json from a backup
+        reference the source machine's process namespace and must never be
+        written over the target's."""
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+        monkeypatch.setattr(Path, "home", lambda: tmp_path)
+
+        # Live runtime files belonging to the target's own processes.
+        (hermes_home / "gateway.pid").write_text("4242")
+        (hermes_home / "processes.json").write_text('{"live": true}')
+
+        zip_path = tmp_path / "backup.zip"
+        self._make_backup_zip(zip_path, {
+            "config.yaml": "model: test\n",
+            "gateway.pid": "9999",
+            "cron.pid": "8888",
+            "gateway.lock": "7777",
+            "processes.json": '{"stale": true}',
+        })
+
+        args = Namespace(zipfile=str(zip_path), force=True)
+
+        from hermes_cli.backup import run_import
+        run_import(args)
+
+        # Live runtime files are untouched; the backup's foreign ones never land.
+        assert (hermes_home / "gateway.pid").read_text() == "4242"
+        assert (hermes_home / "processes.json").read_text() == '{"live": true}'
+        # cron.pid / gateway.lock had no live copy and were not seeded.
+        assert not (hermes_home / "cron.pid").exists()
+        assert not (hermes_home / "gateway.lock").exists()
+
    def test_confirmation_prompt_abort(self, tmp_path, monkeypatch):
        """Import aborts when user says no to confirmation."""
        hermes_home = tmp_path / ".hermes"
@@ -35,7 +35,6 @@ from hermes_cli.profiles import (
    has_bundled_skills_opt_out,
    NO_BUNDLED_SKILLS_MARKER,
    backfill_profile_envs,
-    profiles_to_serve,
 )
 from hermes_cli.config import DEFAULT_CONFIG

@@ -1488,48 +1487,3 @@ class TestEdgeCases:
            delete_profile("coder", yes=True)

        assert get_active_profile() == "default"
-
-
-class TestProfilesToServe:
-    """profiles_to_serve(multiplex) — the gateway's profile-enumeration chokepoint."""
-
-    def test_off_returns_only_active_default(self, profile_env):
-        serve = profiles_to_serve(multiplex=False)
-        assert len(serve) == 1
-        name, home = serve[0]
-        assert name == "default"
-        assert home == _get_default_hermes_home()
-
-    def test_off_returns_only_active_named(self, profile_env, monkeypatch):
-        # A named profile's gateway runs with HERMES_HOME pointing at the
-        # profile dir; get_active_profile_name() infers the name from there.
-        create_profile("coder", no_alias=True)
-        monkeypatch.setenv("HERMES_HOME", str(get_profile_dir("coder")))
-        serve = profiles_to_serve(multiplex=False)
-        assert len(serve) == 1
-        assert serve[0][0] == "coder"
-        assert serve[0][1] == get_profile_dir("coder")
-
-    def test_on_returns_default_plus_all_named(self, profile_env):
-        create_profile("coder", no_alias=True)
-        create_profile("writer", no_alias=True)
-        serve = dict(profiles_to_serve(multiplex=True))
-        assert set(serve) == {"default", "coder", "writer"}
-        assert serve["default"] == _get_default_hermes_home()
-        assert serve["coder"] == get_profile_dir("coder")
-
-    def test_on_default_always_first(self, profile_env):
-        create_profile("coder", no_alias=True)
-        serve = profiles_to_serve(multiplex=True)
-        assert serve[0][0] == "default"
-
-    def test_on_active_profile_does_not_change_set(self, profile_env):
-        """Enumeration is independent of which profile is active."""
-        create_profile("coder", no_alias=True)
-        set_active_profile("coder")
-        serve = dict(profiles_to_serve(multiplex=True))
-        assert set(serve) == {"default", "coder"}
-
-    def test_on_no_named_profiles_returns_just_default(self, profile_env):
-        serve = profiles_to_serve(multiplex=True)
-        assert [n for n, _ in serve] == ["default"]
@@ -331,3 +331,108 @@ def test_hosted_policy_locks_to_opt_data(monkeypatch):

    assert str(policy.locked_root) == "/opt/data"
    assert policy.can_change_path is False
+
+
+# ---------------------------------------------------------------------------
+# Streaming multipart upload (/api/files/upload-stream) — NS-501
+# ---------------------------------------------------------------------------
+
+
+def test_stream_upload_roundtrip(forced_files_client):
+    """The multipart endpoint writes raw bytes to disk and reports the entry."""
+    client, root = forced_files_client
+    file_path = root / "out" / "backup.zip"
+    payload = b"PK\x03\x04 not really a zip but binary enough \x00\x01\x02"
+
+    created = client.post(
+        "/api/files/upload-stream",
+        data={"path": str(file_path), "overwrite": "true"},
+        files={"file": ("backup.zip", payload, "application/zip")},
+    )
+    assert created.status_code == 200, created.text
+    assert created.json()["entry"]["path"] == str(file_path)
+    assert created.json()["locked_root"] == str(root)
+    # Bytes land verbatim — no base64 round-trip, no corruption.
+    assert file_path.read_bytes() == payload
+
+
+def test_stream_upload_rejects_oversized_without_clobbering(forced_files_client, monkeypatch):
+    """Over-limit uploads return 413 and never overwrite an existing file.
+
+    The size cap is enforced while streaming (not after buffering), and the
+    temp-file + atomic-rename design means a rejected upload leaves any
+    pre-existing file at the target path untouched.
+    """
+    client, root = forced_files_client
+    file_path = root / "out" / "big.bin"
+
+    # Seed an existing file at the target path.
+    seeded = client.post(
+        "/api/files/upload-stream",
+        data={"path": str(file_path), "overwrite": "true"},
+        files={"file": ("big.bin", b"original-contents", "application/octet-stream")},
+    )
+    assert seeded.status_code == 200
+    assert file_path.read_bytes() == b"original-contents"
+
+    # Shrink the cap so a small payload trips it deterministically.
+    monkeypatch.setattr(web_server, "_MANAGED_FILE_MAX_BYTES", 8)
+    rejected = client.post(
+        "/api/files/upload-stream",
+        data={"path": str(file_path), "overwrite": "true"},
+        files={"file": ("big.bin", b"way too many bytes for the cap", "application/octet-stream")},
+    )
+    assert rejected.status_code == 413
+    # The original file must survive a rejected overwrite.
+    assert file_path.read_bytes() == b"original-contents"
+    # No stray temp files left behind in the directory.
+    leftovers = [p.name for p in file_path.parent.iterdir() if ".upload" in p.name]
+    assert leftovers == [], f"temp upload files leaked: {leftovers}"
+
+
+def test_stream_upload_respects_overwrite_false(forced_files_client):
+    client, root = forced_files_client
+    file_path = root / "keep.txt"
+
+    first = client.post(
+        "/api/files/upload-stream",
+        data={"path": str(file_path), "overwrite": "true"},
+        files={"file": ("keep.txt", b"first", "text/plain")},
+    )
+    assert first.status_code == 200
+
+    conflict = client.post(
+        "/api/files/upload-stream",
+        data={"path": str(file_path), "overwrite": "false"},
+        files={"file": ("keep.txt", b"second", "text/plain")},
+    )
+    assert conflict.status_code == 409
+    assert file_path.read_bytes() == b"first"
+
+
+def test_stream_upload_stays_under_forced_root(forced_files_client):
+    """A relative path with traversal can't escape the locked root."""
+    client, root = forced_files_client
+    escaped = client.post(
+        "/api/files/upload-stream",
+        data={"path": "../../etc/evil.txt", "overwrite": "true"},
+        files={"file": ("evil.txt", b"nope", "text/plain")},
+    )
+    assert escaped.status_code in (400, 403)
+
+
+def test_stream_upload_large_file_under_cap_succeeds(forced_files_client, monkeypatch):
+    """A multi-chunk payload (larger than the 1 MiB chunk) streams correctly."""
+    client, root = forced_files_client
+    file_path = root / "multi-chunk.bin"
+    # 2.5 MiB exercises the chunked read loop across multiple iterations.
+    payload = b"x" * (2 * 1024 * 1024 + 512 * 1024)
+
+    created = client.post(
+        "/api/files/upload-stream",
+        data={"path": str(file_path), "overwrite": "true"},
+        files={"file": ("multi-chunk.bin", payload, "application/octet-stream")},
+    )
+    assert created.status_code == 200
+    assert file_path.stat().st_size == len(payload)
+    assert file_path.read_bytes() == payload
@@ -178,6 +178,7 @@ LAZY_DEPS: dict[str, tuple[str, ...]] = {
        "fastapi==0.133.1",
        "uvicorn[standard]==0.41.0",
        "starlette==1.0.1",  # CVE-2026-48710 (BadHost) — keep lazy-install in sync with pyproject [web]
+        "python-multipart==0.0.20",  # FastAPI UploadFile/Form for streaming uploads (NS-501)
    ),
    # Vision image-resize recovery (Pillow). Pillow is now a CORE dependency
    # (pyproject `dependencies`), so this entry is a belt-and-suspenders fallback
@@ -2662,19 +2662,10 @@ def _interrupted_call_result() -> str:
 # ---------------------------------------------------------------------------

 def _interpolate_env_vars(value):
-    """Recursively resolve ``${VAR}`` placeholders.
-
-    Resolves from the active profile's secret scope when multiplexing is on
-    (so an MCP server config's ``${API_KEY}`` picks up the routed profile's
-    value, not the process-global ``os.environ`` which may hold another
-    profile's), falling back to ``os.environ`` otherwise. Unset vars keep the
-    literal ``${VAR}`` placeholder, as before.
-    """
-    from agent.secret_scope import get_secret as _get_secret
-
+    """Recursively resolve ``${VAR}`` placeholders from ``os.environ``."""
    if isinstance(value, str):
        def _replace(m):
-            return _get_secret(m.group(1), m.group(0)) or m.group(0)
+            return os.environ.get(m.group(1), m.group(0))
        return _ENV_VAR_PATTERN.sub(_replace, value)
    if isinstance(value, dict):
        return {k: _interpolate_env_vars(v) for k, v in value.items()}
@@ -1445,6 +1445,7 @@ dependencies = [
    { name = "pydantic" },
    { name = "pyjwt", extra = ["crypto"] },
    { name = "python-dotenv" },
+    { name = "python-multipart" },
    { name = "pywinpty", marker = "sys_platform == 'win32'" },
    { name = "pyyaml" },
    { name = "requests" },
@@ -1469,6 +1470,7 @@ all = [
    { name = "google-auth-httplib2" },
    { name = "google-auth-oauthlib" },
    { name = "mcp" },
+    { name = "python-multipart" },
    { name = "simple-term-menu" },
    { name = "starlette" },
    { name = "uvicorn", extra = ["standard"] },
@@ -1598,6 +1600,7 @@ termux-all = [
    { name = "google-auth-oauthlib" },
    { name = "honcho-ai" },
    { name = "mcp" },
+    { name = "python-multipart" },
    { name = "python-telegram-bot", extra = ["webhooks"] },
    { name = "simple-term-menu" },
    { name = "starlette" },
@@ -1613,6 +1616,7 @@ voice = [
 ]
 web = [
    { name = "fastapi" },
+    { name = "python-multipart" },
    { name = "starlette" },
    { name = "uvicorn", extra = ["standard"] },
 ]
@@ -1708,6 +1712,8 @@ requires-dist = [
    { name = "pytest", marker = "extra == 'dev'", specifier = "==9.0.2" },
    { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = "==1.3.0" },
    { name = "python-dotenv", specifier = "==1.2.2" },
+    { name = "python-multipart", specifier = ">=0.0.9,<1" },
+    { name = "python-multipart", marker = "extra == 'web'", specifier = "==0.0.31" },
    { name = "python-telegram-bot", extras = ["webhooks"], marker = "extra == 'messaging'", specifier = "==22.6" },
    { name = "python-telegram-bot", extras = ["webhooks"], marker = "extra == 'termux'", specifier = "==22.6" },
    { name = "pywinpty", marker = "sys_platform == 'win32'", specifier = ">=2.0.0,<3" },
@@ -3311,11 +3317,11 @@ wheels = [

 [[package]]
 name = "python-multipart"
-version = "0.0.27"
+version = "0.0.31"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/69/9b/f23807317a113dc36e74e75eb265a02dd1a4d9082abc3c1064acd22997c4/python_multipart-0.0.27.tar.gz", hash = "sha256:9870a6a8c5a20a5bf4f07c017bd1489006ff8836cff097b6933355ee2b49b602", size = 44043, upload-time = "2026-04-27T10:51:26.649Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/64/7e/9b35ad8f3d9ca680f7c87a88f19612fdd8da9796c4d3b46e560ac79dcc4a/python_multipart-0.0.31.tar.gz", hash = "sha256:fc631183bb13e56db3158a4909908dfb2e23565286744e798241e63750e5d680", size = 46689, upload-time = "2026-06-04T08:27:49.014Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/99/78/4126abcbdbd3c559d43e0db7f7b9173fc6befe45d39a2856cc0b8ec2a5a6/python_multipart-0.0.27-py3-none-any.whl", hash = "sha256:6fccfad17a27334bd0193681b369f476eda3409f17381a2d65aa7df3f7275645", size = 29254, upload-time = "2026-04-27T10:51:24.997Z" },
+    { url = "https://files.pythonhosted.org/packages/5e/1e/7f7f299527a5a8ad90acd5f2f78dfa6c8495c6301a3205106ea68a84de96/python_multipart-0.0.31-py3-none-any.whl", hash = "sha256:8408153d68a9773291fc1da39a8b85a50044bddbabd2dd72e9229776b7b15e28", size = 29996, upload-time = "2026-06-04T08:27:47.804Z" },
 ]

 [[package]]
@@ -411,12 +411,21 @@ export const api = {
    fetchJSON<ManagedFileReadResponse>(
      `/api/files/read?path=${encodeURIComponent(path)}`,
    ),
-  uploadFile: (path: string, dataUrl: string, overwrite = true) =>
-    fetchJSON<ManagedFileWriteResponse>("/api/files/upload", {
+  uploadFile: (path: string, file: File, overwrite = true) => {
+    // Stream the raw bytes as multipart/form-data. Do NOT set Content-Type —
+    // the browser adds the multipart boundary automatically. Sending the file
+    // as base64 JSON (the old path) inflated the body ~33%, buffered the whole
+    // file in memory, and 502'd on large backup archives behind the proxy
+    // (NS-501).
+    const form = new FormData();
+    form.append("path", path);
+    form.append("overwrite", String(overwrite));
+    form.append("file", file, file.name);
+    return fetchJSON<ManagedFileWriteResponse>("/api/files/upload-stream", {
      method: "POST",
-      headers: { "Content-Type": "application/json" },
-      body: JSON.stringify({ path, data_url: dataUrl, overwrite }),
-    }),
+      body: form,
+    });
+  },
  createDirectory: (path: string) =>
    fetchJSON<ManagedFileWriteResponse>("/api/files/mkdir", {
      method: "POST",
@@ -58,18 +58,6 @@ function formatBytes(size: number | null): string {
  return `${(size / (1024 * 1024 * 1024)).toFixed(1)} GB`;
 }

-function readAsDataUrl(file: globalThis.File): Promise<string> {
-  return new Promise((resolve, reject) => {
-    const reader = new FileReader();
-    reader.addEventListener("load", () => {
-      if (typeof reader.result === "string") resolve(reader.result);
-      else reject(new Error("Could not read file"));
-    });
-    reader.addEventListener("error", () => reject(reader.error ?? new Error("Could not read file")));
-    reader.readAsDataURL(file);
-  });
-}
-
 function downloadDataUrl(dataUrl: string, name: string) {
  const link = document.createElement("a");
  link.href = dataUrl;
@@ -205,8 +193,7 @@ export default function FilesPage() {
    setUploading(true);
    try {
      for (const file of Array.from(files)) {
-        const dataUrl = await readAsDataUrl(file);
-        await api.uploadFile(joinPath(activePath, file.name), dataUrl, true);
+        await api.uploadFile(joinPath(activePath, file.name), file, true);
      }
      showToast(`${files.length} file${files.length === 1 ? "" : "s"} uploaded`, "success");
      await load();
Author	SHA1	Message	Date
dependabot[bot]	7d57bfd5d4	chore(deps): bump python-multipart from 0.0.20 to 0.0.31 Bumps [python-multipart](https://github.com/Kludex/python-multipart) from 0.0.20 to 0.0.31. - [Release notes](https://github.com/Kludex/python-multipart/releases) - [Changelog](https://github.com/Kludex/python-multipart/blob/main/CHANGELOG.md) - [Commits](https://github.com/Kludex/python-multipart/compare/0.0.20...0.0.31) --- updated-dependencies: - dependency-name: python-multipart dependency-version: 0.0.31 dependency-type: indirect ... Signed-off-by: dependabot[bot] <support@github.com>	2026-06-18 05:57:04 +00:00
Ben Barclay	c661634537	fix(dashboard): stream file uploads via multipart instead of base64 JSON (NS-501) (#47663 ) * fix(dashboard): stream file uploads via multipart instead of base64 JSON The dashboard file manager uploaded files (including backup/restore zip archives) by reading them client-side with FileReader.readAsDataURL and POSTing a base64 data URL inside a JSON body to /api/files/upload. For a large backup this (a) inflates the payload ~33%, (b) buffers the whole file plus its decoded copy in memory, and (c) reliably trips an upstream proxy body-size/timeout limit, surfacing as a 502 with the upload appearing to hang indefinitely (NS-501). Dashboard-only hosted users have no shell fallback to place the archive, so backup restore was unusable. Add a streaming multipart endpoint POST /api/files/upload-stream (UploadFile + Form) that reads the request body in 1 MiB chunks straight to a sibling temp file, enforces the existing 100 MB size cap as it streams (413 on overflow, before buffering the whole file), and atomically renames into place so a partial/aborted/over-limit upload never clobbers an existing file. The frontend api.uploadFile now sends multipart/form-data (raw bytes, no base64, browser-set boundary) and FilesPage passes the File object directly; the dead readAsDataUrl helper is removed. The legacy base64 JSON endpoint stays for backward compat. FastAPI's UploadFile/Form require python-multipart, which is NOT pulled in by fastapi itself, so it is added to the base deps, the [web] extra, and the tool.dashboard lazy-install set (kept in sync). Validated: 5 new endpoint tests (roundtrip, multi-chunk >1 MiB, over-limit 413 without clobbering + no temp-file leak, overwrite=false conflict, forced-root traversal containment); existing base64 tests still pass; web typecheck + vite build clean; and a real uvicorn server E2E (5 MB multipart upload -> HTTP 200 in 0.21s, exact byte match) plus a 30 MB TestClient roundtrip confirm constant-memory streaming end to end. Reported via beta (NS-501). * build(deps): regenerate uv.lock for python-multipart (NS-501) CI ran uv lock --check / uv sync --locked which failed because the python-multipart dependency add was not reflected in uv.lock. Regenerate the lockfile (resolves to 0.0.20, matching the [web] extra pin) after merging current main.	2026-06-18 15:54:32 +10:00
Ben Barclay	9c3c5da356	fix(backup): hermes import never overwrites volatile gateway runtime state (NS-501) (#48243 ) Importing a backup wrote every file from the zip over the target home wholesale. On a hosted instance this clobbered gateway_state.json with the source machine's last recorded run/desired state — driving the container-boot reconciler (container_boot._read_desired_state, which only auto-starts a gateway whose state is "running") off stale/foreign state and leaving the gateway stuck "starting", disconnected from the Nous portal. Add _IMPORT_SKIP_NAMES (gateway_state.json, gateway.pid, cron.pid, gateway.lock, processes.json) and skip them by basename in run_import, so both the root profile and named profiles preserve the target's own runtime state. This mirrors what container_boot._STALE_RUNTIME_FILES already sweeps on every container boot, and protects against older backups that predate the backup-side exclusions. The import summary reports which files were preserved. This is the second half of NS-501 (filed separately as NS-508): the upload 502 was fixed in #47663; this fixes the import-breaks-the-instance half.	2026-06-18 15:27:45 +10:00
Ben Barclay	0ddd21c74e	feat(relay): managed-boot self-provision client (Phase 3, gateway side) (#48242 ) The gateway half of relay Phase 3. On a MANAGED boot with relay configured and no secret pinned, the runtime self-provisions its relay credentials IN-PROCESS: resolve the agent's own Nous access token (resolve_nous_access_token) -> POST the connector's /relay/provision asserting its own endpoint + route keys -> set GATEWAY_RELAY_ID/SECRET/DELIVERY_KEY into os.environ so the immediately- following register_relay_adapter() reads them and dials out authenticated. No human, no enrollment token, no disk write — the creds live only in process memory (save_env_value refuses under managed anyway, and keeping the secret off any volume is the stronger posture). Stateless: process-env creds don't survive a restart, so a managed container re-provisions every boot; the connector's rotation window covers a still-connected prior instance. An explicitly-pinned GATEWAY_RELAY_SECRET is respected (skip). Self-hosted is unchanged: humans keep using `hermes gateway enroll`. Endpoint provenance is gateway-asserted (GATEWAY_RELAY_ENDPOINT + GATEWAY_RELAY_ROUTE_KEYS, env or gateway.relay_* config) — uniform code path whether the operator sets it (self-hosted) or NAS stamps it (hosted, the only case NAS knows the public URL). Both absent -> outbound-only provisioning (credentials, no inbound routes). The connector scopes the asserted endpoint to the verified tenant, so it stays within the security model. - gateway/relay/__init__.py: relay_endpoint(), relay_route_keys(), _provision_url(), _post_provision(), self_provision_if_managed() (never raises — a provision failure logs and boots without relay auth). - gateway/run.py: call self_provision_if_managed() immediately before register_relay_adapter() in the startup path. Tests: 12 unit (trigger logic, respect-pinned-secret, in-process env wiring, endpoint+routes vs outbound-only, fail-soft on token/connector failure); mutation-checked (drop is_managed guard / pinned-secret guard -> tests fail). Cross-repo live E2E driver lands on the connector side (depends on this). EXPERIMENTAL: relay auth scheme may change until >=2 Class-1 platforms validate.	2026-06-18 15:25:29 +10:00