fix(redact): strip complete CSI/SGR sequences from shadow copy (#81012)

agent/agent_init.py

@@ -1763,6 +1763,14 @@ def init_agent(
             agent._memory_manager = None
 
     from agent.memory_manager import inject_memory_provider_tools as _inject_memory_provider_tools
+    # Record the toolset gating on the manager BEFORE injecting tools so a
+    # provider whose tools are gated out can suppress its system_prompt_block
+    # rather than dangling instructions for non-existent tools (#81014).
+    if agent._memory_manager is not None:
+        agent._memory_manager.set_tool_gating(
+            enabled_toolsets=getattr(agent, "enabled_toolsets", None),
+            disabled_toolsets=getattr(agent, "disabled_toolsets", None),
+        )
     _inject_memory_provider_tools(agent)
 
     # Skills config: nudge interval for skill creation reminders

agent/memory_manager.py

@@ -119,11 +119,32 @@ def inject_memory_provider_tools(agent: Any) -> int:
         for tool in tools
         if isinstance(tool, dict)
     }
+    enabled_toolsets = getattr(agent, "enabled_toolsets", None)
+    disabled_toolsets = getattr(agent, "disabled_toolsets", None)
     if not memory_provider_tools_enabled(
-        getattr(agent, "enabled_toolsets", None),
-        getattr(agent, "disabled_toolsets", None),
+        enabled_toolsets,
+        disabled_toolsets,
         memory_tool_present="memory" in existing_tool_names,
     ):
+        # Surface the silent suppression. The provider was initialized above
+        # (it might still produce a system_prompt_block, which we gate in
+        # build_system_prompt() above) but its tools are not in this agent's
+        # tool surface — emit a single WARNING so an operator reading
+        # agent.log can correlate the dangling instructions with the gate
+        # that produced them (#81014).
+        if memory_manager.providers:
+            try:
+                _schemas = list(memory_manager.get_all_tool_schemas())
+                _count = len(_schemas)
+            except Exception:
+                _count = "?"
+            logger.warning(
+                "Memory provider is initialized with %s tool schemas, but "
+                "they are NOT in the agent's tool surface. Enable the "
+                "'memory' toolset in platform_toolsets (or remove "
+                "memory from disabled_toolsets) to expose them (#81014).",
+                _count,
+            )
         return 0
 
     get_schemas = getattr(memory_manager, "get_all_tool_schemas", None)
@@ -381,6 +402,13 @@ def __init__(self, *, external_prefetch_timeout: Optional[float] = None) -> None
             raise ValueError("external_prefetch_timeout must be positive")
         self._external_prefetch_threads: Dict[str, threading.Thread] = {}
         self._external_prefetch_lock = threading.Lock()
+        # Toolset gating state — set via ``set_tool_gating()`` so
+        # ``build_system_prompt()`` can suppress provider blocks whose tools
+        # are not exposed in the agent's tool surface (#81014). ``None``
+        # means "no restriction" — the legacy behavior that always injects
+        # the provider block.
+        self._enabled_toolsets: Optional[List[str]] = None
+        self._disabled_toolsets: Optional[List[str]] = None
         # Background executor for end-of-turn sync/prefetch. Lazily created on
         # first use so the common builtin-only path spawns no extra threads.
         # A single worker serializes a provider's writes (turn N must land
@@ -483,15 +511,58 @@ def get_provider(self, name: str) -> Optional[MemoryProvider]:
 
     # -- System prompt -------------------------------------------------------
 
+    def set_tool_gating(
+        self,
+        *,
+        enabled_toolsets: Optional[List[str]] = None,
+        disabled_toolsets: Optional[List[str]] = None,
+    ) -> None:
+        """Record the agent's current toolset gating so ``build_system_prompt``
+        can suppress provider blocks whose tools are not exposed (#81014).
+
+        Called from agent_init after the agent's enabled/disabled toolset
+        configuration is known. ``None`` for either argument means "no
+        restriction" — the legacy behavior that always emits the block.
+        """
+        self._enabled_toolsets = enabled_toolsets
+        self._disabled_toolsets = disabled_toolsets
+
     def build_system_prompt(self) -> str:
         """Collect system prompt blocks from all providers.
 
         Returns combined text, or empty string if no providers contribute.
-        Each non-empty block is labeled with the provider name.
+        Each non-empty block is labeled with the provider name. A provider
+        whose tools are gated out of the agent's tool surface is skipped —
+        emitting instructions for tools that do not exist would dangle
+        (#81014).
         """
+        # Decide whether the provider tools are exposed. When no gating
+        # state has been set yet (``enabled_toolsets`` is ``None``) we
+        # preserve the legacy behavior — always emit. After agent_init
+        # calls ``set_tool_gating()`` the gate is honored.
+        gate_set = (
+            self._enabled_toolsets is not None
+            or self._disabled_toolsets is not None
+        )
+        tools_exposed = True
+        if gate_set:
+            tools_exposed = memory_provider_tools_enabled(
+                self._enabled_toolsets,
+                self._disabled_toolsets,
+            )
         blocks = []
         for provider in self._providers:
             try:
+                if gate_set and not tools_exposed:
+                    logger.warning(
+                        "Memory provider '%s' is initialized but its tools "
+                        "are NOT in the tool surface (platform_toolsets/"
+                        "disabled_toolsets gate). Suppressing "
+                        "system_prompt_block() to avoid dangling tool "
+                        "references (#81014).",
+                        provider.name,
+                    )
+                    continue
                 block = provider.system_prompt_block()
                 if block and block.strip():
                     blocks.append(block)

agent/redact.py

@@ -497,12 +497,51 @@ def _mask_control_split_tokens(text: str, mask_fn) -> str:
     contains solely token-body and control chars (a match that crosses into a
     different line's unrelated text, e.g. ``EXA_API_KEY=*** is rejected).
     """
-    stripped = _CONTROL_CHARS_RE.sub("", text)
+    # Full CSI / SGR sequences (``\x1b[...letter``). Stripped from the shadow
+    # copy before the bare control-char strip so a token like
+    # ``\x1b[32msk-AAA…BBB\x1b[0m`` collapses to ``sk-AAA…BBB`` and matches
+    # ``_PREFIX_RE`` — without this, the bare ESC strip leaves ``[32m``
+    # glued to the head and the ``[m`` byte defeats the prefix lookbehind
+    # (#81012). Reuses the shape from ``tools/ansi_strip.py`` (CSI only —
+    # OSC/DCS can carry arbitrary payloads and are not safe to delete from
+    # redaction shadow-copies; the bare-ESC strip below still handles them
+    # as control bytes).
+    _CSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]")
+
+    # First pass: build a single global shadow-copy of the text with every
+    # CSI sequence AND every bare control/zero-width char removed. The
+    # CSI-then-control order matters — stripping the bare ESC byte first
+    # would leave ``[32m`` glued to the head and the ``[m`` byte defeats
+    # the prefix lookbehind (#81012).
+    stripped = _CONTROL_CHARS_RE.sub("", _CSI_RE.sub("", text))
     if stripped == text:
         return text
-    orig_idx = [i for i, c in enumerate(text) if not _CONTROL_CHARS_RE.match(c)]
+
+    # The back-map from a position in ``stripped`` to the position in the
+    # original ``text`` is non-trivial because (a) a CSI sequence removes
+    # multiple chars per match and (b) ``_CONTROL_CHARS_RE`` removes one
+    # char per match. Build it iteratively so multi-char CSI deletions
+    # collapse correctly.
+    csi_spans = [m.span() for m in _CSI_RE.finditer(text)]
+    csi_positions = set()
+    for a, b in csi_spans:
+        for i in range(a, b):
+            csi_positions.add(i)
+
+    def _is_collapsible(idx: int) -> bool:
+        if idx in csi_positions:
+            return True
+        if _CONTROL_CHARS_RE.match(text[idx]):
+            return True
+        return False
+
+    orig_idx: list[int] = []
+    for i in range(len(text)):
+        if not _is_collapsible(i):
+            orig_idx.append(i)
+
     out = list(text)
-    matches = []
+    matches: list = []
     for m in _PREFIX_RE.finditer(stripped):
         body = m.group(1)
         start_orig = orig_idx[m.start(1)]
@@ -515,10 +554,10 @@ def _mask_control_split_tokens(text: str, mask_fn) -> str:
         # the self-matching fragment is handled by the ordinary prefix pass
         # (any remainder past the newline is left unmasked — accepted
         # residual to preserve line structure).
-        # For NON-newline controls (ESC, ZWSP, ...) the join proceeds even
-        # when a fragment self-matches: those bytes never legitimately sit
-        # between a token and adjacent prose, and skipping there let the
-        # non-matching remainder of a split token leak
+        # For NON-newline controls (ESC, ZWSP, CSI sequences) the join
+        # proceeds even when a fragment self-matches: those bytes never
+        # legitimately sit between a token and adjacent prose, and skipping
+        # there let the non-matching remainder of a split token leak
         # (``sk-<head>\x1b<tail>`` masked only the head).
         span = text[start_orig:end_orig]
         if ("\n" in span or "\r" in span) and _PREFIX_RE.search(span):
@@ -528,10 +567,23 @@ def _mask_control_split_tokens(text: str, mask_fn) -> str:
         # token body, so the regex matched across unrelated lines). Also
         # reject when the match runs into a ``KEY=`` name: a real token value
         # is followed by a newline/space/end, not ``=``.
-        if (all(c in _TOKEN_BODY_CHARS or _CONTROL_CHARS_RE.match(c)
-                for c in span)
-                and (end_orig >= len(text) or text[end_orig] != "=")):
-            matches.append((start_orig, end_orig, mask_fn(body)))
+        # A position is considered "collapsible" (i.e. safely erased in
+        # the shadow) when the original char is a token-body char, a bare
+        # control char, or part of a CSI sequence. CSI chars are
+        # permitted here even though they aren't in _TOKEN_BODY_CHARS —
+        # the shadow already erased them, and including them keeps the
+        # legacy "all-eraseable" invariant for split-token joins (#81012).
+        if not all(
+            c in _TOKEN_BODY_CHARS
+            or _CONTROL_CHARS_RE.match(c)
+            or (start_orig + i) in csi_positions
+            for i, c in enumerate(span)
+        ):
+            continue
+        if end_orig < len(text) and text[end_orig] == "=":
+            continue
+        matches.append((start_orig, end_orig, mask_fn(body)))
+
     for start_orig, end_orig, replacement in reversed(matches):
         out[start_orig:end_orig] = list(replacement)
     return "".join(out)

gateway/run.py

@@ -25727,6 +25727,16 @@ def _run_sync_with_timeout_lifecycle():
                 pending_event = None
                 pending = None
 
+            # Tracks (chat_id, session_key) tuples for which the queued-follow-up
+            # fallback path already delivered ``first_response`` via
+            # ``adapter.send()`` (#81052). The normal completion pipeline checks
+            # this set before re-sending the same text, so a slow turn whose
+            # stream consumer didn't confirm final delivery cannot duplicate the
+            # message. The set is scoped to this single pending-message handling
+            # block; ``session_key`` is constant within it, ``chat_id`` matches
+            # the inbound source, and at most one fallback send runs per chat.
+            _delivered_in_fallback = set()
+
             if pending_event or pending:
                 logger.debug("Processing pending message: '%s...'", pending[:40])
 
@@ -25807,6 +25817,14 @@ def _run_sync_with_timeout_lifecycle():
                                 first_response,
                                 metadata=_status_thread_metadata,
                             )
+                            # Mark this chat as having received the turn-final
+                            # response via the queued-follow-up fallback send
+                            # so the subsequent normal completion pipeline does
+                            # not re-send the same ``first_response`` (#81052).
+                            # ``_delivered_in_fallback`` lives in this turn's
+                            # pending-message scope; the set is bounded by the
+                            # inbound chat and the request lifetime.
+                            _delivered_in_fallback.add((source.chat_id, session_key))
                         except Exception as e:
                             logger.warning("Failed to send first response before queued message: %s", e)
                     elif first_response:
@@ -26074,13 +26092,14 @@ def _run_sync_with_timeout_lifecycle():
                 _final,
                 previewed=_previewed,
             )
-            if not _is_empty_sentinel and not _transformed and (_streamed or _content_delivered):
+            if not _is_empty_sentinel and not _transformed and (_streamed or _content_delivered or (source.chat_id, session_key) in _delivered_in_fallback):
                 logger.info(
-                    "Suppressing normal final send for session %s: final delivery already confirmed (streamed=%s previewed=%s content_delivered=%s).",
+                    "Suppressing normal final send for session %s: final delivery already confirmed (streamed=%s previewed=%s content_delivered=%s fallback=%s).",
                     session_key or "?",
                     _streamed,
                     _previewed,
                     _content_delivered,
+                    (source.chat_id, session_key) in _delivered_in_fallback,
                 )
                 response["already_sent"] = True
             elif not _is_empty_sentinel and not _transformed and _stale_finalized and _sc is not None:

hermes_cli/web_routers/mcp.py

@@ -131,6 +131,24 @@ async def remove_mcp_server(name: str, profile: Optional[str] = None):
         removed = _remove_mcp_server(name)
     if not removed:
         raise HTTPException(status_code=404, detail=f"Server '{name}' not found")
+
+    # Also evict any cached OAuth provider and delete on-disk token state
+    # (.json, .client.json, .meta.json) so a server removed via the dashboard
+    # cannot be revived at the next gateway restart by leftover files in
+    # mcp-tokens/. The CLI `hermes mcp remove` path already routes through
+    # MCPOAuthManager.remove() to achieve this (#81050); the dashboard DELETE
+    # path must do the same.
+    try:
+        from tools.mcp_oauth_manager import get_manager
+
+        get_manager().remove(name)
+    except Exception:
+        # Token cleanup is best-effort: a missing tokens directory is fine,
+        # but the underlying HermesTokenStorage.remove() is the same call
+        # path the CLI uses, so failures here indicate a real disk problem
+        # and the next gateway start will surface it.
+        pass
+
     return {"ok": True}