diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py
index 57f943eda1f03..365715128efb1 100644
--- a/gateway/platforms/api_server.py
+++ b/gateway/platforms/api_server.py
@@ -6129,18 +6129,57 @@ def _response_messages_turn_start_index(
         user_message: Any,
         result: Dict[str, Any],
     ) -> int:
-        """Detect transcript-shaped result["messages"] and return turn start."""
+        """Detect transcript-shaped result["messages"] and return turn start.
+
+        Uses role+content matching (ignoring metadata fields like timestamp,
+        finish_reason, etc.) because the agent modifies messages during the
+        conversation loop — timestamps are added, content may be truncated,
+        and fields like finish_reason/reasoning are stamped on. Full dict
+        equality (``==``) fails on these modifications, causing the prefix
+        match to return 0 and the full history to be returned instead of just
+        the current turn. See #89891.
+        """
         agent_messages = result.get("messages") if isinstance(result, dict) else None
         if not isinstance(agent_messages, list) or not agent_messages:
             return 0
 
+        def _match(expected: Dict[str, Any], actual: Dict[str, Any]) -> bool:
+            """Compare role + content, ignoring metadata fields."""
+            if expected.get("role") != actual.get("role"):
+                return False
+            # Compare content (may be str, list, or None)
+            exp_content = expected.get("content")
+            act_content = actual.get("content")
+            if exp_content != act_content:
+                # Handle string content that may be truncated by agent
+                if isinstance(exp_content, str) and isinstance(act_content, str):
+                    # Allow prefix match for content (agent may truncate)
+                    if not act_content.startswith(exp_content[:100]):
+                        return False
+                else:
+                    return False
+            return True
+
         prior = list(conversation_history)
         current_user = {"role": "user", "content": user_message}
         expected_prefix = prior + [current_user]
-        if agent_messages[:len(expected_prefix)] == expected_prefix:
-            return len(expected_prefix)
-        if prior and agent_messages[:len(prior)] == prior:
-            return len(prior)
+
+        # Try matching with current user message
+        if len(agent_messages) >= len(expected_prefix):
+            if all(
+                _match(expected, actual)
+                for expected, actual in zip(expected_prefix, agent_messages[:len(expected_prefix)])
+            ):
+                return len(expected_prefix)
+
+        # Try matching without current user message (edge case)
+        if prior and len(agent_messages) >= len(prior):
+            if all(
+                _match(expected, actual)
+                for expected, actual in zip(prior, agent_messages[:len(prior)])
+            ):
+                return len(prior)
+
         return 0
 
     @classmethod
diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py
index 29ae25d7f700d..50ef009ae73a0 100644
--- a/hermes_cli/config_defaults.py
+++ b/hermes_cli/config_defaults.py
@@ -2200,6 +2200,13 @@
                 "On it.",
             ],
         },
+        # Max lifetime (seconds) for the persistent typing loop. The loop that
+        # POSTs /channels/{id}/typing every 12s has no natural exit condition
+        # other than stop_typing() or a non-429 error — if stop_typing never
+        # reaches the adapter (e.g. a crashed run, or a thread-vs-parent-channel
+        # key mismatch), the loop runs forever and the "typing…" badge sticks
+        # until the gateway restarts. Set to 0 to disable the deadline guard.
+        "typing_loop_max_seconds": 600,
     },
 
     # WhatsApp platform settings (gateway mode)
diff --git a/hermes_cli/mcp_catalog.py b/hermes_cli/mcp_catalog.py
index 8bbec9b06f348..04bb520093005 100644
--- a/hermes_cli/mcp_catalog.py
+++ b/hermes_cli/mcp_catalog.py
@@ -571,6 +571,15 @@ def _build_server_config(
             cfg["args"] = [_expand_install_dir(a, install_dir) for a in t.args]
         if t.env:
             cfg["env"] = dict(t.env)
+        # Wire auth.env credentials into the stdio child's environment.
+        # install_entry() already saved these to .env via _prompt_env_vars(),
+        # but without an env-backed reference here, _build_safe_env() would
+        # exclude them and the child would start without its API key (#89316).
+        if entry.auth.type == "api_key" and entry.auth.env:
+            env = cfg.get("env") or {}
+            for spec in entry.auth.env:
+                env[spec.name] = f"${{{spec.name}}}"
+            cfg["env"] = env
     elif t.type == "http":
         cfg["url"] = t.url
         if entry.auth.type == "oauth":
diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py
index ef0cdd14d74d4..0e979096057c4 100644
--- a/plugins/platforms/discord/adapter.py
+++ b/plugins/platforms/discord/adapter.py
@@ -1113,6 +1113,7 @@ def __init__(self, config: PlatformConfig):
         # Persistent typing indicator loops per channel (DMs don't reliably
         # show the standard typing gateway event for bots)
         self._typing_tasks: Dict[str, asyncio.Task] = {}
+        self._typing_loop_max_seconds = self._load_typing_loop_max_seconds()
         self._bot_task: Optional[asyncio.Task] = None
         self._post_connect_task: Optional[asyncio.Task] = None
         # WebSocket-level liveness probe. Discord REST and Gateway are distinct
@@ -4356,6 +4357,14 @@ def _load_playback_timeout(self) -> int:
             minimum=1,
         )
 
+    def _load_typing_loop_max_seconds(self) -> int:
+        """Return max typing-loop lifetime in seconds; 0 disables the deadline."""
+        return self._load_discord_int_config(
+            "typing_loop_max_seconds",
+            600,
+            minimum=0,
+        )
+
     def _voice_timeout_limit(self) -> int:
         return int(getattr(self, "_voice_timeout_seconds", self.VOICE_TIMEOUT))
 
@@ -5588,6 +5597,12 @@ async def send_typing(self, chat_id: str, metadata=None) -> None:
         warning, sleeps for the ``retry_after`` duration (or a sensible
         default), and continues — it does NOT die on a single rate-limit
         hit.  Only CancelledError (from stop_typing) stops the loop.
+
+        A max-lifetime deadline (configurable via
+        ``discord.typing_loop_max_seconds``, default 600s) guards against
+        orphaned loops that never receive ``stop_typing()`` — e.g. a crashed
+        run, or a thread-vs-parent-channel key mismatch. When the deadline
+        elapses, the loop exits cleanly on its own.
         """
         if not self._client:
             return
@@ -5595,9 +5610,26 @@ async def send_typing(self, chat_id: str, metadata=None) -> None:
         if chat_id in self._typing_tasks:
             return
 
+        _typing_loop_max_seconds = self._typing_loop_max_seconds
+
         async def _typing_loop() -> None:
             try:
+                _loop_deadline = (
+                    time.monotonic() + _typing_loop_max_seconds
+                    if _typing_loop_max_seconds > 0
+                    else None
+                )
                 while True:
+                    if (
+                        _loop_deadline is not None
+                        and time.monotonic() >= _loop_deadline
+                    ):
+                        logger.info(
+                            "Typing loop max lifetime (%ss) elapsed for %s — stopping",
+                            _typing_loop_max_seconds,
+                            chat_id,
+                        )
+                        return
                     try:
                         route = discord.http.Route(
                             "POST", "/channels/{channel_id}/typing",
@@ -5622,7 +5654,17 @@ async def _typing_loop() -> None:
                             return
                         await asyncio.sleep(retry_after)
                         continue
-                    await asyncio.sleep(12)
+                    # Typing indicator lasts ~10s on Discord's side, so we
+                    # refresh every 12s. Bound the sleep by the deadline so
+                    # the loop wakes up in time to honor the max-lifetime
+                    # guard instead of sleeping past it.
+                    if _loop_deadline is not None:
+                        remaining = _loop_deadline - time.monotonic()
+                        if remaining <= 0:
+                            return
+                        await asyncio.sleep(min(12, remaining))
+                    else:
+                        await asyncio.sleep(12)
             except asyncio.CancelledError:
                 pass
             finally:
diff --git a/tests/gateway/test_api_server.py b/tests/gateway/test_api_server.py
index 9094bc617e101..9668be9ba6daf 100644
--- a/tests/gateway/test_api_server.py
+++ b/tests/gateway/test_api_server.py
@@ -2130,6 +2130,153 @@ async def test_truncation_auto_preserves_non_leading_compaction_summary(self, ad
         assert history[-1]["content"] == "msg 147"
 
 
+# ---------------------------------------------------------------------------
+# Turn-start detection — role+content matching (ignoring metadata)
+# Regression tests for #89891
+# ---------------------------------------------------------------------------
+
+
+class TestTurnStartDetection:
+    """Response-side turn-start detection uses role+content matching
+    (ignoring metadata) so it survives the agent's in-loop message
+    modifications (timestamps, content truncation, finish_reason).
+    """
+
+    def test_timestamps_on_messages_does_not_break_detection(self):
+        """Agent adds timestamp fields — must still detect turn start."""
+        history = [
+            {"role": "user", "content": "Hello"},
+            {"role": "assistant", "content": "Hi there!"},
+        ]
+        result = {
+            "messages": [
+                {"role": "user", "content": "Hello", "timestamp": 1000},
+                {"role": "assistant", "content": "Hi there!", "timestamp": 1001},
+                {"role": "user", "content": "What is 2+2?", "timestamp": 1002},
+                {"role": "assistant", "content": "4", "timestamp": 1003},
+            ]
+        }
+        assert APIServerAdapter._response_messages_turn_start_index(
+            history, "What is 2+2?", result
+        ) == 3
+
+    def test_exact_match_still_works(self):
+        """Full dict equality path still works for unmodified messages."""
+        history = [
+            {"role": "user", "content": "Hello"},
+            {"role": "assistant", "content": "Hi!"},
+        ]
+        result = {
+            "messages": [
+                {"role": "user", "content": "Hello"},
+                {"role": "assistant", "content": "Hi!"},
+                {"role": "user", "content": "Follow up"},
+                {"role": "assistant", "content": "OK"},
+            ]
+        }
+        assert APIServerAdapter._response_messages_turn_start_index(
+            history, "Follow up", result
+        ) == 3
+
+    def test_empty_history_matches_first_user(self):
+        """Empty history: match the first user message."""
+        result = {
+            "messages": [
+                {"role": "user", "content": "Hello"},
+                {"role": "assistant", "content": "Hi!"},
+            ]
+        }
+        assert APIServerAdapter._response_messages_turn_start_index(
+            [], "Hello", result
+        ) == 1
+
+    def test_tool_calls_with_timestamps(self):
+        """Tool call messages with timestamps — detect correctly."""
+        history = [
+            {"role": "user", "content": "Hi"},
+            {"role": "assistant", "content": "Hello!"},
+        ]
+        result = {
+            "messages": [
+                {"role": "user", "content": "Hi", "timestamp": 100},
+                {"role": "assistant", "content": "Hello!", "timestamp": 101},
+                {"role": "user", "content": "Compute", "timestamp": 102},
+                {
+                    "role": "assistant",
+                    "content": None,
+                    "tool_calls": [{"id": "1", "function": {"name": "calc", "arguments": "{}"}}],
+                    "timestamp": 103,
+                },
+                {"role": "tool", "content": "42", "tool_call_id": "1", "timestamp": 104},
+                {"role": "assistant", "content": "42", "timestamp": 105},
+            ]
+        }
+        assert APIServerAdapter._response_messages_turn_start_index(
+            history, "Compute", result
+        ) == 3
+
+    def test_truncated_content_matches(self):
+        """Agent may truncate long content — prefix match should still work."""
+        long_content = "A" * 200
+        history = [
+            {"role": "user", "content": long_content},
+            {"role": "assistant", "content": "OK"},
+        ]
+        truncated = "A" * 150 + "..."  # agent truncated
+        result = {
+            "messages": [
+                {"role": "user", "content": truncated, "timestamp": 1},
+                {"role": "assistant", "content": "OK", "timestamp": 2},
+                {"role": "user", "content": "Next", "timestamp": 3},
+                {"role": "assistant", "content": "Done", "timestamp": 4},
+            ]
+        }
+        # First 100 chars of expected content match the truncated version
+        assert APIServerAdapter._response_messages_turn_start_index(
+            history, "Next", result
+        ) == 3
+
+    def test_no_match_returns_zero(self):
+        """No prefix match at all — return 0 (use full messages)."""
+        history = [
+            {"role": "user", "content": "Completely different"},
+        ]
+        result = {
+            "messages": [
+                {"role": "user", "content": "Something else"},
+                {"role": "assistant", "content": "???"},
+            ]
+        }
+        assert APIServerAdapter._response_messages_turn_start_index(
+            history, "Something else", result
+        ) == 0
+
+    def test_empty_messages_returns_zero(self):
+        """Empty or missing messages list — return 0."""
+        assert APIServerAdapter._response_messages_turn_start_index([], "Hi", {"messages": []}) == 0
+        assert APIServerAdapter._response_messages_turn_start_index([], "Hi", {}) == 0
+        assert APIServerAdapter._response_messages_turn_start_index([], "Hi", {"messages": None}) == 0
+
+    def test_turn_transcript_messages_returns_current_turn_only(self):
+        """_turn_transcript_messages returns only the current turn, not full history."""
+        history = [
+            {"role": "user", "content": "Hi"},
+            {"role": "assistant", "content": "Hello!"},
+        ]
+        result = {
+            "messages": [
+                {"role": "user", "content": "Hi", "timestamp": 100},
+                {"role": "assistant", "content": "Hello!", "timestamp": 101},
+                {"role": "user", "content": "What is 2+2?", "timestamp": 102},
+                {"role": "assistant", "content": "4", "timestamp": 103},
+            ]
+        }
+        turn = APIServerAdapter._turn_transcript_messages(history, "What is 2+2?", result)
+        # Only the assistant's "4" reply should be in the turn transcript
+        assert len(turn) == 1
+        assert turn[0].get("content") == "4"
+
+
 # ---------------------------------------------------------------------------
 # Response-side truncation / failure handling (issue #22496)
 # ---------------------------------------------------------------------------
diff --git a/tests/gateway/test_discord_typing_max_lifetime.py b/tests/gateway/test_discord_typing_max_lifetime.py
new file mode 100644
index 0000000000000..72acedc071be9
--- /dev/null
+++ b/tests/gateway/test_discord_typing_max_lifetime.py
@@ -0,0 +1,129 @@
+"""Tests for DiscordAdapter typing-loop max-lifetime deadline guard.
+
+Issue #90151: the persistent typing loop in ``DiscordAdapter.send_typing``
+has no natural exit condition other than ``stop_typing()`` or a non-429
+error. If ``stop_typing`` never reaches the adapter (e.g. a crashed run,
+or a thread-vs-parent-channel key mismatch), the loop runs forever and
+Discord keeps showing the "…is typing" badge until the gateway restarts.
+
+The fix adds a configurable max-lifetime deadline
+(``discord.typing_loop_max_seconds``, default 600s, 0 disables it).
+"""
+
+import asyncio
+import sys
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import PlatformConfig
+
+
+def _ensure_discord_mock():
+    if "discord" in sys.modules and hasattr(sys.modules["discord"], "__file__"):
+        return
+
+    discord_mod = MagicMock()
+    discord_mod.Intents.default.return_value = MagicMock()
+    discord_mod.Client = MagicMock
+    discord_mod.File = MagicMock
+    discord_mod.DMChannel = type("DMChannel", (), {})
+    discord_mod.Thread = type("Thread", (), {})
+    discord_mod.ForumChannel = type("ForumChannel", (), {})
+    discord_mod.ui = SimpleNamespace(
+        View=object, button=lambda *a, **k: (lambda fn: fn), Button=object
+    )
+    discord_mod.ButtonStyle = SimpleNamespace(
+        success=1, primary=2, secondary=2, danger=3, green=1, grey=2, blurple=2, red=3
+    )
+    discord_mod.Color = SimpleNamespace(
+        orange=lambda: 1, green=lambda: 2, blue=lambda: 3, red=lambda: 4, purple=lambda: 5
+    )
+    discord_mod.Interaction = object
+    discord_mod.Embed = MagicMock
+    discord_mod.app_commands = SimpleNamespace(
+        describe=lambda **kwargs: (lambda fn: fn),
+        choices=lambda **kwargs: (lambda fn: fn),
+        Choice=lambda **kwargs: SimpleNamespace(**kwargs),
+    )
+
+    ext_mod = MagicMock()
+    commands_mod = MagicMock()
+    commands_mod.Bot = MagicMock
+    ext_mod.commands = commands_mod
+
+    sys.modules.setdefault("discord", discord_mod)
+    sys.modules.setdefault("discord.ext", ext_mod)
+    sys.modules.setdefault("discord.ext.commands", commands_mod)
+
+
+_ensure_discord_mock()
+
+from types import SimpleNamespace  # noqa: E402
+
+from plugins.platforms.discord.adapter import DiscordAdapter  # noqa: E402
+
+
+def _make_adapter(max_seconds: int = 600):
+    """Build a DiscordAdapter with a mocked client ready to start typing."""
+    adapter = DiscordAdapter(PlatformConfig(enabled=True, token="test"))
+    adapter._client = MagicMock()
+    adapter._client.http = MagicMock()
+    adapter._typing_tasks = {}
+    adapter._typing_loop_max_seconds = max_seconds
+    return adapter
+
+
+class TestTypingLoopMaxLifetime:
+    @pytest.mark.asyncio
+    async def test_loop_expires_after_max_lifetime(self):
+        """The typing loop must stop once the max-lifetime deadline elapses.
+
+        Uses a very short deadline (1s) so the test finishes quickly, and a
+        mocked client whose request resolves instantly so the loop spins at
+        full speed — each iteration hits the deadline check first.
+        """
+        adapter = _make_adapter(max_seconds=1)
+        adapter._client.http.request = AsyncMock()
+
+        await adapter.send_typing("channel-1")
+        assert "channel-1" in adapter._typing_tasks
+
+        # Wait long enough for the deadline to elapse and the loop to exit.
+        await asyncio.sleep(1.5)
+
+        # The loop should have removed itself from the registry.
+        assert "channel-1" not in adapter._typing_tasks
+
+    @pytest.mark.asyncio
+    async def test_stop_typing_still_works_with_deadline(self):
+        """stop_typing must still cancel the loop cleanly before deadline."""
+        adapter = _make_adapter(max_seconds=600)
+        adapter._client.http.request = AsyncMock()
+
+        await adapter.send_typing("channel-2")
+        assert "channel-2" in adapter._typing_tasks
+
+        await adapter.stop_typing("channel-2")
+        assert "channel-2" not in adapter._typing_tasks
+
+    @pytest.mark.asyncio
+    async def test_zero_max_seconds_disables_deadline(self):
+        """Setting max_seconds to 0 must disable the deadline guard.
+
+        The loop runs indefinitely (until stop_typing) — verified by letting
+        it spin well past where a 1s deadline would have fired and
+        asserting it's still alive.
+        """
+        adapter = _make_adapter(max_seconds=0)
+        adapter._client.http.request = AsyncMock()
+
+        await adapter.send_typing("channel-3")
+        # Give it enough time that a 1s deadline would have expired.
+        await asyncio.sleep(1.5)
+        assert "channel-3" in adapter._typing_tasks
+
+        # Clean up.
+        await adapter.stop_typing("channel-3")
+        assert "channel-3" not in adapter._typing_tasks
\ No newline at end of file
diff --git a/tests/hermes_cli/test_mcp_catalog.py b/tests/hermes_cli/test_mcp_catalog.py
index bd9f19be259f6..15de2e83c79fe 100644
--- a/tests/hermes_cli/test_mcp_catalog.py
+++ b/tests/hermes_cli/test_mcp_catalog.py
@@ -271,6 +271,87 @@ def test_install_with_api_key_prompts_and_saves(self, catalog_dir, monkeypatch):
         assert get_env_value("DEMO_KEY") == "secret-val"
         assert "demo" in load_config()["mcp_servers"]
 
+    def test_install_stdio_api_key_wires_env_references(self, catalog_dir, monkeypatch):
+        """stdio + api_key: auth.env must reach the generated MCP config as
+        env-backed references so the child process receives the credentials.
+
+        Regression test for #89316 — before the fix, _build_server_config
+        dropped auth.env entirely for stdio transports, so the stdio child
+        started without its API key even though install_entry() had saved it
+        to .env.
+        """
+        body = _basic_manifest(
+            name="example",
+            transport={"type": "stdio", "command": "/bin/true", "args": []},
+            auth={
+                "type": "api_key",
+                "env": [
+                    {"name": "EXAMPLE_BASE_URL", "prompt": "URL", "secret": False},
+                    {"name": "EXAMPLE_API_KEY", "prompt": "key", "secret": True},
+                ],
+            },
+        )
+        _write_manifest(catalog_dir, "example", body)
+
+        from hermes_cli import mcp_catalog
+
+        monkeypatch.setattr(
+            mcp_catalog, "_prompt_input", lambda prompt, **kw: "secret-val"
+        )
+
+        from hermes_cli.mcp_catalog import install_entry
+        from hermes_cli.config import get_config_path, load_config
+
+        install_entry(_entry("example"), enable=True)
+
+        server = load_config()["mcp_servers"]["example"]
+        assert server["command"] == "/bin/true"
+        # load_config resolves ${VAR} from .env — verify the resolved values
+        # reach the config (proving the template wired them through).
+        assert server["env"] == {
+            "EXAMPLE_BASE_URL": "secret-val",
+            "EXAMPLE_API_KEY": "secret-val",
+        }
+
+        # The raw file must carry ${...} templates, never the secret itself.
+        raw = get_config_path().read_text()
+        assert "${EXAMPLE_API_KEY}" in raw
+        assert "secret-val" not in raw
+
+    def test_install_stdio_api_key_merges_with_transport_env(
+        self, catalog_dir, monkeypatch
+    ):
+        """When both transport.env and auth.env exist, both must be present."""
+        body = _basic_manifest(
+            name="example",
+            transport={
+                "type": "stdio",
+                "command": "/bin/true",
+                "args": [],
+                "env": {"DEBUG": "1"},
+            },
+            auth={
+                "type": "api_key",
+                "env": [{"name": "EXAMPLE_KEY", "prompt": "key", "secret": True}],
+            },
+        )
+        _write_manifest(catalog_dir, "example", body)
+
+        from hermes_cli import mcp_catalog
+
+        monkeypatch.setattr(
+            mcp_catalog, "_prompt_input", lambda prompt, **kw: "secret-val"
+        )
+
+        from hermes_cli.mcp_catalog import install_entry
+        from hermes_cli.config import load_config
+
+        install_entry(_entry("example"), enable=True)
+
+        env = load_config()["mcp_servers"]["example"]["env"]
+        assert env["DEBUG"] == "1"
+        assert env["EXAMPLE_KEY"] == "secret-val"
+
     def test_install_http_api_key_writes_bearer_headers(self, catalog_dir, monkeypatch):
         body = _basic_manifest(
             transport={"type": "http", "url": "https://mcp.example.com/sse"},