Merge remote-tracking branch 'origin/main' into fix/discord-interaction-command-normalization

The multiplex cron path only used use_cron_store() to scope storage paths
(jobs.json, heartbeat files), but _get_lock_paths() and the agent execution
path in cron/scheduler.py resolve via _get_hermes_home() → get_hermes_home()
which checks _HERMES_HOME_OVERRIDE, a separate ContextVar. Without
set_hermes_home_override(), the .tick.lock, config.yaml, .env, and secrets
all resolved to the default profile instead of the per-profile home.

This matches the web_server.py pattern (line 11994) which sets both
set_hermes_home_override(home) AND use_cron_store(home), and the
_profile_runtime_scope pattern used for the multiplexed inbound path.

Found via 3-agent parallel review of salvaged PR #69529.

agent/memory_manager.py

@@ -80,8 +80,17 @@ def normalize_tool_schema(schema: Any) -> Optional[Dict[str, Any]]:
     return schema
 
 
-def memory_provider_tools_enabled(enabled_toolsets: Optional[List[str]]) -> bool:
+def memory_provider_tools_enabled(
+    enabled_toolsets: Optional[List[str]],
+    disabled_toolsets: Optional[List[str]] = None,
+    *,
+    memory_tool_present: bool = False,
+) -> bool:
     """Return whether external memory-provider tools should be exposed."""
+    if disabled_toolsets and "memory" in disabled_toolsets:
+        return False
+    if memory_tool_present:
+        return True
     if enabled_toolsets is None:
         return True
     if not enabled_toolsets:
@@ -110,9 +119,10 @@ def inject_memory_provider_tools(agent: Any) -> int:
         for tool in tools
         if isinstance(tool, dict)
     }
-    if (
-        "memory" not in existing_tool_names
-        and not memory_provider_tools_enabled(getattr(agent, "enabled_toolsets", None))
+    if not memory_provider_tools_enabled(
+        getattr(agent, "enabled_toolsets", None),
+        getattr(agent, "disabled_toolsets", None),
+        memory_tool_present="memory" in existing_tool_names,
     ):
         return 0
 

cron/jobs.py

@@ -816,15 +816,20 @@ def record_ticker_heartbeat(success: bool = False) -> None:
     (both fresh) — a ticker stuck failing every tick would otherwise keep the
     plain heartbeat fresh and falsely report healthy (#32612, #32895).
 
+    Resolution uses ``_current_cron_store()`` so the heartbeat is correctly
+    scoped to the active profile's store — critical under multiplex_profiles
+    where each profile needs its own liveness signal (#69377).
+
     Best-effort: a write failure must never disrupt the tick loop.
     """
+    store = _current_cron_store()
     try:
-        _atomic_write_epoch(TICKER_HEARTBEAT_FILE)
+        _atomic_write_epoch(store.cron_dir / "ticker_heartbeat")
     except Exception:
         pass
     if success:
         try:
-            _atomic_write_epoch(TICKER_SUCCESS_FILE)
+            _atomic_write_epoch(store.cron_dir / "ticker_last_success")
         except Exception:
             pass
 
@@ -842,13 +847,24 @@ def get_ticker_heartbeat_age() -> Optional[float]:
 
     None = heartbeat file missing/unreadable (older build, never ran, or a
     torn read). Callers treat None as "cannot determine", not "dead".
+
+    Resolution uses ``_current_cron_store()`` so the heartbeat is correctly
+    scoped to the active profile — critical under multiplex_profiles where
+    ``hermes cron status`` must report per-profile liveness (#69377).
     """
-    return _epoch_file_age(TICKER_HEARTBEAT_FILE)
+    store = _current_cron_store()
+    return _epoch_file_age(store.cron_dir / "ticker_heartbeat")
 
 
 def get_ticker_success_age() -> Optional[float]:
-    """Seconds since the ticker last completed a tick WITHOUT raising, or None."""
-    return _epoch_file_age(TICKER_SUCCESS_FILE)
+    """Seconds since the ticker last completed a tick WITHOUT raising, or None.
+
+    Resolution uses ``_current_cron_store()`` so the heartbeat is correctly
+    scoped to the active profile — critical under multiplex_profiles where
+    ``hermes cron status`` must report per-profile liveness (#69377).
+    """
+    store = _current_cron_store()
+    return _epoch_file_age(store.cron_dir / "ticker_last_success")
 
 
 # =============================================================================

cron/scheduler_provider.py

@@ -173,13 +173,42 @@ class InProcessCronScheduler(CronScheduler):
     def name(self) -> str:
         return "builtin"
 
-    def start(self, stop_event, *, adapters=None, loop=None, interval=60, can_dispatch=None):
+    def start(
+        self,
+        stop_event,
+        *,
+        adapters=None,
+        loop=None,
+        interval=60,
+        can_dispatch=None,
+        profile_homes=None,
+    ):
         import logging
         from cron.scheduler import tick as cron_tick
         from cron.jobs import record_ticker_heartbeat
 
         logger = logging.getLogger("cron.scheduler_provider")
         logger.info("In-process cron scheduler started (interval=%ds)", interval)
+
+        # ── Multiplex profiles ────────────────────────────────────────────
+        # When profile_homes is set (multiplex_profiles on), tick EACH profile's
+        # cron store on every tick cycle so secondary-profile jobs actually fire
+        # instead of languishing in a store no ticker owns (#69377). Without this,
+        # only the process-global HERMES_HOME (the default profile) is ticked.
+        # Heartbeats and recovery are also scoped per profile so `hermes cron
+        # status` reflects liveness for every profile independently.
+        if profile_homes:
+            self._start_multiplex(
+                stop_event,
+                profile_homes=profile_homes,
+                adapters=adapters,
+                loop=loop,
+                interval=interval,
+                can_dispatch=can_dispatch,
+            )
+            return
+
+        # ── Single-profile (legacy) path ──────────────────────────────────
         recovered = self.recover_interrupted()
         if recovered:
             logger.warning(
@@ -217,3 +246,84 @@ class InProcessCronScheduler(CronScheduler):
             # "actually firing jobs" (#32612, #32895).
             record_ticker_heartbeat(success=ok)
             stop_event.wait(interval)
+
+    def _start_multiplex(
+        self,
+        stop_event,
+        *,
+        profile_homes,
+        adapters=None,
+        loop=None,
+        interval=60,
+        can_dispatch=None,
+    ):
+        """Tick every served profile's cron store when multiplex_profiles is on.
+
+        Each profile uses ``set_hermes_home_override()`` + ``use_cron_store()``
+        to scope its tick, heartbeat, recovery, lock file, config/.env, and
+        agent execution to that profile's home — mirroring how
+        ``_profile_runtime_scope`` scopes the multiplexed inbound path and
+        ``web_server.py`` scopes per-profile cron API calls.
+        """
+        import logging
+        from cron.scheduler import tick as cron_tick
+        from cron.jobs import record_ticker_heartbeat, use_cron_store
+        from hermes_constants import set_hermes_home_override, reset_hermes_home_override
+
+        logger = logging.getLogger("cron.scheduler_provider")
+        logger.info(
+            "Multiplex cron scheduler started for %d profile(s): %s",
+            len(profile_homes),
+            [p[0] if isinstance(p, tuple) else p for p in profile_homes],
+        )
+
+        # Recovery + initial heartbeat for every profile.
+        for entry in profile_homes:
+            home = entry[1] if isinstance(entry, tuple) else entry
+            home_token = set_hermes_home_override(str(home))
+            try:
+                with use_cron_store(home):
+                    recovered = self.recover_interrupted()
+                    if recovered:
+                        logger.warning(
+                            "Marked %d interrupted cron execution(s) for profile at %s",
+                            recovered,
+                            home,
+                        )
+                    record_ticker_heartbeat()
+            finally:
+                reset_hermes_home_override(home_token)
+
+        while not stop_event.is_set():
+            ok = False
+            try:
+                if can_dispatch is not None and not can_dispatch():
+                    logger.debug("Cron dispatch paused while gateway drains existing work")
+                else:
+                    for entry in profile_homes:
+                        home = entry[1] if isinstance(entry, tuple) else entry
+                        home_token = set_hermes_home_override(str(home))
+                        try:
+                            with use_cron_store(home):
+                                cron_tick(
+                                    verbose=False,
+                                    adapters=adapters,
+                                    loop=loop,
+                                    sync=False,
+                                    can_dispatch=can_dispatch,
+                                )
+                        finally:
+                            reset_hermes_home_override(home_token)
+                ok = True
+            except BaseException as e:
+                logger.error("Cron tick error: %s", e, exc_info=True)
+            # Record per-profile heartbeat after each tick cycle.
+            for entry in profile_homes:
+                home = entry[1] if isinstance(entry, tuple) else entry
+                home_token = set_hermes_home_override(str(home))
+                try:
+                    with use_cron_store(home):
+                        record_ticker_heartbeat(success=ok)
+                finally:
+                    reset_hermes_home_override(home_token)
+            stop_event.wait(interval)

gateway/platforms/base.py

@@ -2555,30 +2555,6 @@ class BasePlatformAdapter(ABC):
         """
         return len
 
-    def max_message_length_for_chat(self, chat_id: str) -> int:
-        """Per-chat max message length, in ``message_len_fn_for_chat`` units.
-
-        Default: the adapter-scalar ``MAX_MESSAGE_LENGTH`` (4096 when absent) —
-        for a native adapter every chat lives on the same platform so the
-        scalar is already correct. The relay adapter overrides this: one relay
-        adapter fronts N platforms with different caps (Discord 2000 vs
-        Telegram 4096 vs Slack 39000), and the right cap depends on which
-        platform the chat's inbound arrived from.
-        """
-        try:
-            return int(getattr(self, "MAX_MESSAGE_LENGTH", 4096) or 4096)
-        except (TypeError, ValueError):
-            return 4096
-
-    def message_len_fn_for_chat(self, chat_id: str) -> Callable[[str], int]:
-        """Per-chat length function (companion to max_message_length_for_chat).
-
-        Default: the adapter-wide ``message_len_fn``. The relay adapter
-        overrides it so a Telegram-fronted chat measures UTF-16 units while a
-        Discord-fronted chat on the same adapter measures codepoints.
-        """
-        return self.message_len_fn
-
     @property
     def enforces_own_access_policy(self) -> bool:
         """Whether this adapter gates inbound access before dispatch.

gateway/relay/adapter.py

@@ -20,7 +20,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
-from typing import Any, Callable, Dict, Optional, cast
+from typing import Any, Callable, Dict, Optional
 
 from gateway.config import Platform, PlatformConfig
 from gateway.platforms.base import BasePlatformAdapter, MessageEvent, SendResult
@@ -107,42 +107,6 @@ class RelayAdapter(BasePlatformAdapter):
     def message_len_fn(self) -> Callable[[str], int]:
         return _LEN_FNS.get(self.descriptor.len_unit, len)
 
-    # ── per-chat capability resolution (Phase 1.5 multi-platform) ─────────
-    def _descriptor_for_chat(self, chat_id: str) -> CapabilityDescriptor:
-        """The capability descriptor governing a specific chat.
-
-        A multi-platform gateway fronts N platforms on ONE adapter, but the
-        scalar `descriptor`/`MAX_MESSAGE_LENGTH` surface can only carry one
-        platform's profile (the primary identity's). Platform caps genuinely
-        differ — Discord 2000 / Telegram 4096 / Slack 39000 — so applying the
-        primary's cap to every chat either fragments needlessly (small primary)
-        or over-sends into a platform 400 (large primary; the live bug: 2,543
-        and 2,641-char sends rejected by Discord). Resolve the chat's platform
-        from what we saw inbound (`_platform_by_chat`, the same map per-frame
-        egress uses) and look up that platform's negotiated descriptor on the
-        transport. Falls back to the scalar descriptor when the chat's platform
-        is unknown (never saw inbound) or the transport predates the map.
-        """
-        platform = self._platform_by_chat.get(str(chat_id))
-        if platform and self._transport is not None:
-            resolve = getattr(self._transport, "descriptor_for_platform", None)
-            if callable(resolve):
-                try:
-                    per_platform = cast(
-                        Optional[CapabilityDescriptor], resolve(platform)
-                    )
-                except Exception:  # noqa: BLE001 - capability lookup must never break a send
-                    per_platform = None
-                if per_platform is not None:
-                    return per_platform
-        return self.descriptor
-
-    def max_message_length_for_chat(self, chat_id: str) -> int:
-        return self._descriptor_for_chat(chat_id).max_message_length
-
-    def message_len_fn_for_chat(self, chat_id: str) -> Callable[[str], int]:
-        return _LEN_FNS.get(self._descriptor_for_chat(chat_id).len_unit, len)
-
     def supports_draft_streaming(
         self,
         chat_type: Optional[str] = None,
@@ -397,12 +361,13 @@ class RelayAdapter(BasePlatformAdapter):
 
         NEVER raises: a malformed forward must not kill the read loop.
 
-        NOTE (open semantic sub-design, flagged for review): the interaction ->
-        MessageEvent mapping below is the v1 default. The exact agent UX for a
-        slash-command / button interaction (vs. a plain message) — command name
-        surfacing, option rendering, deferred-vs-immediate response — is the open
-        piece tracked in the spec; the TRANSPORT + receive mechanism (this whole
-        path) is settled.
+        Interaction -> MessageEvent command mapping (formerly flagged here as an
+        open sub-design, now implemented): an APPLICATION_COMMAND interaction is
+        normalized to a leading-slash COMMAND event ("/name arg…", mirroring the
+        connector's Slack slash-command lane, normalizeSlackCommand), so the
+        dispatcher routes it as a command instead of plain chat. Component
+        interactions (custom_id) still surface as best-effort TEXT; the
+        deferred-vs-immediate response UX remains connector-side.
         """
         try:
             platform = getattr(forward, "platform", "") or ""
@@ -445,8 +410,21 @@ class RelayAdapter(BasePlatformAdapter):
         # 3 = MESSAGE_COMPONENT; 5 = MODAL_SUBMIT. Surface a best-effort text.
         itype = payload.get("type")
         data = payload.get("data") or {}
+        message_type = MessageType.TEXT
         if itype == 2:
-            text = str(data.get("name") or "")
+            # Normalize a real slash-command interaction to a leading-slash
+            # command string — the shape the dispatcher (MessageEvent.is_command:
+            # text.startswith("/")) and the native Discord adapter's
+            # _run_simple_slash lane (f"/model {name}".strip()) both expect.
+            # Options render space-separated: scalar options contribute their
+            # value; SUB_COMMAND/SUB_COMMAND_GROUP (types 1/2) contribute their
+            # name then their nested options. Mirrors the connector's Slack
+            # slash lane (normalizeSlackCommand: `${command} ${args}`.trim()).
+            text = ("/" + str(data.get("name") or "")).rstrip("/") or ""
+            if text:
+                parts = [text] + self._render_interaction_options(data.get("options"))
+                text = " ".join(parts).strip()
+                message_type = MessageType.COMMAND
         elif itype == 3:
             text = str(data.get("custom_id") or "")
         else:
@@ -464,7 +442,36 @@ class RelayAdapter(BasePlatformAdapter):
             scope_id=str(guild_id) if guild_id else None,  # Discord guild → generic scope slot
             message_id=str(payload.get("id")) if payload.get("id") else None,
         )
-        return MessageEvent(text=text, message_type=MessageType.TEXT, source=source)
+        return MessageEvent(text=text, message_type=message_type, source=source)
+
+    @staticmethod
+    def _render_interaction_options(options) -> list:
+        """Render Discord interaction options to space-separated text parts.
+
+        Discord's `data.options` is a list of {name, value, type}. Scalar
+        options (STRING/INTEGER/BOOLEAN/…) contribute just their value —
+        matching the native adapter's `f"/model {name}".strip()` shape, where
+        only the value follows the command. SUB_COMMAND (1) and
+        SUB_COMMAND_GROUP (2) contribute their *name* then recurse into their
+        nested `options` list (one level of nesting per Discord's schema:
+        group -> subcommand -> scalars).
+        """
+        parts: list = []
+        if not isinstance(options, list):
+            return parts
+        for opt in options:
+            if not isinstance(opt, dict):
+                continue
+            if opt.get("type") in (1, 2):  # SUB_COMMAND / SUB_COMMAND_GROUP
+                sub_name = str(opt.get("name") or "").strip()
+                if sub_name:
+                    parts.append(sub_name)
+                parts.extend(RelayAdapter._render_interaction_options(opt.get("options")))
+            else:
+                value = opt.get("value")
+                if value is not None and str(value).strip():
+                    parts.append(str(value).strip())
+        return parts
 
     async def disconnect(self) -> None:
         # Phase 7 Unit 7d-B: stop the revocation monitor first so it can't fire a

gateway/relay/ws_transport.py

@@ -370,11 +370,6 @@ class WebSocketRelayTransport:
         self._reader: Optional[asyncio.Task[None]] = None
         self._inbound: Optional[InboundHandler] = None
         self._descriptor: Optional[CapabilityDescriptor] = None
-        # Phase 1.5 multi-platform: descriptors keyed by the underlying platform
-        # (one per hello'd identity). `_descriptor` above stays the FIRST
-        # (primary-identity) descriptor for back-compat; this map is the
-        # per-platform capability surface read via `descriptor_for_platform`.
-        self._descriptors_by_platform: Dict[str, CapabilityDescriptor] = {}
         self._descriptor_ready: asyncio.Future[CapabilityDescriptor] | None = None
         # requestId -> future awaiting the matching outbound_result.
         self._pending: Dict[str, asyncio.Future[Dict[str, Any]]] = {}
@@ -404,10 +399,8 @@ class WebSocketRelayTransport:
         loop = asyncio.get_running_loop()
         self._descriptor_ready = loop.create_future()
         # A fresh handshake is coming; clear any stale descriptor so handshake()
-        # awaits the new one (matters on a re-dial). The per-platform map resets
-        # with it — a reconnected connector re-sends one descriptor per hello.
+        # awaits the new one (matters on a re-dial).
         self._descriptor = None
-        self._descriptors_by_platform = {}
         # scale-to-zero (D12): a successful (re-)dial ends any dormant state — we
         # are live again, so a subsequent UNEXPECTED close should reconnect on the
         # normal fast backoff, not the dormant cadence.
@@ -480,18 +473,6 @@ class WebSocketRelayTransport:
             raise RuntimeError("handshake() called before connect()")
         return await asyncio.wait_for(self._descriptor_ready, timeout=self._connect_timeout_s)
 
-    def descriptor_for_platform(self, platform: str) -> Optional[CapabilityDescriptor]:
-        """The negotiated descriptor for one fronted platform, or None.
-
-        Phase 1.5 multi-platform: the connector replies one descriptor per
-        hello'd identity; they accumulate here keyed by the descriptor's own
-        ``platform`` field. Callers (RelayAdapter) use this to resolve PER-CHAT
-        capabilities — e.g. Discord's 2000-char max_message_length vs
-        Telegram's 4096 — instead of applying the primary identity's scalar
-        descriptor to every platform this gateway fronts.
-        """
-        return self._descriptors_by_platform.get(platform)
-
     @property
     def auth_revoked(self) -> bool:
         """True once the connector closed the socket with 4401 AFTER a prior
@@ -767,19 +748,7 @@ class WebSocketRelayTransport:
         ftype = frame.get("type")
         if ftype == "descriptor":
             descriptor = CapabilityDescriptor.from_json(json.dumps(frame.get("descriptor", {})))
-            # Phase 1.5 multi-platform: one descriptor frame arrives per hello'd
-            # identity. Accumulate them keyed by the descriptor's own platform so
-            # the adapter can resolve PER-CHAT capabilities (e.g. Discord's 2000
-            # vs Telegram's 4096 max_message_length) instead of collapsing N
-            # platforms onto whichever descriptor arrived last.
-            if descriptor.platform:
-                self._descriptors_by_platform[descriptor.platform] = descriptor
-            # The FIRST descriptor of this connection generation is the session
-            # default (the primary identity's) — later arrivals must NOT
-            # overwrite it, or the scalar capability surface silently becomes
-            # last-writer-wins across platforms.
-            if self._descriptor is None:
-                self._descriptor = descriptor
+            self._descriptor = descriptor
             # Phase 7 Unit 7d-B: a received descriptor means the WS upgrade auth
             # passed and the connector accepted us — record that we've handshaked
             # at least once, so a LATER 4401 close is read as a revocation

gateway/run.py

@@ -20551,17 +20551,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 _raw_progress_limit = int(getattr(adapter, "MAX_MESSAGE_LENGTH", 4000) or 4000)
             except Exception:
                 _raw_progress_limit = 4000
-            # Per-chat resolution (relay adapter fronting N platforms): the cap
-            # and length unit follow the chat's underlying platform. Native
-            # adapters return their scalar/property unchanged.
-            if isinstance(adapter, BasePlatformAdapter):
-                try:
-                    _raw_progress_limit = int(
-                        adapter.max_message_length_for_chat(source.chat_id) or 4000
-                    )
-                    _progress_len_fn = adapter.message_len_fn_for_chat(source.chat_id)
-                except Exception:
-                    pass
             # Leave a little room for platform quirks / formatting.  For tiny
             # test adapters keep the limit usable instead of clamping to 500+.
             _PROGRESS_TEXT_LIMIT = max(
@@ -23984,7 +23973,35 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
     from cron.scheduler_provider import InProcessCronScheduler, resolve_cron_scheduler
     cron_stop = threading.Event()
     cron_provider = resolve_cron_scheduler()
-    cron_start_kwargs = {"adapters": runner.adapters, "loop": asyncio.get_running_loop()}
+    cron_start_kwargs: Dict[str, Any] = {"adapters": runner.adapters, "loop": asyncio.get_running_loop()}
+
+    # Multiplex profiles: tell the built-in ticker which profile homes to
+    # tick so secondary-profile cron jobs actually fire (#69377).
+    # Without this, only the process-global HERMES_HOME (default profile)
+    # is iterated and every secondary profile's cron store is silently
+    # ignored — jobs show as "scheduled" with a valid next_run_at but
+    # never execute because no ticker owns that store.
+    if (
+        isinstance(cron_provider, InProcessCronScheduler)
+        and getattr(runner.config, "multiplex_profiles", False)
+    ):
+        try:
+            from hermes_cli.profiles import profiles_to_serve
+
+            profile_homes = list(profiles_to_serve(multiplex=True))
+            if profile_homes:
+                cron_start_kwargs["profile_homes"] = profile_homes
+                logger.info(
+                    "Cron scheduler will tick %d profile(s) under multiplex: %s",
+                    len(profile_homes),
+                    [p[0] if isinstance(p, tuple) else p for p in profile_homes],
+                )
+        except Exception as exc:
+            logger.warning(
+                "Could not resolve profile homes for multiplex cron: %s",
+                exc,
+            )
+
     # External cron providers own their remote scheduling contract. Only the
     # in-process ticker polls local due jobs, so only it receives the local
     # external-drain dispatch gate.

gateway/stream_consumer.py

@@ -675,13 +675,10 @@ class GatewayStreamConsumer:
         # Platform message length limit — leave room for cursor + formatting.
         # Use the adapter's length function (e.g. utf16_len for Telegram) so
         # overflow detection matches what the platform actually enforces.
-        # Both resolve PER-CHAT (max_message_length_for_chat): a relay adapter
-        # fronting N platforms has different caps per chat (Discord 2000 vs
-        # Telegram 4096); native adapters return their scalar unchanged.
         # Gate on isinstance(BasePlatformAdapter) so test MagicMocks (whose
         # auto-attributes return mock objects, not callables) fall back to len.
         _len_fn: "Callable[[str], int]" = (
-            self.adapter.message_len_fn_for_chat(self.chat_id)
+            self.adapter.message_len_fn
             if isinstance(self.adapter, _BasePlatformAdapter)
             else len
         )
@@ -1362,15 +1359,6 @@ class GatewayStreamConsumer:
             if isinstance(self.adapter, _BasePlatformAdapter)
             else len
         )
-        # Per-chat resolution (relay adapter fronting N platforms): the cap and
-        # length unit follow the chat's underlying platform, not the adapter
-        # scalar. Native adapters return their scalar/property unchanged.
-        if isinstance(self.adapter, _BasePlatformAdapter):
-            try:
-                raw_limit = self.adapter.max_message_length_for_chat(self.chat_id)
-                _len_fn = self.adapter.message_len_fn_for_chat(self.chat_id)
-            except Exception as e:
-                logger.debug("per-chat limit resolution failed: %s", e)
         safe_limit = max(500, raw_limit - 100)
         chunks = self._split_text_chunks(continuation, safe_limit, len_fn=_len_fn)
 
@@ -1756,11 +1744,8 @@ class GatewayStreamConsumer:
         """Per-message length budget (in the adapter's ``message_len_fn`` units)
         before the consumer splits an overflowing reply.
 
-        Resolved PER-CHAT via ``max_message_length_for_chat`` — a relay adapter
-        fronting N platforms has a different cap per chat (Discord 2000 vs
-        Telegram 4096 vs Slack 39000); native adapters return their scalar
-        ``MAX_MESSAGE_LENGTH`` unchanged. Adapters with a richer send/draft
-        path (e.g. Telegram rich messages) can raise this above the base via
+        Adapters with a richer send/draft path (e.g. Telegram rich messages)
+        can raise this above ``MAX_MESSAGE_LENGTH`` via
         ``streaming_overflow_limit`` so a reply that fits one rich message isn't
         fragmented at the legacy edit limit.  Falls back to
         ``MAX_MESSAGE_LENGTH`` (4096 default) for everyone else.
@@ -1769,10 +1754,6 @@ class GatewayStreamConsumer:
         # isinstance gate: MagicMock adapters return mock objects (truthy, not
         # ints) for arbitrary attribute access — keep them on the base limit.
         if isinstance(self.adapter, _BasePlatformAdapter):
-            try:
-                base = self.adapter.max_message_length_for_chat(self.chat_id)
-            except Exception as e:
-                logger.debug("max_message_length_for_chat failed: %s", e)
             try:
                 cap = self.adapter.streaming_overflow_limit()
             except Exception as e:

hermes_cli/memory_setup.py

@@ -379,6 +379,21 @@ def cmd_setup(args) -> None:
     print("\n  Start a new session to activate.\n")
 
 
+def _env_line_safe(value) -> str:
+    """Neutralize characters that would break ``.env`` line structure.
+
+    ``.env`` is strictly line-oriented (one ``KEY=VALUE`` per line) and
+    values are interpolated straight into that line. A pasted secret with an
+    embedded CR/LF would spill onto a new line and be re-parsed as a
+    *separate* ``KEY=VALUE`` entry on the next read — injecting an arbitrary
+    variable into the credentials file. Strip every separator recognized by
+    ``str.splitlines()`` plus NUL so a value can only occupy its own line.
+    Mirrors the openviking plugin's writer and ``config.save_env_value``.
+    """
+    text = value if isinstance(value, str) else str(value)
+    return "".join(text.replace("\x00", "").splitlines())
+
+
 def _write_env_vars(env_path: Path, env_writes: dict) -> None:
     """Append or update env vars in .env file."""
     env_path.parent.mkdir(parents=True, exist_ok=True)
@@ -392,14 +407,14 @@ def _write_env_vars(env_path: Path, env_writes: dict) -> None:
     for line in existing_lines:
         key_match = line.split("=", 1)[0].strip() if "=" in line else ""
         if key_match in env_writes:
-            new_lines.append(f"{key_match}={env_writes[key_match]}")
+            new_lines.append(f"{key_match}={_env_line_safe(env_writes[key_match])}")
             updated_keys.add(key_match)
         else:
             new_lines.append(line)
 
     for key, val in env_writes.items():
         if key not in updated_keys:
-            new_lines.append(f"{key}={val}")
+            new_lines.append(f"{key}={_env_line_safe(val)}")
 
     env_path.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
     # Restrict permissions — .env holds API keys and tokens.

plugins/memory/openviking/__init__.py

@@ -126,6 +126,11 @@ _GENERATED_MEMORY_SUMMARY_FILENAMES = {
 }
 _LOCAL_OPENVIKING_HOSTS = {"localhost", "127.0.0.1", "::1"}
 _LOCAL_OPENVIKING_AUTOSTART_TIMEOUT = 60.0
+# After a refresh attempt fails for a given (unchanged) config, skip re-probing
+# for this long. Keeps "unavailable endpoints reconnect on a later access"
+# true while preventing every provider access from paying a 3s health probe
+# (and emitting a warning) under _client_refresh_lock while a server is down.
+_FAILED_CONFIG_RETRY_COOLDOWN_SECONDS = 30.0
 _OPENVIKING_SERVER_LOG_RELATIVE_PATH = Path("logs") / "openviking-server.log"
 _OPENVIKING_RESPONDED_FAILURE_PREFIX = "OpenViking server responded"
 _PENDING_SESSIONS_RELATIVE_DIR = Path("openviking") / "pending_sessions"
@@ -294,12 +299,23 @@ class _VikingClient:
 
     @staticmethod
     def _needs_trusted_identity_retry(exc: Exception) -> bool:
+        """Detect errors that indicate missing tenant-scoped identity headers.
+
+        Trusted mode can ask for ``X-OpenViking-Account`` /
+        ``X-OpenViking-User`` using slightly different wording across
+        OpenViking versions. Match that trusted-mode missing-identity shape
+        instead of enumerating every exact string, while keeping deliberate
+        API-key permission denials non-retriable.
+        """
         message = str(exc)
-        return (
-            "Trusted mode requests must include X-OpenViking-Account" in message
-            or "Trusted mode requests must include X-OpenViking-User" in message
-            or "Trusted mode requests must include X-OpenViking-Account or explicit account_id" in message
-        )
+        if "Trusted mode requests must include" not in message:
+            return False
+        if "X-OpenViking-Account" not in message and "X-OpenViking-User" not in message:
+            return False
+        status_code = getattr(exc, "status_code", None)
+        if status_code is not None and status_code != 400:
+            return False
+        return True
 
     def _send_with_trusted_identity_retry(self, send, *, multipart: bool = False) -> dict:
         try:
@@ -880,9 +896,9 @@ def _is_local_openviking_url(value: str) -> bool:
 
 def _load_hermes_openviking_config() -> dict:
     try:
-        from hermes_cli.config import load_config
+        from hermes_cli.config import load_config_readonly
 
-        config = load_config()
+        config = load_config_readonly()
         memory_config = config.get("memory", {}) if isinstance(config, dict) else {}
         provider_config = memory_config.get("openviking", {}) if isinstance(memory_config, dict) else {}
         return dict(provider_config) if isinstance(provider_config, dict) else {}
@@ -962,6 +978,23 @@ def _precreate_secret_file(path: Path) -> None:
         logger.debug("Could not pre-create secret file %s: %s", path, e)
 
 
+def _env_line_safe(value: Any) -> str:
+    """Neutralize characters that would break ``.env`` line structure.
+
+    A ``.env`` file is strictly line-oriented (one ``KEY=VALUE`` per line),
+    and ``_write_env_vars`` interpolates each value straight into that line.
+    A value carrying an embedded CR/LF would therefore spill onto a new line
+    and be re-parsed as a *separate* ``KEY=VALUE`` entry on the next
+    ``read_text().splitlines()`` round-trip — letting a malformed or pasted
+    secret (e.g. an api_key copied with a trailing record) inject an
+    arbitrary additional variable into the persisted credentials file. Strip
+    the line separators recognized by ``splitlines()`` and the NUL byte so a
+    value can only ever occupy the single line it is written on.
+    """
+    text = value if isinstance(value, str) else str(value)
+    return "".join(text.replace("\x00", "").splitlines())
+
+
 def _write_env_vars(env_path: Path, env_writes: dict, remove_keys: tuple[str, ...] = ()) -> None:
     env_path.parent.mkdir(parents=True, exist_ok=True)
     remove_set = set(remove_keys) - set(env_writes)
@@ -973,13 +1006,13 @@ def _write_env_vars(env_path: Path, env_writes: dict, remove_keys: tuple[str, ..
         if key_match in remove_set:
             continue
         if key_match in env_writes:
-            new_lines.append(f"{key_match}={env_writes[key_match]}")
+            new_lines.append(f"{key_match}={_env_line_safe(env_writes[key_match])}")
             updated_keys.add(key_match)
         else:
             new_lines.append(line)
     for key, val in env_writes.items():
         if key not in updated_keys:
-            new_lines.append(f"{key}={val}")
+            new_lines.append(f"{key}={_env_line_safe(val)}")
     # Pre-create with 0600 so secrets are never briefly world-readable.
     _precreate_secret_file(env_path)
     env_path.write_text("\n".join(new_lines) + ("\n" if new_lines else ""), encoding="utf-8")
@@ -1205,9 +1238,20 @@ def _start_local_openviking_server(endpoint: str) -> tuple[bool, str]:
     return True, f"Started openviking-server on {host}:{port} in the background. Logs: {log_path}"
 
 
-def _wait_for_openviking_health(endpoint: str, *, timeout_seconds: float = 15.0) -> bool:
+def _wait_for_openviking_health(
+    endpoint: str,
+    *,
+    timeout_seconds: float = 15.0,
+    should_stop=None,
+) -> bool:
     deadline = time.monotonic() + timeout_seconds
     while time.monotonic() < deadline:
+        # Bail out promptly if the provider is being torn down, so the daemon
+        # thread running this waiter can be join()ed at shutdown instead of
+        # lingering up to ``timeout_seconds`` (a worker still alive at
+        # interpreter exit aborts CPython with SIGABRT at Py_FinalizeEx).
+        if should_stop is not None and should_stop():
+            return False
         ok, _message = _validate_openviking_reachability(endpoint)
         if ok:
             return True
@@ -1830,6 +1874,11 @@ class OpenVikingMemoryProvider(MemoryProvider):
         self._run_id = uuid.uuid4().hex
         self._run_lock_file: Optional[Any] = None
         self._run_lock_path: Optional[Path] = None
+        # Set once initialize() has resolved the connection baseline. Until then
+        # _ensure_client() must not re-resolve from the environment — callers
+        # that wire up a client directly (e.g. tests) would otherwise have it
+        # discarded. See _ensure_client() / #21130.
+        self._env_refresh_enabled = False
         # Guards the (_session_id, _turn_count) pair. sync_turn runs on the
         # MemoryManager's background sync executor while on_session_end /
         # on_session_switch run on the caller's thread, so the snapshot+reset
@@ -1848,8 +1897,22 @@ class OpenVikingMemoryProvider(MemoryProvider):
         self._committed_session_ids: Set[str] = set()
         self._committed_session_lock = threading.Lock()
         self._pending_marked_sids: Set[str] = set()
+        # Connection settings and _client are one published state. Serialize
+        # refreshes so callers never observe a new config with the old client.
+        self._client_refresh_lock = threading.Lock()
+        # Last connection identity that passed a health check, published as a
+        # single tuple assignment (atomic in CPython) so lock-free background
+        # writers (_new_client, on_memory_write) never see a torn mix of old
+        # and new fields, and never target an endpoint that failed health.
+        self._conn_snapshot: Optional[tuple] = None
+        # (settings tuple, monotonic timestamp) of the last refresh attempt
+        # that failed. While the resolved config still matches and the retry
+        # cooldown hasn't elapsed, _ensure_client_locked() returns None without
+        # re-probing — keeping provider accesses cheap while a server is down.
+        self._failed_refresh: Optional[tuple] = None
         self._runtime_start_lock = threading.Lock()
         self._runtime_start_thread: Optional[threading.Thread] = None
+        self._runtime_start_pending = False
         self._memory_write_lock = threading.Lock()
         self._memory_write_threads: Set[threading.Thread] = set()
         self._profile_prefetched_sessions: Set[str] = set()
@@ -2065,72 +2128,98 @@ class OpenVikingMemoryProvider(MemoryProvider):
     def _start_runtime_openviking_waiter(
         self,
         *,
+        endpoint: str,
         status_callback=None,
         warning_callback=None,
     ) -> None:
-        with self._runtime_start_lock:
-            if self._runtime_start_thread and self._runtime_start_thread.is_alive():
-                return
-            self._runtime_start_thread = threading.Thread(
-                target=self._finish_runtime_openviking_start,
-                kwargs={
-                    "status_callback": status_callback,
-                    "warning_callback": warning_callback,
-                },
-                daemon=True,
-                name="openviking-runtime-start",
-            )
-            self._runtime_start_thread.start()
+        # Precondition: caller holds _runtime_start_lock. Local process start
+        # ownership is reserved with _runtime_start_pending before callbacks run.
+        if self._runtime_start_thread and self._runtime_start_thread.is_alive():
+            return
+        self._runtime_start_thread = threading.Thread(
+            target=self._finish_runtime_openviking_start,
+            kwargs={
+                "endpoint": endpoint,
+                "status_callback": status_callback,
+                "warning_callback": warning_callback,
+            },
+            daemon=True,
+            name="openviking-runtime-start",
+        )
+        self._runtime_start_thread.start()
 
     def _finish_runtime_openviking_start(
         self,
         *,
+        endpoint: Optional[str] = None,
         status_callback=None,
         warning_callback=None,
     ) -> None:
-        endpoint = self._endpoint
+        endpoint = endpoint or self._endpoint
         if not _wait_for_openviking_health(
             endpoint,
             timeout_seconds=_LOCAL_OPENVIKING_AUTOSTART_TIMEOUT,
+            should_stop=lambda: self._shutting_down or self._endpoint != endpoint,
         ):
+            if self._shutting_down or self._endpoint != endpoint:
+                return
             _emit_runtime_warning(
                 _runtime_openviking_timeout_message(endpoint),
                 warning_callback,
             )
             return
 
-        try:
-            client = _VikingClient(
-                endpoint,
-                self._api_key,
-                account=self._account,
-                user=self._user,
-                agent=self._agent,
-            )
-            if not client.health():
-                _emit_runtime_warning(
-                    f"OpenViking server at {endpoint} is still not reachable after auto-start; "
-                    "OpenViking memory disabled for this Hermes run.",
-                    warning_callback,
-                )
+        warning_message = ""
+        status_message = ""
+        with self._client_refresh_lock:
+            if self._shutting_down or self._endpoint != endpoint:
                 return
-        except ImportError:
-            logger.warning("httpx not installed — OpenViking plugin disabled")
-            return
-        except Exception as e:
+            try:
+                client = _VikingClient(
+                    endpoint,
+                    self._api_key,
+                    account=self._account,
+                    user=self._user,
+                    agent=self._agent,
+                )
+                healthy = client.health()
+                if self._shutting_down or self._endpoint != endpoint:
+                    return
+                if not healthy:
+                    warning_message = (
+                        f"OpenViking server at {endpoint} is still not reachable after auto-start; "
+                        "OpenViking memory disabled for this Hermes run."
+                    )
+                else:
+                    self._client = client
+                    self._conn_snapshot = (
+                        endpoint, self._api_key, self._account, self._user, self._agent,
+                    )
+                    self._failed_refresh = None
+                    status_message = (
+                        f"Local OpenViking server at {endpoint} is reachable; "
+                        "OpenViking memory is active for later turns."
+                    )
+            except ImportError:
+                logger.warning("httpx not installed — OpenViking plugin disabled")
+                return
+            except Exception as e:
+                warning_message = (
+                    f"OpenViking server at {endpoint} could not be attached after auto-start: {e}. "
+                    "OpenViking memory disabled for this Hermes run."
+                )
+
+        if warning_message:
             _emit_runtime_warning(
-                f"OpenViking server at {endpoint} could not be attached after auto-start: {e}. "
-                "OpenViking memory disabled for this Hermes run.",
+                warning_message,
                 warning_callback,
             )
             return
-
-        self._client = client
-        self._recover_pending_sessions()
-        _emit_runtime_status(
-            f"Local OpenViking server at {endpoint} is reachable; OpenViking memory is active for later turns.",
-            status_callback,
-        )
+        if status_message:
+            # Client attached: recover orphaned sessions outside the refresh
+            # lock (network I/O), then announce.
+            self._recover_pending_sessions()
+            _emit_runtime_status(status_message, status_callback)
 
     def _handle_runtime_openviking_unreachable(
         self,
@@ -2149,25 +2238,52 @@ class OpenVikingMemoryProvider(MemoryProvider):
             self._client = None
             return
 
-        started, start_message = _start_local_openviking_server(endpoint)
-        if not started:
+        warning_message = ""
+        status_message = ""
+        should_start_waiter = False
+        with self._runtime_start_lock:
+            if (
+                self._shutting_down
+                or self._runtime_start_pending
+                or (self._runtime_start_thread and self._runtime_start_thread.is_alive())
+            ):
+                self._client = None
+                return
+
+            self._runtime_start_pending = True
+            started, start_message = _start_local_openviking_server(endpoint)
+            if not started:
+                self._runtime_start_pending = False
+                warning_message = (
+                    f"Local OpenViking server at {endpoint} is not reachable. {start_message} "
+                    "OpenViking memory disabled for this Hermes run."
+                )
+                self._client = None
+            else:
+                self._client = None
+                status_message = (
+                    f"{start_message} OpenViking memory is starting in the background and will attach when ready."
+                )
+                should_start_waiter = True
+
+        if warning_message:
             _emit_runtime_warning(
-                f"Local OpenViking server at {endpoint} is not reachable. {start_message} "
-                "OpenViking memory disabled for this Hermes run.",
+                warning_message,
                 warning_callback,
             )
-            self._client = None
             return
-
-        self._client = None
-        _emit_runtime_status(
-            f"{start_message} OpenViking memory is starting in the background and will attach when ready.",
-            status_callback,
-        )
-        self._start_runtime_openviking_waiter(
-            status_callback=status_callback,
-            warning_callback=warning_callback,
-        )
+        if status_message:
+            _emit_runtime_status(status_message, status_callback)
+        if should_start_waiter:
+            with self._runtime_start_lock:
+                self._runtime_start_pending = False
+                if self._shutting_down:
+                    return
+                self._start_runtime_openviking_waiter(
+                    endpoint=endpoint,
+                    status_callback=status_callback,
+                    warning_callback=warning_callback,
+                )
 
     def initialize(self, session_id: str, **kwargs) -> None:
         settings = _resolve_connection_settings(_load_hermes_openviking_config())
@@ -2176,6 +2292,11 @@ class OpenVikingMemoryProvider(MemoryProvider):
         self._account = settings["account"]
         self._user = settings["user"]
         self._agent = settings["agent"]
+        # Baseline established — subsequent accesses may refresh from env
+        # (#21130). Set here (not at the end of initialize) so an exception in
+        # the connection attempt below — swallowed by MemoryManager's guard —
+        # can't leave the provider silently stuck in never-refresh mode.
+        self._env_refresh_enabled = True
         self._session_id = session_id
         self._turn_count = 0
         hermes_home = str(kwargs.get("hermes_home") or "").strip()
@@ -2221,14 +2342,114 @@ class OpenVikingMemoryProvider(MemoryProvider):
             self._client = None
 
         if self._client:
+            self._conn_snapshot = (
+                self._endpoint, self._api_key, self._account, self._user, self._agent,
+            )
             self._recover_pending_sessions()
 
         # Register as the last active provider for atexit safety net
         global _last_active_provider
         _last_active_provider = self
 
+    def _ensure_client(self) -> Optional["_VikingClient"]:
+        """Return the active client, rebuilding it if the resolved config changed.
+
+        ``/reload`` only refreshes ``os.environ`` — the existing provider
+        instance is not re-initialized — so OPENVIKING_* values added to
+        ``~/.hermes/.env`` after startup never reach the live client and tools
+        keep running against stale auth until the user restarts hermes (#21130).
+
+        Re-resolve the connection settings on each access (same layering as
+        ``initialize``) and rebuild + health-check only when a value actually
+        changed; otherwise reuse the cached client so the hot path stays at one
+        dict comparison with zero network calls.
+        """
+        # Before initialize() runs there is no env baseline to refresh against;
+        # return whatever client the caller wired up (matches legacy behavior).
+        if not self._env_refresh_enabled:
+            return self._client
+
+        with self._client_refresh_lock:
+            return self._ensure_client_locked()
+
+    def _ensure_client_locked(self) -> Optional["_VikingClient"]:
+        """Resolve and publish one client/config state under the refresh lock."""
+        if self._shutting_down:
+            self._client = None
+            return None
+
+        settings = _resolve_connection_settings(_load_hermes_openviking_config())
+        endpoint = settings["endpoint"]
+        api_key = settings["api_key"]
+        account = settings["account"]
+        user = settings["user"]
+        agent = settings["agent"]
+        settings_key = (endpoint, api_key, account, user, agent)
+
+        config_unchanged = (
+            endpoint == getattr(self, "_endpoint", None)
+            and api_key == getattr(self, "_api_key", None)
+            and account == getattr(self, "_account", None)
+            and user == getattr(self, "_user", None)
+            and agent == getattr(self, "_agent", None)
+        )
+        if config_unchanged and self._client is not None:
+            return self._client
+        if config_unchanged:
+            with self._runtime_start_lock:
+                if (
+                    self._runtime_start_pending
+                    or (self._runtime_start_thread and self._runtime_start_thread.is_alive())
+                ):
+                    return self._client
+            # The last attempt at this exact config failed. Don't pay a
+            # network probe (3s timeout, under the refresh lock) on every
+            # access while the server stays down — retry after a cooldown or
+            # as soon as the resolved config changes.
+            failed = self._failed_refresh
+            if failed is not None:
+                failed_key, failed_at = failed
+                if (
+                    failed_key == settings_key
+                    and time.monotonic() - failed_at < _FAILED_CONFIG_RETRY_COOLDOWN_SECONDS
+                ):
+                    return None
+
+        self._endpoint = endpoint
+        self._api_key = api_key
+        self._account = account
+        self._user = user
+        self._agent = agent
+
+        try:
+            client = _VikingClient(
+                endpoint, api_key, account=account, user=user, agent=agent,
+            )
+        except ImportError:
+            logger.warning("httpx not installed — OpenViking plugin disabled")
+            self._client = None
+            return None
+
+        health_state, health_message = _classify_runtime_openviking_health(client, endpoint)
+        if health_state == "healthy":
+            self._client = client
+            self._conn_snapshot = settings_key
+            self._failed_refresh = None
+            return self._client
+        self._failed_refresh = (settings_key, time.monotonic())
+        if health_state == "responded":
+            logger.warning(
+                "%s OpenViking memory disabled; will retry on a later access "
+                "(after cooldown) or when the config changes.",
+                health_message,
+            )
+        else:  # unreachable
+            self._handle_runtime_openviking_unreachable()
+        self._client = None
+        return None
+
     def system_prompt_block(self) -> str:
-        if not self._client:
+        if not self._ensure_client():
             return ""
         # Provide brief info about the knowledge base
         try:
@@ -2278,7 +2499,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
     def prefetch(self, query: str, *, session_id: str = "") -> str:
         """Return recall context for this query/session."""
         query_text = _derive_openviking_user_text(query).strip()
-        if not self._client:
+        if not self._ensure_client():
             return ""
 
         effective_session_id = str(session_id or self._session_id or "").strip()
@@ -2424,6 +2645,18 @@ class OpenVikingMemoryProvider(MemoryProvider):
                 t.join(timeout=slice_left)
 
     def _new_client(self) -> _VikingClient:
+        # Read the connection identity as ONE tuple load: these builders run on
+        # background writer threads without _client_refresh_lock, and reading
+        # the five fields individually could observe a torn mix of old and new
+        # values mid-refresh (new endpoint + old api_key). The snapshot is only
+        # published after a successful health check; fall back to the field
+        # reads for legacy/hand-wired paths where no snapshot exists yet.
+        snapshot = self._conn_snapshot
+        if snapshot is not None:
+            endpoint, api_key, account, user, agent = snapshot
+            return _VikingClient(
+                endpoint, api_key, account=account, user=user, agent=agent,
+            )
         return _VikingClient(
             self._endpoint,
             self._api_key,
@@ -2849,17 +3082,23 @@ class OpenVikingMemoryProvider(MemoryProvider):
         client: Optional[_VikingClient] = None,
     ) -> str:
         query_text = (query or "").strip()
-        if not self._client or len(query_text) < _RECALL_QUERY_MIN_CHARS:
+        if len(query_text) < _RECALL_QUERY_MIN_CHARS:
+            return ""
+        if client is None:
+            if self._env_refresh_enabled:
+                client = self._ensure_client()
+            elif self._client is not None:
+                # Legacy/hand-wired path: no env baseline yet. Build from the
+                # cached identity, degrading to "" like the rest of prefetch.
+                try:
+                    client = self._new_client()
+                except Exception as e:
+                    logger.debug("OpenViking prefetch client build failed: %s", e)
+                    return ""
+        if client is None:
             return ""
 
         try:
-            client = client or _VikingClient(
-                self._endpoint,
-                self._api_key,
-                account=self._account,
-                user=self._user,
-                agent=self._agent,
-            )
             cfg = self._recall_config()
             candidate_limit = max(cfg["limit"] * 4, 20)
             deadline = time.monotonic() + cfg["timeout_seconds"]
@@ -3712,7 +3951,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
         messages: Optional[List[Dict[str, Any]]] = None,
     ) -> None:
         """Record the conversation turn in OpenViking's session (non-blocking)."""
-        if not self._client:
+        if not self._ensure_client():
             return
 
         user_content = _derive_openviking_user_text(user_content)
@@ -3863,7 +4102,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
         OpenViking automatically extracts 6 categories of memories:
         profile, preferences, entities, events, cases, and patterns.
         """
-        if not self._client:
+        if not self._ensure_client():
             return
 
         # Snapshot sid + turn count atomically against a concurrent sync_turn
@@ -3914,7 +4153,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
         ``_turn_count``.
         """
         new_id = str(new_session_id or "").strip()
-        if not new_id or not self._client:
+        if not new_id or not self._ensure_client():
             return
 
         rewound = bool(kwargs.get("rewound"))
@@ -3974,7 +4213,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
         metadata: Optional[Dict[str, Any]] = None,
     ) -> None:
         """Mirror successful built-in memory additions to OpenViking."""
-        if not self._client or action != "add" or not content:
+        if action != "add" or not content or not self._ensure_client():
             return
 
         subdir = _MEMORY_WRITE_TARGET_SUBDIR_MAP.get(target, _DEFAULT_MEMORY_SUBDIR)
@@ -3982,10 +4221,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
 
         def _write():
             try:
-                client = _VikingClient(
-                    self._endpoint, self._api_key,
-                    account=self._account, user=self._user, agent=self._agent,
-                )
+                client = self._new_client()
                 client.post("/api/v1/content/write", {
                     "uri": uri,
                     "content": content,
@@ -4019,7 +4255,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
         ]
 
     def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
-        if not self._client:
+        if not self._ensure_client():
             return tool_error("OpenViking server not connected")
 
         try:
@@ -4052,6 +4288,12 @@ class OpenVikingMemoryProvider(MemoryProvider):
             deferred_workers = list(self._deferred_commit_threads)
         with self._memory_write_lock:
             memory_write_workers = list(self._memory_write_threads)
+        # The runtime-autostart waiter is a tracked daemon thread that blocks on
+        # network health probes; it must be joined too, or it can be left alive
+        # at interpreter exit (SIGABRT at Py_FinalizeEx). Setting _shutting_down
+        # above makes its health-wait loop bail out promptly so the join lands.
+        with self._runtime_start_lock:
+            runtime_start_thread = self._runtime_start_thread
         for t in all_workers:
             if t.is_alive():
                 t.join(timeout=5.0)
@@ -4061,6 +4303,8 @@ class OpenVikingMemoryProvider(MemoryProvider):
         for t in memory_write_workers:
             if t.is_alive():
                 t.join(timeout=5.0)
+        if runtime_start_thread is not None and runtime_start_thread.is_alive():
+            runtime_start_thread.join(timeout=5.0)
         # Clear atexit reference so it doesn't double-commit.
         global _last_active_provider
         if _last_active_provider is self:

scripts/release.py

@@ -81,6 +81,7 @@ LEGACY_AUTHOR_MAP = {
     "marceloparra.hm@gmail.com": "marcelohildebrand",  # PR #42346 salvage (lmstudio: JIT load mode)
     "qlskssk@gmail.com": "Soju06",  # agent turn-latency perf PRs
     "m.guttmann@journaway.com": "mguttmann",  # PR #63738 salvage (Anthropic setup-token pool auth normalization)
+    "wangzhe00zju@gmail.com": "flyingdoubleG",  # PR #18166 salvage (memory-provider tools honor disabled_toolsets in initial and MCP-refresh injection)
     "VrtxOmega@pm.me": "VrtxOmega",  # PR #43809 salvage (desktop: WSL folder-picker path bridge)
     "gn00742754@gmail.com": "SemonCat",  # PR #56786 salvage (Slack Agent View manifests and Assistant APIs)
     "KCAYAAI@users.noreply.github.com": "KCAYAAI",  # PR #62248 partial salvage (resume typing after clarify reply)
@@ -1995,6 +1996,8 @@ LEGACY_AUTHOR_MAP = {
     "andrewdmwalker@gmail.com": "capt-marbles",  # PR #38440 salvage (resolve xAI OAuth credentials across profiles; #43589)
     "infinitycrew39@gmail.com": "infinitycrew39",  # PR #47945 salvage (scope langfuse trace state by turn/request ids; #48292)
     "eurekaxun@163.com": "huangxun375-stack",  # PR #37251 / #48894 structured OpenViking sync
+    "koshaji@gmail.com": "koshaji",  # PR #49832 salvage (OpenViking runtime autostart shutdown drain)
+    "thor753@foxmail.com": "wgd753",  # PR #59454 salvage (OpenViking trusted-mode retry matching)
     "218421507+Sahil-SS9@users.noreply.github.com": "Sahil-SS9",  # PR #48466/#44919/#44909/#42209 salvage (cron/checkpoint/kanban/skill)
     "mango001@126.com": "max-chen",  # PR #51194 salvage (single-pass list_profiles alias map + skill-count cache; #54751)
     # v0.17.0 additions

tests/agent/test_memory_provider.py

@@ -1379,17 +1379,19 @@ class TestMemoryToolToolsetGate:
     These tests exercise the shared gate used by agent init and ACP refreshes.
     The gate condition is:
 
+        disabled_toolsets includes memory → skip injection
         enabled_toolsets is None        → no filter, inject (backward compat)
         selected toolsets include memory → user opted in, inject
         otherwise (incl. [])            → skip injection
     """
 
     @staticmethod
-    def _run_memory_injection(enabled_toolsets, memory_manager):
+    def _run_memory_injection(enabled_toolsets, memory_manager, disabled_toolsets=None):
         """Run the shared memory-tool injection helper against a fake agent."""
         fake_agent = SimpleNamespace(
             _memory_manager=memory_manager,
             enabled_toolsets=enabled_toolsets,
+            disabled_toolsets=disabled_toolsets,
             tools=[],
             valid_tool_names=set(),
         )
@@ -1426,6 +1428,18 @@ class TestMemoryToolToolsetGate:
         assert "hindsight_recall" in names
         assert any(t["function"]["name"] == "hindsight_recall" for t in tools)
 
+    @pytest.mark.parametrize("enabled_toolsets", [None, ["memory"], ["all"], ["hermes-acp"]])
+    def test_disabled_memory_toolset_blocks_injection(self, enabled_toolsets):
+        """An explicit memory disable wins over default or composite enablement."""
+        mgr = self._mgr_with_tools("hindsight_recall")
+        tools, names = self._run_memory_injection(
+            enabled_toolsets,
+            mgr,
+            disabled_toolsets=["memory"],
+        )
+        assert tools == []
+        assert names == set()
+
     def test_empty_toolsets_blocks_injection(self):
         """`platform_toolsets: telegram: []` must suppress memory tools. (#5544)"""
         mgr = self._mgr_with_tools("fact_store")

tests/cron/test_scheduler_provider.py

@@ -691,3 +691,103 @@ class TestGuardJobCredentialExfil:
 
         monkeypatch.setattr(ct, "_validate_cron_base_url", _boom)
         assert _guard_job_credential_exfil({"id": "j8", "provider": "anthropic"}) is None
+
+
+# ── Multiplex profiles: cron per secondary profile (issue #69377) ─────────
+
+
+def test_multiplex_ticker_ticks_each_profile_once(tmp_path, monkeypatch):
+    """The multiplex cron scheduler calls tick() once per profile home,
+    scoped via use_cron_store, so secondary-profile jobs actually fire
+    instead of languishing in an unticked store."""
+    from cron.scheduler_provider import InProcessCronScheduler
+
+    # Set up two profile directories.
+    p1 = tmp_path / "default"
+    p2 = tmp_path / "home-ops"
+    for d in (p1, p2):
+        (d / "cron").mkdir(parents=True)
+
+    profile_homes = [("default", p1), ("home-ops", p2)]
+
+    # Count tick() calls — should be called once per profile per iteration.
+    tick_count: list[int] = []
+
+    def _tracking_tick(*args, **kwargs):
+        tick_count.append(1)
+        return 0
+
+    stop = threading.Event()
+    prov = InProcessCronScheduler()
+
+    with patch("cron.scheduler.tick", side_effect=_tracking_tick), \
+         patch("cron.jobs.record_ticker_heartbeat", lambda **kw: None):
+        t = threading.Thread(
+            target=prov.start,
+            args=(stop,),
+            kwargs={"interval": 0, "profile_homes": profile_homes},
+            daemon=True,
+        )
+        t.start()
+        # Wait for at least len(profile_homes) tick calls (one full cycle).
+        deadline = time.monotonic() + 10
+        while len(tick_count) < len(profile_homes) and time.monotonic() < deadline:
+            time.sleep(0.005)
+        # Give one more cycle to ensure it keeps ticking.
+        deadline = time.monotonic() + 3
+        while len(tick_count) < len(profile_homes) * 2 and time.monotonic() < deadline:
+            time.sleep(0.005)
+        stop.set()
+        t.join(timeout=5)
+
+    assert not t.is_alive()
+    # The ticker called tick() at least once per profile per iteration.
+    # With 2 profiles and multiple iterations, we should have seen at least 2 calls.
+    assert len(tick_count) >= len(profile_homes), \
+        f"Expected >= {len(profile_homes)} tick calls, got {len(tick_count)}"
+
+
+def test_multiplex_heartbeat_scoped_per_profile(tmp_path, monkeypatch):
+    """record_ticker_heartbeat is scoped to each profile's store under
+    multiplex, so 'hermes cron status' can report liveness per profile."""
+    from cron.scheduler_provider import InProcessCronScheduler
+    from cron.jobs import record_ticker_heartbeat as _real_heartbeat
+
+    p_default = tmp_path / "default"
+    p_sec = tmp_path / "home-ops"
+    for d in (p_default, p_sec):
+        (d / "cron").mkdir(parents=True)
+    profile_homes = [("default", p_default), ("home-ops", p_sec)]
+
+    beat_log: list[str] = []
+
+    def _track_beat(*, success=False):
+        beat_log.append(str(success))
+        # Write the real heartbeat files so we can check them after.
+        _real_heartbeat(success=success)
+
+    stop = threading.Event()
+    prov = InProcessCronScheduler()
+
+    with patch("cron.scheduler.tick", return_value=0), \
+         patch("cron.jobs.record_ticker_heartbeat", side_effect=_track_beat):
+        t = threading.Thread(
+            target=prov.start,
+            args=(stop,),
+            kwargs={"interval": 0, "profile_homes": profile_homes},
+            daemon=True,
+        )
+        t.start()
+        deadline = time.monotonic() + 10
+        # Wait for at least 2 tick iterations over all profiles (2 profiles).
+        while len(beat_log) < len(profile_homes) * 2 and time.monotonic() < deadline:
+            time.sleep(0.005)
+        stop.set()
+        t.join(timeout=5)
+
+    assert not t.is_alive()
+    # Every profile should have a heartbeat file.
+    assert (p_default / "cron" / "ticker_heartbeat").exists(), \
+        "default profile heartbeat file missing"
+    assert (p_sec / "cron" / "ticker_heartbeat").exists(), \
+        "secondary profile heartbeat file missing"

tests/gateway/relay/test_relay_passthrough.py

@@ -118,7 +118,11 @@ async def test_discord_interaction_routes_through_handle_message(adapter, monkey
 
     assert len(seen) == 1
     ev = seen[0]
-    assert ev.text == "summarize"
+    # APPLICATION_COMMAND interactions are normalized to a leading-slash
+    # command (the dispatcher's contract), not the bare registered name.
+    assert ev.text == "/summarize"
+    assert ev.is_command() is True
+    assert ev.get_command() == "summarize"
     assert ev.source.chat_id == "chan-9"
     assert ev.source.scope_id == "guild-7"
     assert ev.source.user_id == "user-3"
@@ -151,6 +155,147 @@ async def test_message_component_interaction_uses_custom_id(adapter, monkeypatch
     await stub.push_passthrough(fwd)
     assert len(seen) == 1
     assert seen[0].text == "approve_btn"
+    # Component interactions stay plain text — only APPLICATION_COMMANDs are
+    # normalized to slash commands.
+    assert seen[0].is_command() is False
+
+
+@pytest.mark.asyncio
+async def test_application_command_no_options_is_slash_command(adapter, monkeypatch):
+    """/new with no options -> text '/new', dispatched as a COMMAND event."""
+    from gateway.platforms.base import MessageType
+
+    await adapter.connect()
+    stub = adapter._transport
+    seen = []
+
+    async def fake_handle(event):
+        seen.append(event)
+
+    monkeypatch.setattr(adapter, "handle_message", fake_handle)
+    fwd = _interaction_forward(
+        {
+            "id": "i-new",
+            "type": 2,
+            "channel_id": "c3",
+            "guild_id": "g3",
+            "data": {"name": "new"},
+            "member": {"user": {"id": "u3", "username": "ben"}},
+        }
+    )
+    await stub.push_passthrough(fwd)
+    assert len(seen) == 1
+    ev = seen[0]
+    assert ev.text == "/new"
+    assert ev.message_type == MessageType.COMMAND
+    # Behavior contract: the dispatcher must recognize this as command 'new'.
+    assert ev.is_command() is True
+    assert ev.get_command() == "new"
+    assert ev.get_command_args() == ""
+
+
+@pytest.mark.asyncio
+async def test_application_command_scalar_options_append_values(adapter, monkeypatch):
+    """Scalar options append their values space-separated: /model gpt-x."""
+    await adapter.connect()
+    stub = adapter._transport
+    seen = []
+
+    async def fake_handle(event):
+        seen.append(event)
+
+    monkeypatch.setattr(adapter, "handle_message", fake_handle)
+    fwd = _interaction_forward(
+        {
+            "id": "i-model",
+            "type": 2,
+            "channel_id": "c4",
+            "guild_id": "g4",
+            "data": {
+                "name": "model",
+                "options": [{"name": "name", "type": 3, "value": "gpt-x"}],
+            },
+            "member": {"user": {"id": "u4", "username": "ben"}},
+        }
+    )
+    await stub.push_passthrough(fwd)
+    assert len(seen) == 1
+    ev = seen[0]
+    assert ev.text == "/model gpt-x"
+    assert ev.is_command() is True
+    assert ev.get_command() == "model"
+    assert ev.get_command_args() == "gpt-x"
+
+
+@pytest.mark.asyncio
+async def test_application_command_subcommand_nesting_renders_names_then_values(
+    adapter, monkeypatch
+):
+    """SUB_COMMAND (type 1) appends its name, then recurses into its options."""
+    await adapter.connect()
+    stub = adapter._transport
+    seen = []
+
+    async def fake_handle(event):
+        seen.append(event)
+
+    monkeypatch.setattr(adapter, "handle_message", fake_handle)
+    fwd = _interaction_forward(
+        {
+            "id": "i-sub",
+            "type": 2,
+            "channel_id": "c5",
+            "guild_id": "g5",
+            "data": {
+                "name": "skill",
+                "options": [
+                    {
+                        "name": "run",
+                        "type": 1,  # SUB_COMMAND
+                        "options": [{"name": "target", "type": 3, "value": "deploy"}],
+                    }
+                ],
+            },
+            "member": {"user": {"id": "u5", "username": "ben"}},
+        }
+    )
+    await stub.push_passthrough(fwd)
+    assert len(seen) == 1
+    ev = seen[0]
+    assert ev.text == "/skill run deploy"
+    assert ev.is_command() is True
+    assert ev.get_command() == "skill"
+    assert ev.get_command_args() == "run deploy"
+
+
+@pytest.mark.asyncio
+async def test_ping_interaction_produces_no_command(adapter, monkeypatch):
+    """A PING (type 1) body — never normally forwarded — stays empty TEXT, not
+    a phantom command."""
+    from gateway.platforms.base import MessageType
+
+    await adapter.connect()
+    stub = adapter._transport
+    seen = []
+
+    async def fake_handle(event):
+        seen.append(event)
+
+    monkeypatch.setattr(adapter, "handle_message", fake_handle)
+    fwd = _interaction_forward(
+        {
+            "id": "i-ping",
+            "type": 1,  # PING
+            "channel_id": "c6",
+            "user": {"id": "u6", "username": "ben"},
+        }
+    )
+    await stub.push_passthrough(fwd)
+    assert len(seen) == 1
+    ev = seen[0]
+    assert ev.text == ""
+    assert ev.message_type == MessageType.TEXT
+    assert ev.is_command() is False
 
 
 @pytest.mark.asyncio

tests/gateway/relay/test_relay_per_platform_caps.py

@@ -1,252 +0,0 @@
-"""Per-platform capability descriptors on the relay (multi-platform Phase 1.5).
-
-The bug class: one relay adapter fronts N platforms on one WS, but the
-capability surface (``MAX_MESSAGE_LENGTH`` / ``message_len_fn``) was a SCALAR
-from whichever descriptor resolved the handshake — so a Discord chat on a
-gateway whose primary identity was Telegram inherited Telegram's 4,096-char
-cap and over-sent into Discord's 2,000-char API 400 (observed live: 2,543 and
-2,641-char sends rejected).
-
-Covers:
-  - the transport accumulating one descriptor per platform (first = session
-    default, later frames must NOT overwrite it),
-  - the map resetting on a re-dial,
-  - RelayAdapter.max_message_length_for_chat / message_len_fn_for_chat
-    resolving from the chat's inbound platform,
-  - fallback to the scalar descriptor for unknown chats / transports without
-    the map,
-  - the stream consumer's _raw_message_limit honoring the per-chat cap.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import json
-from typing import Any, Dict, List, Optional
-
-import pytest
-
-from gateway.config import Platform, PlatformConfig
-from gateway.platforms.base import MessageEvent, MessageType
-from gateway.relay.adapter import RelayAdapter
-from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
-from gateway.session import SessionSource
-
-from tests.gateway.relay.stub_connector import StubConnector
-
-
-def _descriptor(platform: str, max_len: int, len_unit: str = "chars") -> CapabilityDescriptor:
-    return CapabilityDescriptor(
-        contract_version=CONTRACT_VERSION,
-        platform=platform,
-        label=platform.title(),
-        max_message_length=max_len,
-        supports_draft_streaming=False,
-        supports_edit=True,
-        supports_threads=False,
-        markdown_dialect="plain",
-        len_unit=len_unit,
-    )
-
-
-DISCORD = _descriptor("discord", 2000)
-TELEGRAM = _descriptor("telegram", 4096, len_unit="utf16")
-
-
-class MultiDescriptorStub(StubConnector):
-    """StubConnector extended with the per-platform descriptor map."""
-
-    def __init__(self, primary: CapabilityDescriptor, *others: CapabilityDescriptor) -> None:
-        super().__init__(primary)
-        self._by_platform = {d.platform: d for d in (primary, *others)}
-
-    def descriptor_for_platform(self, platform: str) -> Optional[CapabilityDescriptor]:
-        return self._by_platform.get(platform)
-
-
-async def _push(stub: StubConnector, platform: Platform, chat_id: str) -> None:
-    await stub.push_inbound(
-        MessageEvent(
-            text="hi",
-            message_type=MessageType.TEXT,
-            source=SessionSource(
-                platform=platform, chat_id=chat_id, chat_type="dm", user_id="u-1"
-            ),
-        )
-    )
-
-
-# ───────────────────── transport descriptor accumulation ─────────────────────
-
-
-def _make_transport():
-    from gateway.relay.ws_transport import WebSocketRelayTransport
-
-    return WebSocketRelayTransport(
-        "wss://connector.example/relay",
-        "telegram",
-        "bot-9",
-        identities=[("telegram", "bot-9"), ("discord", "app-1")],
-    )
-
-
-@pytest.mark.asyncio
-async def test_transport_accumulates_descriptors_first_wins_as_default():
-    """One descriptor frame per hello: the map holds each platform's, and the
-    scalar `_descriptor` (the handshake result / session default) stays the
-    FIRST one — the regression was last-writer-wins across platforms."""
-    t = _make_transport()
-    loop = asyncio.get_running_loop()
-    t._descriptor_ready = loop.create_future()
-
-    frame = {"type": "descriptor", "descriptor": TELEGRAM.__dict__}
-    await t._handle_frame(json.dumps(frame))
-    frame2 = {"type": "descriptor", "descriptor": DISCORD.__dict__}
-    await t._handle_frame(json.dumps(frame2))
-
-    # Per-platform map has both.
-    assert t.descriptor_for_platform("telegram").max_message_length == 4096
-    assert t.descriptor_for_platform("discord").max_message_length == 2000
-    assert t.descriptor_for_platform("slack") is None
-    # The session default is the FIRST (primary identity) — NOT overwritten.
-    assert t._descriptor.platform == "telegram"
-    assert (await t.handshake()).platform == "telegram"
-
-
-@pytest.mark.asyncio
-async def test_transport_descriptor_map_resets_on_redial(monkeypatch):
-    """A re-dial starts a fresh handshake generation: stale per-platform
-    descriptors must not survive into the new connection."""
-    t = _make_transport()
-    loop = asyncio.get_running_loop()
-    t._descriptor_ready = loop.create_future()
-    await t._handle_frame(json.dumps({"type": "descriptor", "descriptor": DISCORD.__dict__}))
-    assert t.descriptor_for_platform("discord") is not None
-
-    # Simulate _dial_and_start's reset preamble without a real socket.
-    class _FakeWs:
-        async def close(self):  # pragma: no cover - not called
-            pass
-
-    sent: List[str] = []
-
-    async def _fake_connect(url, **kwargs):
-        return _FakeWs()
-
-    async def _fake_send(payload):
-        sent.append(payload)
-
-    import gateway.relay.ws_transport as wst
-
-    monkeypatch.setattr(wst, "websockets", type("M", (), {"connect": staticmethod(_fake_connect)}))
-    monkeypatch.setattr(t, "_send", _fake_send)
-    monkeypatch.setattr(
-        t, "_read_loop", lambda: asyncio.sleep(0)
-    )  # substitute a no-op coroutine factory
-    await t._dial_and_start()
-
-    assert t.descriptor_for_platform("discord") is None
-    assert t._descriptor is None
-
-
-# ───────────────────── adapter per-chat capability surface ─────────────────────
-
-
-@pytest.mark.asyncio
-async def test_adapter_resolves_per_chat_limits_from_inbound_platform():
-    """The live bug shape: primary identity Telegram (4096), a Discord chat on
-    the same adapter must get Discord's 2000 — not Telegram's scalar."""
-    stub = MultiDescriptorStub(TELEGRAM, DISCORD)
-    stub._identities = [("telegram", "bot-9"), ("discord", "app-1")]
-    adapter = RelayAdapter(PlatformConfig(), TELEGRAM, transport=stub)
-    await adapter.connect()
-
-    await _push(stub, Platform.DISCORD, "dc-1")
-    await _push(stub, Platform.TELEGRAM, "tg-1")
-
-    # Scalar surface still the primary's (back-compat).
-    assert adapter.MAX_MESSAGE_LENGTH == 4096
-    # Per-chat: each chat resolves its own platform's cap.
-    assert adapter.max_message_length_for_chat("dc-1") == 2000
-    assert adapter.max_message_length_for_chat("tg-1") == 4096
-    # Length unit follows the chat too: Telegram utf16, Discord codepoints.
-    surrogate = "\U0001f600"  # 2 UTF-16 units, 1 codepoint
-    assert adapter.message_len_fn_for_chat("tg-1")(surrogate) == 2
-    assert adapter.message_len_fn_for_chat("dc-1")(surrogate) == 1
-
-
-@pytest.mark.asyncio
-async def test_adapter_unknown_chat_falls_back_to_scalar_descriptor():
-    stub = MultiDescriptorStub(TELEGRAM, DISCORD)
-    adapter = RelayAdapter(PlatformConfig(), TELEGRAM, transport=stub)
-    await adapter.connect()
-    # Never saw inbound for this chat — platform unknown -> scalar descriptor.
-    assert adapter.max_message_length_for_chat("never-seen") == 4096
-
-
-@pytest.mark.asyncio
-async def test_adapter_transport_without_map_falls_back_to_scalar():
-    """A plain StubConnector (no descriptor_for_platform) — e.g. an older or
-    test transport — must keep the scalar behavior, not raise."""
-    stub = StubConnector(TELEGRAM)
-    adapter = RelayAdapter(PlatformConfig(), TELEGRAM, transport=stub)
-    await adapter.connect()
-    await _push(stub, Platform.DISCORD, "dc-1")
-    assert adapter.max_message_length_for_chat("dc-1") == 4096
-
-
-def test_native_adapter_defaults_scalar():
-    """BasePlatformAdapter's default per-chat hooks mirror the scalar surface
-    (native adapters are single-platform; nothing changes for them)."""
-    from gateway.platforms.base import BasePlatformAdapter
-
-    class _Native(BasePlatformAdapter):
-        MAX_MESSAGE_LENGTH = 1234
-
-        def __init__(self):  # bypass Base __init__ plumbing
-            pass
-
-        async def connect(self, *, is_reconnect: bool = False) -> bool:  # pragma: no cover
-            return True
-
-        async def disconnect(self) -> None:  # pragma: no cover
-            pass
-
-        async def send(self, chat_id, content, reply_to=None, metadata=None):  # pragma: no cover
-            raise NotImplementedError
-
-        async def get_chat_info(self, chat_id):  # pragma: no cover
-            return {}
-
-    a = _Native()
-    assert a.max_message_length_for_chat("any") == 1234
-    assert a.message_len_fn_for_chat("any") is a.message_len_fn
-
-
-# ───────────────────── stream consumer integration ─────────────────────
-
-
-@pytest.mark.asyncio
-async def test_stream_consumer_raw_limit_uses_per_chat_cap():
-    """_raw_message_limit resolves the CHAT's platform cap on a relay adapter:
-    a Discord chat splits at 2000 even when the scalar descriptor says 4096.
-    Without the fix this returned 4096 and a 2,543-char reply reached Discord
-    whole -> HTTP 400."""
-    from gateway.stream_consumer import GatewayStreamConsumer
-
-    stub = MultiDescriptorStub(TELEGRAM, DISCORD)
-    stub._identities = [("telegram", "bot-9"), ("discord", "app-1")]
-    adapter = RelayAdapter(PlatformConfig(), TELEGRAM, transport=stub)
-    await adapter.connect()
-    await _push(stub, Platform.DISCORD, "dc-1")
-    await _push(stub, Platform.TELEGRAM, "tg-1")
-
-    dc = GatewayStreamConsumer.__new__(GatewayStreamConsumer)
-    dc.adapter = adapter
-    dc.chat_id = "dc-1"
-    assert dc._raw_message_limit() == 2000
-
-    tg = GatewayStreamConsumer.__new__(GatewayStreamConsumer)
-    tg.adapter = adapter
-    tg.chat_id = "tg-1"
-    assert tg._raw_message_limit() == 4096

tests/hermes_cli/test_memory_setup.py

@@ -196,3 +196,37 @@ def test_cmd_setup_generic_choice_cancel_writes_nothing(tmp_path, monkeypatch):
     save_config.assert_not_called()
     provider.save_config.assert_not_called()
     assert not (tmp_path / ".env").exists()
+
+
+def test_write_env_vars_strips_line_separators_and_nul(tmp_path):
+    """A pasted secret with embedded CR/LF/NUL must not inject an extra
+    KEY=VALUE line into .env (mirrors the openviking plugin's writer)."""
+    env_path = tmp_path / ".env"
+
+    memory_setup._write_env_vars(
+        env_path,
+        {"PROVIDER_API_KEY": "good\nINJECTED_KEY=attacker\r\u2028\x00tail"},
+    )
+
+    lines = env_path.read_text(encoding="utf-8").splitlines()
+    assert lines == ["PROVIDER_API_KEY=goodINJECTED_KEY=attackertail"]
+    parsed = dict(line.split("=", 1) for line in lines if "=" in line)
+    assert set(parsed) == {"PROVIDER_API_KEY"}
+
+
+def test_write_env_vars_strips_newlines_when_updating_existing_key(tmp_path):
+    env_path = tmp_path / ".env"
+    env_path.write_text("PROVIDER_API_KEY=old\nKEEP=1\n", encoding="utf-8")
+
+    memory_setup._write_env_vars(env_path, {"PROVIDER_API_KEY": "new\r\nROGUE=1"})
+
+    lines = env_path.read_text(encoding="utf-8").splitlines()
+    assert "PROVIDER_API_KEY=newROGUE=1" in lines
+    assert "KEEP=1" in lines
+    assert all(not line.startswith("ROGUE=") for line in lines)
+
+
+def test_write_env_vars_plain_value_roundtrips(tmp_path):
+    env_path = tmp_path / ".env"
+    memory_setup._write_env_vars(env_path, {"PROVIDER_API_KEY": "sk-plain-1234"})
+    assert env_path.read_text(encoding="utf-8") == "PROVIDER_API_KEY=sk-plain-1234\n"

tests/openviking_plugin/test_openviking.py

@@ -2,6 +2,7 @@
 
 import json
 import threading
+import time
 from http.server import BaseHTTPRequestHandler, HTTPServer
 from typing import Any, cast
 from urllib.parse import parse_qs, urlparse
@@ -1446,3 +1447,558 @@ class TestOpenVikingMemoryUriBuilder:
             assert f"/memories/{subdir}/mem_" in uri, (
                 f"subdir '{subdir}' not placed correctly in URI: {uri}"
             )
+
+
+# ===================================================================
+# Issue #21130 — OPENVIKING_* not reloaded after /reload
+# ===================================================================
+
+
+class TestEnsureClientReloadsEnv:
+    """Verify /reload picks up new OPENVIKING_* values without a restart (#21130)."""
+
+    def test_ensure_client_rebuilds_when_api_key_changes(self, monkeypatch):
+        constructions = []
+
+        class _StubClient:
+            def __init__(self, endpoint, api_key, account="", user="", agent="hermes"):
+                constructions.append({"endpoint": endpoint, "api_key": api_key,
+                                      "account": account, "user": user, "agent": agent})
+                self.endpoint, self.api_key = endpoint, api_key
+                self.account, self.user, self.agent = account, user, agent
+
+            def health(self):
+                return True
+
+        monkeypatch.setattr("plugins.memory.openviking._VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://srv:31933")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "")
+
+        provider = OpenVikingMemoryProvider()
+        provider._env_refresh_enabled = True
+        first = provider._ensure_client()
+        assert first is not None
+        assert first.api_key == ""
+        assert len(constructions) == 1
+
+        # Same env on second call — must reuse cached client (no rebuild).
+        assert provider._ensure_client() is first
+        assert len(constructions) == 1
+
+        # Simulate /reload: env now carries the new API key.
+        monkeypatch.setenv("OPENVIKING_API_KEY", "sk-fresh")
+        rebuilt = provider._ensure_client()
+        assert rebuilt is not None
+        assert rebuilt is not first
+        assert rebuilt.api_key == "sk-fresh"
+        assert len(constructions) == 2
+
+    def test_ensure_client_rebuilds_when_endpoint_changes(self, monkeypatch):
+        builds = []
+
+        class _StubClient:
+            def __init__(self, endpoint, api_key, **kw):
+                builds.append(endpoint)
+                self.endpoint = endpoint
+
+            def health(self):
+                return True
+
+        monkeypatch.setattr("plugins.memory.openviking._VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://a")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "key")
+
+        provider = OpenVikingMemoryProvider()
+        provider._env_refresh_enabled = True
+        provider._ensure_client()
+        provider._ensure_client()  # cached
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://b")
+        provider._ensure_client()  # rebuilds
+        assert builds == ["http://a", "http://b"]
+
+    def test_prefetch_rebuilds_client_when_api_key_changes(self, monkeypatch):
+        posts = []
+
+        class _StubClient:
+            def __init__(self, endpoint, api_key, **kw):
+                self.endpoint = endpoint
+                self.api_key = api_key
+
+            def health(self):
+                return True
+
+            def post(self, path, payload=None, **kwargs):
+                posts.append((self.api_key, path, payload or {}))
+                return {
+                    "result": {
+                        "memories": [
+                            {
+                                "uri": "viking://user/default/memories/pref.md",
+                                "abstract": f"memory from {self.api_key or 'anonymous'}",
+                                "score": 0.9,
+                                "level": 2,
+                            }
+                        ],
+                        "resources": [],
+                    }
+                }
+
+        monkeypatch.setattr("plugins.memory.openviking._VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://srv:31933")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "sk-old")
+        monkeypatch.setenv("OPENVIKING_RECALL_PREFER_ABSTRACT", "true")
+
+        provider = OpenVikingMemoryProvider()
+        provider._env_refresh_enabled = True
+        provider._session_id = "session-1"
+
+        first = provider.prefetch("What should OpenViking recall?", session_id="session-1")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "sk-fresh")
+        second = provider.prefetch("What should OpenViking recall?", session_id="session-1")
+
+        assert "memory from sk-old" in first
+        assert "memory from sk-fresh" in second
+        assert [call[0] for call in posts] == ["sk-old", "sk-fresh"]
+        assert [call[1] for call in posts] == ["/api/v1/search/search", "/api/v1/search/search"]
+        assert posts[0][2]["limit"] == posts[1][2]["limit"]
+        assert "top_k" not in posts[0][2]
+
+    def test_ensure_client_returns_none_when_health_fails(self, monkeypatch):
+        class _StubClient:
+            def __init__(self, *a, **kw):
+                pass
+
+            def health(self):
+                return False
+
+        monkeypatch.setattr("plugins.memory.openviking._VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://dead")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "")
+
+        provider = OpenVikingMemoryProvider()
+        provider._env_refresh_enabled = True
+        assert provider._ensure_client() is None
+        assert provider._client is None
+
+    def test_handle_tool_call_reconnects_after_startup_health_failure(self, monkeypatch):
+        instances = []
+
+        class _StubClient:
+            def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+                self.endpoint = endpoint
+                self.api_key = api_key
+                self.account = account
+                self.user = user
+                self.agent = agent
+                self.index = len(instances)
+                self.posts = []
+                instances.append(self)
+
+            def health(self):
+                return self.index > 0
+
+            def post(self, path, payload=None, **kwargs):
+                self.posts.append((path, payload or {}))
+                return {"result": {"written_bytes": 11}}
+
+        monkeypatch.setattr("plugins.memory.openviking._VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "https://openviking.example")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "sk-test")
+
+        provider = OpenVikingMemoryProvider()
+        provider.initialize("session-1")
+
+        assert provider._client is None
+
+        out = json.loads(provider.handle_tool_call(
+            "viking_remember",
+            {"content": "stable fact"},
+        ))
+
+        assert out["status"] == "stored"
+        assert len(instances) == 2
+        assert instances[1].posts[0][0] == "/api/v1/content/write"
+        assert instances[1].posts[0][1]["content"] == "stable fact"
+        assert instances[1].posts[0][1]["mode"] == "create"
+        assert instances[1].posts[0][1]["uri"].startswith(
+            "viking://user/peers/hermes/memories/"
+        )
+
+    def test_concurrent_refresh_does_not_return_stale_client(self, monkeypatch):
+        refresh_entered = threading.Event()
+        release_refresh = threading.Event()
+
+        class _StubClient:
+            def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+                self.endpoint = endpoint
+                self.api_key = api_key
+
+            def health(self):
+                if self.endpoint == "https://new.example":
+                    refresh_entered.set()
+                    assert release_refresh.wait(2.0)
+                    return False
+                return True
+
+        monkeypatch.setattr(openviking_plugin, "_VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "https://new.example")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "sk-new")
+        monkeypatch.delenv("OPENVIKING_ACCOUNT", raising=False)
+        monkeypatch.delenv("OPENVIKING_USER", raising=False)
+        monkeypatch.delenv("OPENVIKING_AGENT", raising=False)
+
+        provider = OpenVikingMemoryProvider()
+        stale_client = _StubClient("https://old.example", "sk-old")
+        provider._endpoint = stale_client.endpoint
+        provider._api_key = stale_client.api_key
+        provider._account = ""
+        provider._user = ""
+        provider._agent = "hermes"
+        provider._client = stale_client
+        provider._env_refresh_enabled = True
+
+        results = {}
+        errors = []
+        second_started = threading.Event()
+        second_done = threading.Event()
+
+        def refresh_client(name, *, started=None, done=None):
+            if started is not None:
+                started.set()
+            try:
+                results[name] = provider._ensure_client()
+            except BaseException as exc:  # pragma: no cover - surfaced below
+                errors.append(exc)
+            finally:
+                if done is not None:
+                    done.set()
+
+        first = threading.Thread(target=refresh_client, args=("first",))
+        first.start()
+        assert refresh_entered.wait(2.0)
+
+        second = threading.Thread(
+            target=refresh_client,
+            args=("second",),
+            kwargs={"started": second_started, "done": second_done},
+        )
+        second.start()
+        assert second_started.wait(2.0)
+        completed_during_refresh = second_done.wait(0.2)
+
+        release_refresh.set()
+        first.join(timeout=2.0)
+        second.join(timeout=2.0)
+
+        assert not first.is_alive()
+        assert not second.is_alive()
+        assert errors == []
+        assert completed_during_refresh is False
+        assert results == {"first": None, "second": None}
+        assert all(client is not stale_client for client in results.values())
+
+    def test_handle_tool_call_after_reload_to_local_endpoint_starts_runtime_recovery(
+        self,
+        tmp_path,
+        monkeypatch,
+    ):
+        from hermes_cli import config as hermes_config
+
+        known_hermes_env = set(hermes_config.OPTIONAL_ENV_VARS) | hermes_config._EXTRA_ENV_KEYS
+        openviking_tenant_env = {
+            "OPENVIKING_ENDPOINT",
+            "OPENVIKING_API_KEY",
+            "OPENVIKING_ACCOUNT",
+            "OPENVIKING_USER",
+            "OPENVIKING_AGENT",
+        }
+        for key in known_hermes_env | openviking_tenant_env:
+            monkeypatch.delenv(key, raising=False)
+
+        hermes_home = tmp_path / "hermes-home"
+        hermes_home.mkdir()
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+        env_path = hermes_home / ".env"
+        env_path.write_text(
+            "OPENVIKING_ENDPOINT=https://openviking.example\n"
+            "OPENVIKING_API_KEY=sk-old\n",
+            encoding="utf-8",
+        )
+        assert hermes_config.reload_env() >= 1
+
+        class _StubClient:
+            def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+                self.endpoint = endpoint
+                self.api_key = api_key
+                self.posts = []
+
+            def health(self):
+                return self.endpoint == "https://openviking.example"
+
+            def post(self, path, payload=None, **kwargs):
+                self.posts.append((path, payload or {}))
+                return {"result": {"written_bytes": 11}}
+
+        monkeypatch.setattr(openviking_plugin, "_VikingClient", _StubClient)
+        provider = OpenVikingMemoryProvider()
+        provider.initialize("session-1")
+
+        assert provider._client is not None
+        assert provider._client.endpoint == "https://openviking.example"
+
+        env_path.write_text(
+            "OPENVIKING_ENDPOINT=http://127.0.0.1:31933\n"
+            "OPENVIKING_API_KEY=\n",
+            encoding="utf-8",
+        )
+        assert hermes_config.reload_env() >= 1
+
+        start_calls = []
+        waiter_calls = []
+        monkeypatch.setattr(
+            openviking_plugin,
+            "_start_local_openviking_server",
+            lambda endpoint: start_calls.append(endpoint) or (True, "started"),
+        )
+        monkeypatch.setattr(
+            provider,
+            "_start_runtime_openviking_waiter",
+            lambda **kwargs: waiter_calls.append(kwargs),
+            raising=False,
+        )
+
+        out = json.loads(provider.handle_tool_call(
+            "viking_remember",
+            {"content": "stable fact"},
+        ))
+
+        assert "not connected" in out["error"]
+        assert provider._client is None
+        assert start_calls == ["http://127.0.0.1:31933"]
+        assert len(waiter_calls) == 1
+
+    def test_repeated_access_while_local_runtime_starts_does_not_spawn_again(self, monkeypatch):
+        class _AliveThread:
+            def is_alive(self):
+                return True
+
+        class _StubClient:
+            def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+                self.endpoint = endpoint
+
+            def health(self):
+                return False
+
+        monkeypatch.setattr(openviking_plugin, "_VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://127.0.0.1:31933")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "")
+
+        start_calls = []
+        provider = OpenVikingMemoryProvider()
+        provider._env_refresh_enabled = True
+        monkeypatch.setattr(
+            openviking_plugin,
+            "_start_local_openviking_server",
+            lambda endpoint: start_calls.append(endpoint) or (True, "started"),
+        )
+        monkeypatch.setattr(
+            provider,
+            "_start_runtime_openviking_waiter",
+            lambda **kwargs: setattr(provider, "_runtime_start_thread", _AliveThread()),
+            raising=False,
+        )
+
+        assert provider._ensure_client() is None
+        assert provider._ensure_client() is None
+
+        assert start_calls == ["http://127.0.0.1:31933"]
+
+    def test_concurrent_local_runtime_recovery_starts_once(self, monkeypatch):
+        class _AliveThread:
+            def is_alive(self):
+                return True
+
+        provider = OpenVikingMemoryProvider()
+        provider._endpoint = "http://127.0.0.1:31933"
+        start_calls = []
+        start_lock = threading.Lock()
+        first_start_entered = threading.Event()
+        release_start = threading.Event()
+        second_started = threading.Event()
+
+        def start_local(endpoint):
+            with start_lock:
+                start_calls.append(endpoint)
+            first_start_entered.set()
+            release_start.wait(timeout=2)
+            return True, "started"
+
+        monkeypatch.setattr(openviking_plugin, "_start_local_openviking_server", start_local)
+        monkeypatch.setattr(
+            provider,
+            "_start_runtime_openviking_waiter",
+            lambda **kwargs: setattr(provider, "_runtime_start_thread", _AliveThread()),
+            raising=False,
+        )
+
+        errors = []
+
+        def recover(*, started=None):
+            if started is not None:
+                started.set()
+            try:
+                provider._handle_runtime_openviking_unreachable()
+            except BaseException as exc:  # pragma: no cover - surfaced below
+                errors.append(exc)
+
+        threads = [
+            threading.Thread(target=recover, name="openviking-recovery-1"),
+            threading.Thread(
+                target=recover,
+                kwargs={"started": second_started},
+                name="openviking-recovery-2",
+            ),
+        ]
+        for thread in threads:
+            thread.start()
+
+        assert first_start_entered.wait(timeout=2)
+        assert second_started.wait(timeout=2)
+        release_start.set()
+        for thread in threads:
+            thread.join(timeout=2)
+            assert not thread.is_alive()
+
+        assert errors == []
+        assert start_calls == ["http://127.0.0.1:31933"]
+
+    def test_handle_tool_call_uses_ensure_client(self, monkeypatch):
+        provider = OpenVikingMemoryProvider()
+        provider._env_refresh_enabled = True
+
+        class _StubClient:
+            def __init__(self, *a, **kw):
+                pass
+
+            def health(self):
+                return False
+
+        monkeypatch.setattr("plugins.memory.openviking._VikingClient", _StubClient)
+
+        out = provider.handle_tool_call("viking_search", {"query": "x"})
+        assert "not connected" in out.lower()
+
+
+class TestEnsureClientFailureHardening:
+    """Follow-up hardening on top of #21130: failed-config cooldown and
+    torn-identity-safe connection snapshots."""
+
+    def test_failed_config_probes_once_then_cools_down(self, monkeypatch):
+        """A down endpoint must not pay a health probe on every access."""
+        probes = []
+
+        class _StubClient:
+            def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+                self.endpoint = endpoint
+
+            def health(self):
+                probes.append(self.endpoint)
+                return False
+
+        monkeypatch.setattr(openviking_plugin, "_VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "https://down.example")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "sk-x")
+
+        provider = OpenVikingMemoryProvider()
+        provider._env_refresh_enabled = True
+
+        assert provider._ensure_client() is None
+        assert provider._ensure_client() is None
+        assert provider._ensure_client() is None
+        # Only the first access probes; the rest hit the cooldown.
+        assert len(probes) == 1
+
+    def test_failed_config_retries_after_cooldown(self, monkeypatch):
+        probes = []
+
+        class _StubClient:
+            def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+                self.endpoint = endpoint
+
+            def health(self):
+                probes.append(self.endpoint)
+                return len(probes) > 1  # down on first probe, up on retry
+
+        monkeypatch.setattr(openviking_plugin, "_VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "https://flaky.example")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "sk-x")
+
+        provider = OpenVikingMemoryProvider()
+        provider._env_refresh_enabled = True
+
+        assert provider._ensure_client() is None
+        # Simulate the cooldown elapsing.
+        failed_key, _failed_at = provider._failed_refresh
+        provider._failed_refresh = (
+            failed_key,
+            time.monotonic() - openviking_plugin._FAILED_CONFIG_RETRY_COOLDOWN_SECONDS - 1,
+        )
+        assert provider._ensure_client() is not None
+        assert len(probes) == 2
+        assert provider._failed_refresh is None
+
+    def test_failed_config_retries_immediately_on_config_change(self, monkeypatch):
+        probes = []
+
+        class _StubClient:
+            def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+                self.endpoint = endpoint
+
+            def health(self):
+                probes.append(self.endpoint)
+                return self.endpoint == "https://up.example"
+
+        monkeypatch.setattr(openviking_plugin, "_VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "https://down.example")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "sk-x")
+
+        provider = OpenVikingMemoryProvider()
+        provider._env_refresh_enabled = True
+
+        assert provider._ensure_client() is None
+        # /reload lands a new endpoint: cooldown must not block the new config.
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "https://up.example")
+        client = provider._ensure_client()
+        assert client is not None
+        assert client.endpoint == "https://up.example"
+
+    def test_conn_snapshot_published_only_on_healthy(self, monkeypatch):
+        class _StubClient:
+            def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+                self.endpoint = endpoint
+
+            def health(self):
+                return self.endpoint == "https://up.example"
+
+        monkeypatch.setattr(openviking_plugin, "_VikingClient", _StubClient)
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "https://up.example")
+        monkeypatch.setenv("OPENVIKING_API_KEY", "sk-good")
+        monkeypatch.delenv("OPENVIKING_ACCOUNT", raising=False)
+        monkeypatch.delenv("OPENVIKING_USER", raising=False)
+        monkeypatch.delenv("OPENVIKING_AGENT", raising=False)
+
+        provider = OpenVikingMemoryProvider()
+        provider._env_refresh_enabled = True
+        assert provider._ensure_client() is not None
+        healthy_snapshot = provider._conn_snapshot
+        assert healthy_snapshot is not None
+        assert healthy_snapshot[0] == "https://up.example"
+        assert healthy_snapshot[1] == "sk-good"
+
+        # A failed refresh must NOT advance the snapshot: background writers
+        # (_new_client) keep targeting the last identity that passed health.
+        monkeypatch.setenv("OPENVIKING_ENDPOINT", "https://down.example")
+        assert provider._ensure_client() is None
+        assert provider._conn_snapshot == healthy_snapshot
+        built = provider._new_client()
+        assert built.endpoint == "https://up.example"

tests/plugins/memory/test_openviking_provider.py

@@ -1,6 +1,7 @@
 import json
 import os
 import stat
+import threading
 import time
 import zipfile
 from types import SimpleNamespace
@@ -82,6 +83,39 @@ def _allow_setup_validation(monkeypatch, *, root_access: bool = False):
     )
 
 
+def test_openviking_provider_config_loader_uses_readonly_config(monkeypatch):
+    import hermes_cli.config as config_mod
+
+    calls = []
+    backing_config = {
+        "memory": {
+            "openviking": {
+                "endpoint": "http://127.0.0.1:19472",
+                "api_key": "test-key",
+            }
+        }
+    }
+
+    def load_config_readonly():
+        calls.append("readonly")
+        return backing_config
+
+    def load_config():
+        raise AssertionError("OpenViking config loader should use readonly config")
+
+    monkeypatch.setattr(config_mod, "load_config_readonly", load_config_readonly)
+    monkeypatch.setattr(config_mod, "load_config", load_config)
+
+    config = openviking_module._load_hermes_openviking_config()
+
+    assert calls == ["readonly"]
+    assert config == {
+        "endpoint": "http://127.0.0.1:19472",
+        "api_key": "test-key",
+    }
+    assert config is not backing_config["memory"]["openviking"]
+
+
 @pytest.mark.skipif(os.name == "nt", reason="POSIX file modes")
 def test_openviking_env_writer_restricts_file_permissions(tmp_path):
     env_path = tmp_path / ".env"
@@ -91,6 +125,51 @@ def test_openviking_env_writer_restricts_file_permissions(tmp_path):
     assert stat.S_IMODE(env_path.stat().st_mode) == 0o600
 
 
+def test_openviking_env_writer_strips_embedded_newlines_in_values(tmp_path):
+    # A secret pasted with an embedded CR/LF must not spill onto a new line,
+    # or the round-trip re-parses the tail as a separate KEY=VALUE entry and
+    # injects an arbitrary variable into the persisted credentials file.
+    env_path = tmp_path / ".env"
+
+    openviking_module._write_env_vars(
+        env_path,
+        {"OPENVIKING_API_KEY": "good\nINJECTED_KEY=attacker"},
+    )
+
+    lines = env_path.read_text(encoding="utf-8").splitlines()
+    assert lines == ["OPENVIKING_API_KEY=goodINJECTED_KEY=attacker"]
+    # No injected line means a follow-up read sees no rogue key.
+    parsed = dict(line.split("=", 1) for line in lines if "=" in line)
+    assert set(parsed) == {"OPENVIKING_API_KEY"}
+    assert "INJECTED_KEY" not in parsed
+
+
+def test_openviking_env_writer_strips_splitline_separators_and_nul(tmp_path):
+    env_path = tmp_path / ".env"
+
+    openviking_module._write_env_vars(
+        env_path,
+        {"OPENVIKING_API_KEY": "good\u2028INJECTED_KEY=attacker\x00tail"},
+    )
+
+    lines = env_path.read_text(encoding="utf-8").splitlines()
+    assert lines == ["OPENVIKING_API_KEY=goodINJECTED_KEY=attackertail"]
+
+
+def test_openviking_env_writer_strips_newlines_when_updating_existing_key(tmp_path):
+    env_path = tmp_path / ".env"
+    env_path.write_text("OPENVIKING_API_KEY=old\n", encoding="utf-8")
+
+    openviking_module._write_env_vars(
+        env_path,
+        {"OPENVIKING_API_KEY": "new\r\nROGUE=1"},
+    )
+
+    lines = env_path.read_text(encoding="utf-8").splitlines()
+    assert lines == ["OPENVIKING_API_KEY=newROGUE=1"]
+    assert all(not line.startswith("ROGUE=") for line in lines)
+
+
 @pytest.mark.skipif(os.name == "nt", reason="POSIX file modes")
 def test_ovcli_config_writer_restricts_file_permissions(tmp_path):
     config_path = tmp_path / "ovcli.conf"
@@ -875,6 +954,40 @@ def test_initialize_autostarts_local_openviking_in_background_when_runtime_healt
     assert any("starting in the background" in message for message in statuses)
 
 
+def test_initialize_emits_starting_status_before_runtime_waiter_can_attach(monkeypatch):
+    _clear_openviking_env(monkeypatch)
+    monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://127.0.0.1:1934")
+
+    class FakeVikingClient:
+        def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+            assert endpoint == "http://127.0.0.1:1934"
+
+        def health(self):
+            return False
+
+    monkeypatch.setattr(openviking_module, "_VikingClient", FakeVikingClient)
+    monkeypatch.setattr(
+        openviking_module,
+        "_start_local_openviking_server",
+        lambda endpoint: (True, "started"),
+    )
+
+    provider = OpenVikingMemoryProvider()
+    statuses = []
+
+    def start_waiter(*, endpoint, status_callback=None, warning_callback=None):
+        assert endpoint == "http://127.0.0.1:1934"
+        assert callable(status_callback)
+        status_callback("Local OpenViking server is reachable; OpenViking memory is active.")
+
+    monkeypatch.setattr(provider, "_start_runtime_openviking_waiter", start_waiter, raising=False)
+
+    provider.initialize("session-1", platform="cli", status_callback=statuses.append)
+
+    assert "starting in the background" in statuses[0]
+    assert "memory is active" in statuses[1]
+
+
 def test_runtime_openviking_waiter_attaches_client_after_health_recovers(monkeypatch):
     _clear_openviking_env(monkeypatch)
     wait_calls = []
@@ -913,13 +1026,58 @@ def test_runtime_openviking_waiter_attaches_client_after_health_recovers(monkeyp
     assert provider._client is not None
     assert provider._client.endpoint == "http://127.0.0.1:1934"
     assert provider._client.api_key == "secret"
-    assert wait_calls == [(
-        "http://127.0.0.1:1934",
-        {"timeout_seconds": openviking_module._LOCAL_OPENVIKING_AUTOSTART_TIMEOUT},
-    )]
+    assert len(wait_calls) == 1
+    endpoint, wait_kwargs = wait_calls[0]
+    assert endpoint == "http://127.0.0.1:1934"
+    assert wait_kwargs["timeout_seconds"] == openviking_module._LOCAL_OPENVIKING_AUTOSTART_TIMEOUT
+    assert callable(wait_kwargs["should_stop"])
+    assert wait_kwargs["should_stop"]() is False
+    provider._shutting_down = True
+    assert wait_kwargs["should_stop"]() is True
     assert any("OpenViking memory is active" in message for message in statuses)
 
 
+def test_runtime_waiter_does_not_replace_client_after_endpoint_changes(monkeypatch):
+    wait_entered = threading.Event()
+    release_wait = threading.Event()
+
+    def wait_for_health(endpoint, *, timeout_seconds, should_stop=None):
+        assert endpoint == "http://127.0.0.1:1934"
+        wait_entered.set()
+        assert release_wait.wait(2.0)
+        return True
+
+    class FakeVikingClient:
+        def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+            self.endpoint = endpoint
+
+        def health(self):
+            return True
+
+    monkeypatch.setattr(openviking_module, "_wait_for_openviking_health", wait_for_health)
+    monkeypatch.setattr(openviking_module, "_VikingClient", FakeVikingClient)
+
+    provider = OpenVikingMemoryProvider()
+    provider._endpoint = "http://127.0.0.1:1934"
+    provider._api_key = ""
+    provider._account = ""
+    provider._user = ""
+    provider._agent = "hermes"
+
+    waiter = threading.Thread(target=provider._finish_runtime_openviking_start)
+    waiter.start()
+    assert wait_entered.wait(2.0)
+
+    replacement_client = FakeVikingClient("https://new.example")
+    provider._endpoint = replacement_client.endpoint
+    provider._client = replacement_client
+    release_wait.set()
+    waiter.join(timeout=2.0)
+
+    assert not waiter.is_alive()
+    assert provider._client is replacement_client
+
+
 def test_runtime_openviking_waiter_warns_when_background_start_times_out(monkeypatch):
     _clear_openviking_env(monkeypatch)
     monkeypatch.setattr(
@@ -1893,6 +2051,51 @@ def test_viking_client_retries_with_tenant_headers_for_trusted_mode(monkeypatch)
     assert captured_headers[1]["X-OpenViking-User"] == "usr"
 
 
+def test_viking_client_does_not_retry_root_tenant_error_as_trusted_mode(monkeypatch):
+    client = _VikingClient(
+        "https://example.com",
+        api_key="test-key",
+        account="acct",
+        user="usr",
+        agent="hermes",
+    )
+    captured_headers = []
+
+    def capture_post(url, **kwargs):
+        captured_headers.append(kwargs.get("headers") or {})
+        if len(captured_headers) == 1:
+            return SimpleNamespace(
+                status_code=403,
+                text="",
+                json=lambda: {
+                    "error": {
+                        "code": "PERMISSION_DENIED",
+                        "message": (
+                            "ROOT API keys cannot access tenant-scoped data APIs in api_key mode. "
+                            "Use a user/admin API key for data access, or trusted mode for upstream "
+                            "identity assertion."
+                        ),
+                    },
+                },
+                raise_for_status=lambda: None,
+            )
+        return SimpleNamespace(
+            status_code=200,
+            text="",
+            json=lambda: {"status": "ok", "result": {"total": 0}},
+            raise_for_status=lambda: None,
+        )
+
+    monkeypatch.setattr(client._httpx, "post", capture_post)
+
+    with pytest.raises(openviking_module._OpenVikingHTTPError, match="ROOT API keys cannot access"):
+        client.post("/api/v1/search/search", {"query": "status"})
+
+    assert len(captured_headers) == 1
+    assert "X-OpenViking-Account" not in captured_headers[0]
+    assert "X-OpenViking-User" not in captured_headers[0]
+
+
 def test_viking_client_health_sends_auth_headers(monkeypatch):
     _clear_openviking_tenant_env(monkeypatch)
     client = _VikingClient(

tests/plugins/memory/test_openviking_shutdown.py

@@ -0,0 +1,163 @@
+"""Tests for OpenViking memory-provider shutdown teardown.
+
+The runtime-autostart waiter is a tracked ``daemon=True`` thread that blocks
+on network health probes. If ``shutdown()`` doesn't join it (and the waiter
+doesn't bail on the shutdown flag), it can be left alive at interpreter exit,
+which crashes CPython with SIGABRT at ``Py_FinalizeEx``. These tests assert
+the waiter short-circuits on shutdown and that ``shutdown()`` waits for the
+runtime-start thread.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from unittest.mock import patch
+
+import plugins.memory.openviking as openviking_module
+from plugins.memory.openviking import OpenVikingMemoryProvider
+
+
+def test_wait_for_health_short_circuits_on_should_stop():
+    """The health waiter returns False without probing when should_stop is set,
+    so the daemon thread running it can be join()ed promptly at shutdown."""
+    probes: list[str] = []
+
+    def _reach(endpoint):
+        probes.append(endpoint)
+        return (False, "down")
+
+    with patch.object(
+        openviking_module, "_validate_openviking_reachability", _reach
+    ):
+        result = openviking_module._wait_for_openviking_health(
+            "http://example.invalid",
+            timeout_seconds=60.0,
+            should_stop=lambda: True,
+        )
+
+    assert result is False
+    assert probes == []  # bailed before the first network probe
+
+
+def test_shutdown_waits_for_runtime_start_thread():
+    """shutdown() must join the runtime-autostart waiter thread.
+
+    The fake waiter does post-stop work (a short sleep) once it observes the
+    shutdown flag. If shutdown() joins it, that work has completed by the time
+    shutdown() returns; without the join, shutdown() returns early and the
+    thread is still running (the SIGABRT-at-exit failure mode).
+    """
+    provider = OpenVikingMemoryProvider()
+    started = threading.Event()
+    finished = threading.Event()
+
+    def _runtime():
+        started.set()
+        while not provider._shutting_down:
+            time.sleep(0.01)
+        time.sleep(0.2)  # work that must finish during shutdown's join
+        finished.set()
+
+    t = threading.Thread(target=_runtime, daemon=True, name="openviking-runtime-start")
+    provider._runtime_start_thread = t
+    t.start()
+    assert started.wait(2.0)
+
+    provider.shutdown()
+
+    assert finished.is_set()
+    assert not t.is_alive()
+
+
+def test_shutdown_during_pending_runtime_start_does_not_launch_waiter(monkeypatch):
+    """A waiter reserved before shutdown must not start after shutdown returns."""
+    provider = OpenVikingMemoryProvider()
+    provider._endpoint = "http://127.0.0.1:1934"
+    status_entered = threading.Event()
+    release_status = threading.Event()
+    waiter_calls = []
+
+    monkeypatch.setattr(
+        openviking_module,
+        "_start_local_openviking_server",
+        lambda endpoint: (True, "started"),
+    )
+    monkeypatch.setattr(
+        provider,
+        "_start_runtime_openviking_waiter",
+        lambda **kwargs: waiter_calls.append(kwargs),
+    )
+
+    def status_callback(_message):
+        status_entered.set()
+        assert release_status.wait(2.0)
+
+    starter = threading.Thread(
+        target=provider._handle_runtime_openviking_unreachable,
+        kwargs={"status_callback": status_callback},
+        name="openviking-pending-start",
+    )
+    starter.start()
+    assert status_entered.wait(2.0)
+
+    provider.shutdown()
+    release_status.set()
+    starter.join(timeout=2.0)
+
+    assert not starter.is_alive()
+    assert provider._runtime_start_pending is False
+    assert waiter_calls == []
+
+
+def test_shutdown_during_final_health_probe_does_not_publish_client(monkeypatch):
+    """A successful final probe must not reactivate a provider being torn down."""
+    provider = OpenVikingMemoryProvider()
+    provider._endpoint = "http://127.0.0.1:1934"
+    provider._api_key = ""
+    provider._account = ""
+    provider._user = ""
+    provider._agent = "hermes"
+    health_entered = threading.Event()
+    release_health = threading.Event()
+
+    monkeypatch.setattr(
+        openviking_module,
+        "_wait_for_openviking_health",
+        lambda endpoint, **kwargs: True,
+    )
+
+    class FakeVikingClient:
+        def __init__(self, endpoint, api_key="", account="", user="", agent=""):
+            self.endpoint = endpoint
+
+        def health(self):
+            health_entered.set()
+            assert release_health.wait(2.0)
+            return True
+
+    monkeypatch.setattr(openviking_module, "_VikingClient", FakeVikingClient)
+
+    waiter = threading.Thread(
+        target=provider._finish_runtime_openviking_start,
+        name="openviking-final-health",
+    )
+    provider._runtime_start_thread = waiter
+    waiter.start()
+    assert health_entered.wait(2.0)
+
+    shutdown = threading.Thread(target=provider.shutdown, name="openviking-shutdown")
+    shutdown.start()
+    for _ in range(100):
+        if provider._shutting_down:
+            break
+        time.sleep(0.01)
+    assert provider._shutting_down is True
+
+    release_health.set()
+    waiter.join(timeout=2.0)
+    shutdown.join(timeout=2.0)
+
+    assert not waiter.is_alive()
+    assert not shutdown.is_alive()
+    assert provider._client is None

tests/tools/test_refresh_agent_mcp_tools.py

@@ -140,6 +140,32 @@ def test_refresh_preserves_memory_provider_and_context_engine_tools(monkeypatch)
     assert added == {"mcp_new_server_tool"}
 
 
+def test_refresh_does_not_reinject_disabled_memory_provider_tools(monkeypatch):
+    """A refresh removes stale provider tools when memory becomes disabled."""
+    agent = _agent(
+        ["read_file", "memory_search"],
+        enabled=["all"],
+        disabled=["memory"],
+    )
+    agent._memory_manager = types.SimpleNamespace(
+        get_all_tool_schemas=lambda: [
+            {"name": "memory_search", "description": "", "parameters": {}}
+        ]
+    )
+
+    import model_tools
+    monkeypatch.setattr(
+        model_tools,
+        "get_tool_definitions",
+        lambda **kw: [_tool("read_file")],
+    )
+
+    mcp_tool.refresh_agent_mcp_tools(agent)
+
+    assert "memory_search" not in agent.valid_tool_names
+    assert all(t["function"]["name"] != "memory_search" for t in agent.tools)
+
+
 def test_refresh_respects_context_engine_toolset_gate(monkeypatch):
     """#5544: context-engine tools must NOT be re-injected on a restricted
     toolset. A platform with enabled_toolsets that excludes context_engine

tools/mcp_tool.py

@@ -6134,9 +6134,13 @@ def _reinject_post_build_tools(agent, tools_list: list, name_set: set) -> set:
         memory_manager = getattr(agent, "_memory_manager", None)
         get_mem_schemas = getattr(memory_manager, "get_all_tool_schemas", None) if memory_manager else None
         if callable(get_mem_schemas):
-            # Honor the same enablement gate inject_memory_provider_tools uses.
+            # Honor the same toolset gate inject_memory_provider_tools uses.
             from agent.memory_manager import memory_provider_tools_enabled
-            if "memory" in name_set or memory_provider_tools_enabled(getattr(agent, "enabled_toolsets", None)):
+            if memory_provider_tools_enabled(
+                getattr(agent, "enabled_toolsets", None),
+                getattr(agent, "disabled_toolsets", None),
+                memory_tool_present="memory" in name_set,
+            ):
                 for schema in get_mem_schemas():
                     if isinstance(schema, dict):
                         _add(schema)