diff --git a/gateway/platforms/slack.py b/gateway/platforms/slack.py
index 843fb78959cef..82ed959a03c40 100644
--- a/gateway/platforms/slack.py
+++ b/gateway/platforms/slack.py
@@ -290,6 +290,7 @@ def __init__(self, config: PlatformConfig):
         self._bot_user_id: Optional[str] = None
         self._user_name_cache: Dict[str, str] = {}  # user_id → display name
         self._socket_mode_task: Optional[asyncio.Task] = None
+        self._raw_socket_mode_task: Optional[asyncio.Task] = None
         # Multi-workspace support
         self._team_clients: Dict[str, Any] = {}   # team_id → WebClient
         self._team_bot_user_ids: Dict[str, str] = {}          # team_id → bot_user_id
@@ -660,12 +661,22 @@ async def handle_hermes_command(ack, command):
             ):
                 self._app.action(_action_id)(self._handle_slash_confirm_action)
 
-            # Start Socket Mode handler in background
+            raw_socket_mode_enabled = self._raw_socket_mode_fallback_enabled()
+
+            # Start Socket Mode handler in background. The raw fallback is an
+            # optional second reader for deployments where Bolt's background
+            # Socket Mode task goes stale or misses an envelope. Event dedup in
+            # _handle_slack_message suppresses duplicate replies.
             self._handler = AsyncSocketModeHandler(self._app, app_token, proxy=proxy_url)
             _apply_slack_proxy(self._handler.client, proxy_url)
             self._socket_mode_task = asyncio.create_task(self._handler.start_async())
 
             self._running = True
+            if raw_socket_mode_enabled:
+                self._raw_socket_mode_task = asyncio.create_task(
+                    self._raw_socket_mode_fallback(app_token, proxy_url)
+                )
+                logger.warning("[Slack] Raw Socket Mode fallback enabled")
             logger.info(
                 "[Slack] Socket Mode connected (%d workspace(s))",
                 len(self._team_clients),
@@ -681,6 +692,16 @@ async def handle_hermes_command(ack, command):
 
     async def disconnect(self) -> None:
         """Disconnect from Slack."""
+        if self._raw_socket_mode_task:
+            self._raw_socket_mode_task.cancel()
+            try:
+                await self._raw_socket_mode_task
+            except asyncio.CancelledError:
+                pass
+            except Exception as e:  # pragma: no cover - defensive logging
+                logger.warning("[Slack] Raw Socket Mode fallback close error: %s", e, exc_info=True)
+            finally:
+                self._raw_socket_mode_task = None
         if self._handler:
             try:
                 await self._handler.close_async()
@@ -692,6 +713,106 @@ async def disconnect(self) -> None:
 
         logger.info("[Slack] Disconnected")
 
+    def _raw_socket_mode_fallback_enabled(self) -> bool:
+        configured = self.config.extra.get("socket_raw_fallback")
+        if configured is None:
+            configured = os.getenv("SLACK_SOCKET_RAW_FALLBACK", "")
+        return str(configured).strip().lower() in {"1", "true", "yes", "on"}
+
+    async def _raw_socket_mode_fallback(self, app_token: str, proxy_url: Optional[str]) -> None:
+        """Read Slack Socket Mode directly when Bolt's background task misses events."""
+        reconnect_delay = 5
+        while self._running:
+            try:
+                async with aiohttp.ClientSession() as session:
+                    async with session.post(
+                        "https://slack.com/api/apps.connections.open",
+                        headers={"Authorization": f"Bearer {app_token}"},
+                        proxy=proxy_url,
+                        timeout=aiohttp.ClientTimeout(total=20),
+                    ) as response:
+                        payload = await response.json()
+                    if not payload.get("ok") or not payload.get("url"):
+                        logger.warning(
+                            "[Slack] Raw Socket Mode fallback connection failed: %s",
+                            payload.get("error") or response.status,
+                        )
+                        await asyncio.sleep(reconnect_delay)
+                        continue
+
+                    async with session.ws_connect(
+                        payload["url"],
+                        heartbeat=20,
+                        proxy=proxy_url,
+                    ) as websocket:
+                        logger.warning("[Slack] Raw Socket Mode fallback connected")
+                        async for message in websocket:
+                            if not self._running:
+                                break
+                            if message.type == aiohttp.WSMsgType.TEXT:
+                                try:
+                                    envelope = json.loads(message.data)
+                                except json.JSONDecodeError:
+                                    continue
+                                envelope_id = envelope.get("envelope_id")
+                                if envelope_id:
+                                    await websocket.send_json({"envelope_id": envelope_id})
+                                event_payload = envelope.get("payload") or {}
+                                event = event_payload.get("event") or {}
+                                if event_payload.get("team_id") and not event.get("team"):
+                                    event["team"] = event_payload.get("team_id")
+                                event_type = event.get("type")
+                                channel_id = event.get("channel") or ""
+                                event_text = event.get("text") or ""
+                                bot_mentioned = (
+                                    self._bot_user_id
+                                    and f"<@{self._bot_user_id}>" in event_text
+                                )
+                                should_dispatch = (
+                                    event_type == "app_mention"
+                                    or (
+                                        event_type == "message"
+                                        and (
+                                            channel_id.startswith("D")
+                                            or bot_mentioned
+                                        )
+                                    )
+                                )
+                                if should_dispatch:
+                                    if (
+                                        event_type in {"message", "app_mention"}
+                                        and not channel_id.startswith("D")
+                                        and event.get("ts")
+                                        and (bot_mentioned or event_type == "app_mention")
+                                    ):
+                                        event["_hermes_dedup_key"] = f"mentioned:{event.get('ts')}"
+                                    logger.info(
+                                        "[Slack] Raw Socket Mode fallback dispatch: %s channel=%s ts=%s",
+                                        event_type,
+                                        channel_id,
+                                        event.get("ts"),
+                                    )
+                                    event_for_handler = dict(event)
+                                    task = asyncio.create_task(self._handle_slack_message(event_for_handler))
+                                    task.add_done_callback(self._log_raw_fallback_dispatch_result)
+                            elif message.type in {aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR}:
+                                break
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:  # pragma: no cover - network watchdog
+                if self._running:
+                    logger.warning("[Slack] Raw Socket Mode fallback error: %s", e, exc_info=True)
+            if self._running:
+                await asyncio.sleep(reconnect_delay)
+
+    def _log_raw_fallback_dispatch_result(self, task: asyncio.Task) -> None:
+        try:
+            task.result()
+        except asyncio.CancelledError:
+            pass
+        except Exception as e:  # pragma: no cover - diagnostic callback
+            logger.warning("[Slack] Raw Socket Mode fallback dispatch failed: %s", e, exc_info=True)
+
     def _get_client(self, chat_id: str) -> Any:
         """Return the workspace-specific WebClient for a channel."""
         team_id = self._channel_team.get(chat_id)
@@ -1710,8 +1831,28 @@ async def _handle_assistant_thread_lifecycle_event(self, event: dict) -> None:
 
     async def _handle_slack_message(self, event: dict) -> None:
         """Handle an incoming Slack message event."""
-        # Dedup: Slack Socket Mode can redeliver events after reconnects (#4777)
-        event_ts = event.get("ts", "")
+        # Dedup: Slack Socket Mode can redeliver events after reconnects (#4777).
+        # Slack can emit both `message` and `app_mention` envelopes for the same
+        # channel @mention. Normalize mentioned channel posts to one key so the
+        # agent only gets one turn even when Bolt or a raw Socket Mode fallback
+        # delivers both envelopes.
+        event_ts = event.get("_hermes_dedup_key")
+        if not event_ts:
+            raw_ts = event.get("ts", "")
+            raw_text = event.get("text", "")
+            raw_channel = event.get("channel", "")
+            raw_mentioned = bool(
+                raw_ts
+                and not raw_channel.startswith("D")
+                and (
+                    event.get("type") == "app_mention"
+                    or (
+                        self._bot_user_id
+                        and f"<@{self._bot_user_id}>" in raw_text
+                    )
+                )
+            )
+            event_ts = f"mentioned:{raw_ts}" if raw_mentioned else raw_ts
         if event_ts and self._dedup.is_duplicate(event_ts):
             return
 
@@ -1915,7 +2056,34 @@ async def _handle_slack_message(self, event: dict) -> None:
                         user_id=user_id,
                     )
                 )
-                if not reply_to_bot_thread and not in_mentioned_thread and not has_session:
+                parent_mentions_bot = False
+                if (
+                    is_thread_reply
+                    and event_thread_ts
+                    and not reply_to_bot_thread
+                    and not in_mentioned_thread
+                    and not has_session
+                ):
+                    parent_text = await self._fetch_thread_parent_text(
+                        channel_id=channel_id,
+                        thread_ts=event_thread_ts,
+                        team_id=team_id,
+                        strip_bot_mention=False,
+                    )
+                    parent_mentions_bot = bool(bot_uid and f"<@{bot_uid}>" in parent_text)
+                    if parent_mentions_bot:
+                        self._mentioned_threads.add(event_thread_ts)
+                        if len(self._mentioned_threads) > self._MENTIONED_THREADS_MAX:
+                            to_remove = list(self._mentioned_threads)[:self._MENTIONED_THREADS_MAX // 2]
+                            for t in to_remove:
+                                self._mentioned_threads.discard(t)
+
+                if (
+                    not reply_to_bot_thread
+                    and not in_mentioned_thread
+                    and not has_session
+                    and not parent_mentions_bot
+                ):
                     return
 
         if is_mentioned:
@@ -1925,8 +2093,8 @@ async def _handle_slack_message(self, event: dict) -> None:
             # Skipped in strict mode: strict_mention=true bots must be
             # re-mentioned every turn, so remembering the thread would
             # defeat the feature (and re-enable agent-to-agent ack loops).
-            if event_thread_ts and not self._slack_strict_mention():
-                self._mentioned_threads.add(event_thread_ts)
+            if thread_ts and not self._slack_strict_mention():
+                self._mentioned_threads.add(thread_ts)
                 if len(self._mentioned_threads) > self._MENTIONED_THREADS_MAX:
                     to_remove = list(self._mentioned_threads)[:self._MENTIONED_THREADS_MAX // 2]
                     for t in to_remove:
@@ -2638,9 +2806,13 @@ async def _fetch_thread_context(
             return ""
 
     async def _fetch_thread_parent_text(
-        self, channel_id: str, thread_ts: str, team_id: str = "",
+        self,
+        channel_id: str,
+        thread_ts: str,
+        team_id: str = "",
+        strip_bot_mention: bool = True,
     ) -> str:
-        """Return the raw text of the thread parent message (for reply_to_text).
+        """Return the text of the thread parent message.
 
         Uses the same per-thread cache as :meth:`_fetch_thread_context` to avoid
         hitting ``conversations.replies`` twice. Falls back to a cheap single-
@@ -2671,7 +2843,7 @@ async def _fetch_thread_parent_text(
                 return ""
             bot_uid = self._team_bot_user_ids.get(team_id, self._bot_user_id)
             text = (parent.get("text") or "").strip()
-            if bot_uid:
+            if strip_bot_mention and bot_uid:
                 text = text.replace(f"<@{bot_uid}>", "").strip()
             return text
         except Exception as exc:  # pragma: no cover - defensive
diff --git a/gateway/run.py b/gateway/run.py
index ba40995b859eb..1caa6cc3fd1f5 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -3132,9 +3132,10 @@ def _schedule_resume_pending_sessions(self) -> int:
         ``resume_pending`` already preserves the transcript AND the existing
         ``_is_resume_pending`` branch in ``_handle_message_with_agent``
         injects a reason-aware recovery system note on the next turn.  This
-        method closes the UX gap by synthesizing that next turn once
-        adapters are back online — the event text is empty so the existing
-        injection path owns the wording and we never double up.
+        method closes the UX gap by synthesizing that next turn once adapters
+        are back online.  When the interrupted user message was captured before
+        shutdown, replay it; otherwise send a nonblank continuation instruction
+        so the model does not treat recovery as a user-sent empty message.
 
         Adapters that are not yet ready (adapter missing from
         ``self.adapters``) are skipped silently; their sessions stay
@@ -3173,11 +3174,18 @@ def _schedule_resume_pending_sessions(self) -> int:
                 )
                 continue
 
-            # Empty-text internal event — the _is_resume_pending branch in
-            # _handle_message_with_agent prepends the proper reason-aware
-            # system note before the turn runs.
+            resume_text = (getattr(entry, "in_flight_user_message", None) or "").strip()
+            if not resume_text:
+                resume_text = (
+                    "[Internal gateway auto-resume: continue the interrupted "
+                    "turn from the existing conversation history. This is not "
+                    "a user-sent blank message. Do not tell the user their "
+                    "message came through empty or ask them to repeat the same "
+                    "command.]"
+                )
+
             event = MessageEvent(
-                text="",
+                text=resume_text,
                 message_type=MessageType.TEXT,
                 source=source,
                 internal=True,
@@ -7102,6 +7110,11 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
         )
         if message_text is None:
             return
+        if session_key:
+            try:
+                self.session_store.mark_in_flight(session_key, message_text)
+            except Exception as _e:
+                logger.debug("mark_in_flight failed for %s: %s", session_key, _e)
 
         # Bind this gateway run generation to the adapter's active-session
         # event so deferred post-delivery callbacks can be released by the
@@ -7199,6 +7212,13 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                         "clear_resume_pending failed for %s: %s",
                         session_key, _e,
                     )
+                try:
+                    self.session_store.clear_in_flight(session_key)
+                except Exception as _e:
+                    logger.debug(
+                        "clear_in_flight failed for %s: %s",
+                        session_key, _e,
+                    )
 
             # Normalize empty responses: surface errors, partial failures, and
             # the case where agent did work but returned no text. Fix for #18765.
diff --git a/gateway/session.py b/gateway/session.py
index be393e48e6fc8..751ef046e979e 100644
--- a/gateway/session.py
+++ b/gateway/session.py
@@ -490,6 +490,8 @@ class SessionEntry:
     resume_pending: bool = False
     resume_reason: Optional[str] = None  # e.g. "restart_timeout"
     last_resume_marked_at: Optional[datetime] = None
+    in_flight_user_message: Optional[str] = None
+    in_flight_marked_at: Optional[datetime] = None
 
     def to_dict(self) -> Dict[str, Any]:
         result = {
@@ -517,6 +519,12 @@ def to_dict(self) -> Dict[str, Any]:
                 if self.last_resume_marked_at
                 else None
             ),
+            "in_flight_user_message": self.in_flight_user_message,
+            "in_flight_marked_at": (
+                self.in_flight_marked_at.isoformat()
+                if self.in_flight_marked_at
+                else None
+            ),
             "is_fresh_reset": self.is_fresh_reset,
         }
         if self.origin:
@@ -543,6 +551,13 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionEntry":
                 last_resume_marked_at = datetime.fromisoformat(_lrma)
             except (TypeError, ValueError):
                 last_resume_marked_at = None
+        in_flight_marked_at = None
+        _ifma = data.get("in_flight_marked_at")
+        if _ifma:
+            try:
+                in_flight_marked_at = datetime.fromisoformat(_ifma)
+            except (TypeError, ValueError):
+                in_flight_marked_at = None
 
         return cls(
             session_key=data["session_key"],
@@ -566,6 +581,8 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionEntry":
             resume_pending=data.get("resume_pending", False),
             resume_reason=data.get("resume_reason"),
             last_resume_marked_at=last_resume_marked_at,
+            in_flight_user_message=data.get("in_flight_user_message"),
+            in_flight_marked_at=in_flight_marked_at,
             is_fresh_reset=data.get("is_fresh_reset", False),
         )
 
@@ -1028,6 +1045,39 @@ def clear_resume_pending(self, session_key: str) -> bool:
             self._save()
             return True
 
+    def mark_in_flight(self, session_key: str, message: str) -> bool:
+        """Remember the user turn currently being processed by the agent.
+
+        Gateway restart recovery can synthesize an internal resume event before
+        the interrupted user turn has been persisted to the transcript.  Keeping
+        this text on the session entry lets startup auto-resume replay the real
+        pending turn instead of delivering a blank message to the model.
+        """
+        with self._lock:
+            self._ensure_loaded_locked()
+            entry = self._entries.get(session_key)
+            if entry is None:
+                return False
+            entry.in_flight_user_message = message
+            entry.in_flight_marked_at = _now()
+            self._save()
+            return True
+
+    def clear_in_flight(self, session_key: str) -> bool:
+        """Clear the remembered in-flight user turn after completion."""
+        with self._lock:
+            self._ensure_loaded_locked()
+            entry = self._entries.get(session_key)
+            if entry is None or (
+                entry.in_flight_user_message is None
+                and entry.in_flight_marked_at is None
+            ):
+                return False
+            entry.in_flight_user_message = None
+            entry.in_flight_marked_at = None
+            self._save()
+            return True
+
     def prune_old_entries(self, max_age_days: int) -> int:
         """Drop SessionEntry records older than max_age_days.
 
diff --git a/tests/gateway/test_restart_resume_pending.py b/tests/gateway/test_restart_resume_pending.py
index 13ef2f6f99ec9..a812de9f0b481 100644
--- a/tests/gateway/test_restart_resume_pending.py
+++ b/tests/gateway/test_restart_resume_pending.py
@@ -189,6 +189,8 @@ def test_defaults(self):
         assert entry.resume_pending is False
         assert entry.resume_reason is None
         assert entry.last_resume_marked_at is None
+        assert entry.in_flight_user_message is None
+        assert entry.in_flight_marked_at is None
 
     def test_roundtrip_with_resume_fields(self):
         now = datetime(2026, 4, 18, 12, 0, 0)
@@ -200,11 +202,15 @@ def test_roundtrip_with_resume_fields(self):
             resume_pending=True,
             resume_reason="restart_timeout",
             last_resume_marked_at=now,
+            in_flight_user_message="[User] run",
+            in_flight_marked_at=now,
         )
         restored = SessionEntry.from_dict(entry.to_dict())
         assert restored.resume_pending is True
         assert restored.resume_reason == "restart_timeout"
         assert restored.last_resume_marked_at == now
+        assert restored.in_flight_user_message == "[User] run"
+        assert restored.in_flight_marked_at == now
 
     def test_from_dict_legacy_without_resume_fields(self):
         """Old sessions.json without the new fields deserialize cleanly."""
@@ -220,6 +226,8 @@ def test_from_dict_legacy_without_resume_fields(self):
         assert restored.resume_pending is False
         assert restored.resume_reason is None
         assert restored.last_resume_marked_at is None
+        assert restored.in_flight_user_message is None
+        assert restored.in_flight_marked_at is None
 
     def test_malformed_timestamp_is_tolerated(self):
         now = datetime.now()
@@ -231,12 +239,16 @@ def test_malformed_timestamp_is_tolerated(self):
             "resume_pending": True,
             "resume_reason": "restart_timeout",
             "last_resume_marked_at": "not-a-timestamp",
+            "in_flight_user_message": "run",
+            "in_flight_marked_at": "also-not-a-timestamp",
         }
         restored = SessionEntry.from_dict(data)
         # resume_pending still honoured, only the broken timestamp drops
         assert restored.resume_pending is True
         assert restored.resume_reason == "restart_timeout"
         assert restored.last_resume_marked_at is None
+        assert restored.in_flight_user_message == "run"
+        assert restored.in_flight_marked_at is None
 
 
 # ---------------------------------------------------------------------------
@@ -319,6 +331,41 @@ def test_returns_false_for_unknown_key(self, tmp_path):
         assert store.clear_resume_pending("no-such-key") is False
 
 
+class TestInFlightUserMessage:
+    def test_mark_and_clear_in_flight(self, tmp_path):
+        store = _make_store(tmp_path)
+        source = _make_source()
+        entry = store.get_or_create_session(source)
+
+        assert store.mark_in_flight(entry.session_key, "[User] run") is True
+        marked = store._entries[entry.session_key]
+        assert marked.in_flight_user_message == "[User] run"
+        assert marked.in_flight_marked_at is not None
+
+        assert store.clear_in_flight(entry.session_key) is True
+        cleared = store._entries[entry.session_key]
+        assert cleared.in_flight_user_message is None
+        assert cleared.in_flight_marked_at is None
+
+    def test_in_flight_survives_roundtrip(self, tmp_path):
+        store = _make_store(tmp_path)
+        source = _make_source()
+        entry = store.get_or_create_session(source)
+
+        store.mark_in_flight(entry.session_key, "check recent inbound messages")
+
+        store2 = _make_store(tmp_path)
+        store2._ensure_loaded()
+        reloaded = store2._entries[entry.session_key]
+        assert reloaded.in_flight_user_message == "check recent inbound messages"
+        assert reloaded.in_flight_marked_at is not None
+
+    def test_returns_false_for_unknown_key(self, tmp_path):
+        store = _make_store(tmp_path)
+        assert store.mark_in_flight("no-such-key", "run") is False
+        assert store.clear_in_flight("no-such-key") is False
+
+
 # ---------------------------------------------------------------------------
 # SessionStore.get_or_create_session resume_pending behaviour
 # ---------------------------------------------------------------------------
@@ -953,6 +1000,8 @@ async def test_startup_auto_resume_schedules_fresh_pending_sessions():
         resume_pending=True,
         resume_reason="restart_timeout",
         last_resume_marked_at=datetime.now(),
+        in_flight_user_message='[Replying to: "check recent inbound messages"]\n\n[User] run',
+        in_flight_marked_at=datetime.now(),
     )
     runner.session_store._entries = {pending_entry.session_key: pending_entry}
     adapter.handle_message = AsyncMock()
@@ -967,10 +1016,38 @@ async def test_startup_auto_resume_schedules_fresh_pending_sessions():
     assert event.internal is True
     assert event.message_type == MessageType.TEXT
     assert event.source == source
-    # Text is empty — the existing _is_resume_pending branch in
-    # _handle_message_with_agent owns the system-note injection so we don't
-    # double it up.
-    assert event.text == ""
+    assert event.text == '[Replying to: "check recent inbound messages"]\n\n[User] run'
+
+
+@pytest.mark.asyncio
+async def test_startup_auto_resume_uses_nonblank_fallback_without_in_flight_text():
+    """Fallback recovery text must not look like a user-sent empty message."""
+    runner, adapter = make_restart_runner()
+    source = make_restart_source(chat_id="resume-chat", thread_id="topic-1")
+    pending_entry = SessionEntry(
+        session_key="agent:main:telegram:group:resume-chat:topic-1",
+        session_id="sid",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        origin=source,
+        platform=Platform.TELEGRAM,
+        chat_type="group",
+        resume_pending=True,
+        resume_reason="restart_timeout",
+        last_resume_marked_at=datetime.now(),
+    )
+    runner.session_store._entries = {pending_entry.session_key: pending_entry}
+    adapter.handle_message = AsyncMock()
+
+    scheduled = runner._schedule_resume_pending_sessions()
+    await asyncio.sleep(0)
+
+    assert scheduled == 1
+    event = adapter.handle_message.await_args.args[0]
+    assert event.internal is True
+    assert event.text
+    assert "Internal gateway auto-resume" in event.text
+    assert "not a user-sent blank message" in event.text
 
 
 @pytest.mark.asyncio
diff --git a/tests/gateway/test_slack.py b/tests/gateway/test_slack.py
index 478370d8c414a..c825bc40cf738 100644
--- a/tests/gateway/test_slack.py
+++ b/tests/gateway/test_slack.py
@@ -281,6 +281,50 @@ def decorator(fn): return fn
         assert adapter._handler is second_handler
 
 
+class TestMentionDedup:
+    @pytest.mark.asyncio
+    async def test_message_and_app_mention_same_ts_process_once(self, adapter):
+        """Slack can deliver both envelopes for one @mention; process one turn."""
+        adapter.config.extra["allow_bots"] = "mentions"
+        shared = {
+            "text": "<@U_BOT> human-authored smoke",
+            "user": "U_USER",
+            "bot_id": "B_SMOKE_HARNESS",
+            "channel": "C123",
+            "channel_type": "channel",
+            "team": "T123",
+            "ts": "1778771432.395979",
+        }
+
+        with patch.object(adapter, "_resolve_user_name", new_callable=AsyncMock) as resolve:
+            resolve.return_value = "Kai Yi"
+            await adapter._handle_slack_message({**shared, "type": "message"})
+            await adapter._handle_slack_message({**shared, "type": "app_mention"})
+
+        adapter.handle_message.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_app_mention_and_message_same_ts_process_once(self, adapter):
+        """Dedup is order-independent for app_mention/message pairs."""
+        adapter.config.extra["allow_bots"] = "mentions"
+        shared = {
+            "text": "<@U_BOT> human-authored smoke",
+            "user": "U_USER",
+            "bot_id": "B_SMOKE_HARNESS",
+            "channel": "C123",
+            "channel_type": "channel",
+            "team": "T123",
+            "ts": "1778771432.395979",
+        }
+
+        with patch.object(adapter, "_resolve_user_name", new_callable=AsyncMock) as resolve:
+            resolve.return_value = "Kai Yi"
+            await adapter._handle_slack_message({**shared, "type": "app_mention"})
+            await adapter._handle_slack_message({**shared, "type": "message"})
+
+        adapter.handle_message.assert_awaited_once()
+
+
 # ---------------------------------------------------------------------------
 # TestSlackProxyBehavior
 # ---------------------------------------------------------------------------
@@ -1922,6 +1966,9 @@ async def test_thread_reply_without_mention_no_session_ignored(
     ):
         """Thread replies without mention should be ignored if no active session."""
         mock_session_store._entries = {}  # No active sessions
+        adapter_with_session_store._app.client.conversations_replies = AsyncMock(
+            return_value={"messages": [{"ts": "123.000", "text": "Parent without mention"}]}
+        )
 
         event = {
             "text": "Just replying in the thread",
@@ -1935,6 +1982,58 @@ async def test_thread_reply_without_mention_no_session_ignored(
         await adapter_with_session_store._handle_slack_message(event)
         adapter_with_session_store.handle_message.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_thread_reply_routes_when_parent_mentioned_bot(
+        self, adapter_with_session_store, mock_session_store
+    ):
+        """A plain thread reply should route when the thread parent mentioned the bot."""
+        mock_session_store._entries = {}
+        adapter_with_session_store._app.client.conversations_replies = AsyncMock(side_effect=[
+            {
+                "messages": [
+                    {
+                        "ts": "123.000",
+                        "user": "U_USER",
+                        "text": "<@U_BOT> check this and ask me for run",
+                    },
+                ],
+            },
+            {
+                "messages": [
+                    {
+                        "ts": "123.000",
+                        "user": "U_USER",
+                        "text": "<@U_BOT> check this and ask me for run",
+                    },
+                    {
+                        "ts": "123.456",
+                        "user": "U_USER",
+                        "text": "run",
+                    },
+                ],
+            },
+        ])
+
+        event = {
+            "text": "run",
+            "user": "U_USER",
+            "channel": "C123",
+            "ts": "123.456",
+            "thread_ts": "123.000",
+            "channel_type": "channel",
+            "team": "T_TEAM",
+        }
+        with patch.object(
+            adapter_with_session_store, "_resolve_user_name", new=AsyncMock(return_value="Kai Yi")
+        ):
+            await adapter_with_session_store._handle_slack_message(event)
+
+        adapter_with_session_store.handle_message.assert_called_once()
+        msg_event = adapter_with_session_store.handle_message.call_args[0][0]
+        assert msg_event.text.endswith("run")
+        assert "check this and ask me for run" in msg_event.text
+        assert "123.000" in adapter_with_session_store._mentioned_threads
+
     @pytest.mark.asyncio
     async def test_thread_reply_without_mention_with_session_processed(
         self, adapter_with_session_store, mock_session_store
@@ -2012,6 +2111,9 @@ async def test_no_session_store_ignores_thread_replies(
     ):
         """If no session store is attached, thread replies without mention should be ignored."""
         # adapter fixture has no session store attached
+        adapter._app.client.conversations_replies = AsyncMock(
+            return_value={"messages": [{"ts": "123.000", "text": "Parent without mention"}]}
+        )
         event = {
             "text": "Thread reply without mention",
             "user": "U_USER",