diff --git a/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx b/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx
index 6a6eadf865..b37a14cdff 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx
+++ b/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx
@@ -846,7 +846,7 @@ const en = {
   "lark.turnTrigger": "When to respond",
   "lark.triggerAddressed": "Only when mentioned or replied to",
   "lark.triggerHuman": "Respond to group members without @",
-  "lark.triggerDescription": "Everyone in this connected group can start a conversation. Bot messages and historical backfill stay as context. Requires the Lark app permission to receive all group messages; tool and delegation permissions stay unchanged.",
+  "lark.triggerDescription": "Everyone in the connected group can start a conversation. A native @ of the bound Bot can adjust an active Turn. Bot messages and history stay as context. Receiving all group messages requires the corresponding Lark permission.",
   "lark.readGoalScope": "Goal evidence access",
   "lark.readAllGoals": "All registered Goals, including new ones",
   "lark.readSelectedGoals": "Keep the selected Goal scope",
@@ -856,9 +856,9 @@ const en = {
   "lark.ingressLegacy": "Upgrade available",
   "lark.ingressLegacyDescription": "Save this connection to upgrade to the Agent inbox. Existing messages remain in the same Topic.",
   "lark.ingressQueue": "Queuing",
-  "lark.ingressQueueDescription": "Enter the same Agent Session's bounded FIFO queue and run after the current Turn.",
+  "lark.ingressQueueDescription": "Queue ordinary messages. A native @ adds text to a supported active Turn.",
   "lark.ingressSteering": "Steering",
-  "lark.ingressSteeringDescription": "Deliver to the active Turn and reject safely when no exact active Turn exists.",
+  "lark.ingressSteeringDescription": "Steer eligible messages. A native @ while idle starts a new Turn.",
   "lark.loading": "Loading Lark configuration…",
   "lark.management": "Lark management",
   "lark.openEventSettings": "Open Feishu event settings",
@@ -2509,7 +2509,7 @@ const zhCN: Record<WorkspaceMessageKey, string> = {
   "lark.turnTrigger": "何时回应",
   "lark.triggerAddressed": "仅 @ 或回复管家时",
   "lark.triggerHuman": "群成员直接发消息，无需 @",
-  "lark.triggerDescription": "此群所有成员都可以发起对话；机器人消息和补拉历史只作背景。需要飞书应用具备接收群内所有消息的权限；工具权限和委派授权保持不变。",
+  "lark.triggerDescription": "群成员可发起对话，原生 @ 已绑定的 Bot 可补充当前回合。机器人消息和补拉历史只作背景。接收群内所有消息需要相应飞书权限。",
   "lark.readGoalScope": "Goal 证据读取范围",
   "lark.readAllGoals": "全部已注册 Goal（含后续新增）",
   "lark.readSelectedGoals": "保留选定 Goal 范围",
@@ -2519,9 +2519,9 @@ const zhCN: Record<WorkspaceMessageKey, string> = {
   "lark.ingressLegacy": "待升级",
   "lark.ingressLegacyDescription": "保存连接即可升级为 Agent 异步收件箱，已有消息保留在原 Topic 中。",
   "lark.ingressQueue": "排队",
-  "lark.ingressQueueDescription": "进入同一 Agent Session 的有界 FIFO 队列，当前 Turn 完成后处理。",
+  "lark.ingressQueueDescription": "普通消息排队。原生 @ Bot 可向当前回合追加文字。",
   "lark.ingressSteering": "实时引导",
-  "lark.ingressSteeringDescription": "投到当前活跃 Turn；没有精确活跃 Turn 时安全拒绝。",
+  "lark.ingressSteeringDescription": "符合条件的消息补充当前回合。空闲时原生 @ Bot 可发起新回合。",
   "lark.loading": "加载 Lark 配置…",
   "lark.management": "Lark 管理",
   "lark.openEventSettings": "查看飞书事件配置",
diff --git a/loopx/capabilities/native_chat/queued_admission.py b/loopx/capabilities/native_chat/queued_admission.py
index 3f8c3fd814..7c3c30a76c 100644
--- a/loopx/capabilities/native_chat/queued_admission.py
+++ b/loopx/capabilities/native_chat/queued_admission.py
@@ -65,6 +65,73 @@ def enqueue_turn(controller: Any, *, session_id: str, client_turn_id: str, messa
     return turn, created
 
 
+def deliver_native_mention(
+    controller, *, session_id: str, ingress_id: str, target_turn_id: str | None,
+    binding_matches: bool, text_only: bool, may_deliver_new: bool,
+    steering_text: str, queued_text: str, origin: str,
+) -> dict[str, Any]:
+    """Project Chat receipts through the shared mention ingress rule."""
+    from ...control_plane.chat_turn_acceptance import plan_native_mention_ingress
+
+    def plan():
+        receipt = controller.store.steering_receipt(session_id, ingress_id) if binding_matches else None
+        queued_turn_exists = (
+            binding_matches and receipt is None
+            and controller.store.turn_for_client(session_id, ingress_id) is not None
+        )
+        message = queued_text if (receipt or {}).get("mode") == "session_queue" else steering_text
+        return plan_native_mention_ingress(
+            mentioned=True, binding_matches=binding_matches, text_only=text_only, may_deliver_new=may_deliver_new,
+            target_turn_id=target_turn_id, queued_turn_exists=queued_turn_exists,
+            receipt=({key: receipt.get(key) for key in ("mode", "status", "expected_turn_id", "active_turn_id", "error_code")}
+                     if receipt is not None else None),
+            receipt_message_matches=receipt is not None and receipt.get("message") == message,
+        )
+
+    try:
+        decision = plan()
+        if decision["action"] == "enqueue":
+            # Reserve the delivery mode before creating a FIFO Turn. Replays
+            # use this mode even if a different Turn is now running.
+            controller.store.create_ingress_receipt(
+                session_id, client_ingress_id=ingress_id, mode="session_queue", message=queued_text,
+            )
+        elif decision["action"] == "steer":
+            try:
+                controller.steer_active_turn(
+                    session_id=session_id, client_ingress_id=ingress_id, message=steering_text,
+                    expected_turn_id=decision["target_turn_id"], origin=origin,
+                )
+            except Exception:
+                # The native response can be lost after sending. The existing
+                # receipt, rather than the exception, decides whether it landed.
+                pass
+            decision = plan()
+            if decision["action"] == "steer":
+                return {"action": "unresolved"}
+        return decision
+    except (OSError, ValueError):
+        return {"action": "unresolved"}
+
+
+def record_steering_delivery(
+    store, *, session_id: str, turn_id: str, ingress_id: str,
+    message: str, queued_turn_id: str | None, origin: str | None,
+) -> None:
+    """Persist transcript evidence before settling the native ingress receipt."""
+    payload = {"client_ingress_id": ingress_id}
+    if origin is not None:
+        payload["origin"] = origin
+    if queued_turn_id is None:
+        store.append_message(session_id, role="user", text=message, turn_id=turn_id, origin=origin)
+    else:
+        payload["queued_turn_id"] = queued_turn_id
+    store.append_event(session_id, turn_id, kind="turn.steered", payload=payload)
+    store.update_ingress_receipt(
+        session_id, ingress_id, status="delivered", active_turn_id=turn_id, error_code=None,
+    )
+
+
 def _enqueue_bound(controller: Any, session: dict[str, Any], client_turn_id: str, message: str,
                    attachments: list[dict[str, Any]] | None, origin: str) -> tuple[dict[str, Any], bool]:
     contexts = controller.project_contexts
diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py
index eb4b9d69a0..799ebd17cb 100644
--- a/loopx/chat_runtime.py
+++ b/loopx/chat_runtime.py
@@ -1325,6 +1325,7 @@ def plan(phase):
     def steer_active_turn(
         self, *, session_id: str, client_ingress_id: str, message: str,
         expected_turn_id: str | None = None, _queued_turn_id: str | None = None,
+        origin: str | None = None,
     ) -> tuple[dict[str, Any], bool]:
         """Steer the exact active Codex Turn with durable ingress deduplication."""
         session = self._require_steering_session(session_id)
@@ -1416,21 +1417,10 @@ def steer_active_turn(
                 error_code="live_steering_rejected",
             )
             raise
-        if _queued_turn_id is None:
-            self.store.append_message(session_id, role="user", text=message, turn_id=active_turn_id)
-        self.store.append_event(
-            session_id,
-            active_turn_id,
-            kind="turn.steered",
-            payload={"client_ingress_id": client_ingress_id,
-                **({"queued_turn_id": _queued_turn_id} if _queued_turn_id is not None else {})},
-        )
-        self.store.update_ingress_receipt(
-            session_id,
-            client_ingress_id,
-            status="delivered",
-            active_turn_id=active_turn_id,
-            error_code=None,
+        from .capabilities.native_chat.queued_admission import record_steering_delivery
+        record_steering_delivery(
+            self.store, session_id=session_id, turn_id=active_turn_id, ingress_id=client_ingress_id,
+            message=message, queued_turn_id=_queued_turn_id, origin=origin,
         )
         turn = self.store.load_turn(session_id, active_turn_id)
         if turn is None:
diff --git a/loopx/control_plane/chat_turn_acceptance.py b/loopx/control_plane/chat_turn_acceptance.py
index de7a9fab8f..000839ec4c 100644
--- a/loopx/control_plane/chat_turn_acceptance.py
+++ b/loopx/control_plane/chat_turn_acceptance.py
@@ -184,6 +184,11 @@ def queued_chat_turn_ids(
     })["turn_ids"]
 
 
+def plan_native_mention_ingress(**facts: object) -> dict[str, Any]:
+    """Ask the shared queue/steering owner to select one exact ingress."""
+    return effect_runtime_result("chat.queue.mention", facts)
+
+
 def plan_queued_chat_steering(
     session: dict[str, Any], turn: dict[str, Any], active: dict[str, Any] | None,
     receipt: dict[str, Any] | None, *, ingress_id: str, operation_id: str, target: str, phase: str,
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index af454f6393..e6cadaf4ec 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -623,6 +623,7 @@ export function createEffectRuntimeHandlers(
     ["collaboration.conversation.reply_context", lazyHandler(() => import("./collaboration/conversation_reply_context.ts"), ({projectConversationReplyContext}) => projectConversationReplyContext)],
     ["chat.turn.accept", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({planChatTurnAcceptance}) => planChatTurnAcceptance)],
     ["chat.queue.select", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({selectQueuedChatTurns}) => selectQueuedChatTurns)],
+    ["chat.queue.mention", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({planNativeMentionIngress}) => planNativeMentionIngress)],
     ["chat.queue.steer", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({planQueuedChatSteering}) => planQueuedChatSteering)],
     ["chat.queue.mutate", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({planQueuedChatMutation}) => planQueuedChatMutation)],
     ["chat.turn.execution_allowed", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({mayContinueChatTurn}) => mayContinueChatTurn)],
diff --git a/loopx/control_plane/turn_driver/chat_queue_steering.ts b/loopx/control_plane/turn_driver/chat_queue_steering.ts
index f42fddb216..2e429cc8ae 100644
--- a/loopx/control_plane/turn_driver/chat_queue_steering.ts
+++ b/loopx/control_plane/turn_driver/chat_queue_steering.ts
@@ -54,6 +54,37 @@ export function selectQueuedChatTurns(input: JsonObject): JsonObject {
   return {turn_ids: ids};
 }
 
+/** Select delivery for a provider-verified native mention in an exact Session.
+ * A recorded ingress owns retries, even after the observed Turn has changed.
+ */
+export function planNativeMentionIngress(input: JsonObject): JsonObject {
+  if (input.mentioned !== true) return {action: "configured"};
+  if (input.binding_matches !== true) return {action: "reject", error_code: "live_steering_session_not_attached"};
+  if (input.text_only !== true) return {action: "reject", error_code: "queue_steering_attachments_unavailable"};
+
+  if (input.receipt != null) {
+    const receipt = requireJsonObject(input.receipt, "mention ingress receipt");
+    if (input.receipt_message_matches !== true) return {action: "unresolved"};
+    if (receipt.mode === "session_queue") return {action: "enqueue"};
+    if (receipt.mode !== "live_steering") return {action: "unresolved"};
+    if (receipt.status === "delivered" && typeof receipt.active_turn_id === "string" &&
+        receipt.active_turn_id && (receipt.expected_turn_id == null || receipt.expected_turn_id === receipt.active_turn_id)) {
+      return {action: "delivered", target_turn_id: receipt.active_turn_id};
+    }
+    if (receipt.status === "failed" && NOT_DELIVERED.has(String(receipt.error_code))) {
+      return {action: "reject", error_code: receipt.error_code};
+    }
+    return {action: "unresolved"};
+  }
+
+  if (input.queued_turn_exists === true) return {action: "enqueue"};
+  // A recovered source event without a Chat receipt cannot bind to a newer
+  // active Turn. Keep the original message available for explicit recovery.
+  if (input.may_deliver_new !== true) return {action: "unresolved"};
+  if (input.target_turn_id === null) return {action: "enqueue"};
+  return {action: "steer", target_turn_id: requireNonEmptyString(input.target_turn_id, "target_turn_id")};
+}
+
 export function planQueuedChatSteering(input: JsonObject): QueueSteeringPlan {
   const session = requireJsonObject(input.session, "session");
   const turn = requireJsonObject(input.turn, "queued turn");
diff --git a/loopx/extensions/lark/goal_topic_dispatch.py b/loopx/extensions/lark/goal_topic_dispatch.py
index b0dc46f443..d7a04631c9 100644
--- a/loopx/extensions/lark/goal_topic_dispatch.py
+++ b/loopx/extensions/lark/goal_topic_dispatch.py
@@ -21,6 +21,7 @@ def __init__(
         *,
         workers: int = 4,
         capacity: int = 64,
+        reserve_priority: bool = False,
     ) -> None:
         if workers < 1 or capacity < workers:
             raise ValueError("dispatch capacity must cover its positive worker limit")
@@ -28,8 +29,9 @@ def __init__(
         self._stop = stop
         self._workers = workers
         self._capacity = capacity
+        self._reserve_priority = reserve_priority and workers > 1
         self._condition = threading.Condition()
-        self._waiting: dict[str, deque[tuple[tuple[str, ...], str]]] = {}
+        self._waiting: dict[str, deque[tuple[tuple[str, ...], str, bool]]] = {}
         self._ready: deque[str] = deque()
         self._active: dict[str, tuple[str, ...]] = {}
         self._pending = 0
@@ -47,7 +49,7 @@ def failed(self) -> bool:
         with self._condition:
             return self._failure is not None
 
-    def submit(self, lane: tuple[str, ...], line: str) -> bool:
+    def submit(self, lane: tuple[str, ...], line: str, *, priority: bool = False) -> bool:
         with self._condition:
             while self._pending >= self._capacity and not self._closing:
                 if self._stop.is_set():
@@ -59,7 +61,7 @@ def submit(self, lane: tuple[str, ...], line: str) -> bool:
                 return False
             source = lane[0]
             queue = self._waiting.setdefault(source, deque())
-            queue.append((lane, line))
+            queue.append((lane, line, priority))
             self._pending += 1
             if source not in self._active and source not in self._ready:
                 self._ready.append(source)
@@ -75,7 +77,12 @@ def _start_ready(self) -> None:
             if self._cancelled or len(self._active) >= self._workers:
                 break
             source = self._ready.popleft()
-            lane, line = self._waiting[source][0]
+            lane, line, priority = self._waiting[source][0]
+            # Keep one of the existing workers available for a verified live
+            # mention while ordinary answer handlers wait on Chat completion.
+            if self._reserve_priority and not priority and len(self._active) >= self._workers - 1:
+                self._ready.append(source)
+                continue
             occupied = {key for active in self._active.values() for key in active}
             if occupied.intersection(lane):
                 self._ready.append(source)
diff --git a/loopx/extensions/lark/goal_topic_runtime.py b/loopx/extensions/lark/goal_topic_runtime.py
index 63b9e1c858..b03cbc10f9 100644
--- a/loopx/extensions/lark/goal_topic_runtime.py
+++ b/loopx/extensions/lark/goal_topic_runtime.py
@@ -27,6 +27,7 @@
     ManagerAuthorityMode,
 )
 from ..external_connector_runtime import (
+    ExternalEffectKind,
     ExternalResponsePolicy,
     build_external_event_response_receipt,
     decide_external_event_ack,
@@ -44,7 +45,7 @@
     normalize_lark_topic_event_rejection_reason,
 )
 from .goal_channel_targets import goal_channel_target_for_name
-from .goal_channel_transport import record_event_timing
+from .goal_channel_transport import lark_provider_mention_identities, record_event_timing
 from .goal_topic_connections import decide_lark_topic_event
 from .goal_topic_dispatch import ProfileEventDispatch
 from .inbox_reply import CommandRunner, reply_lark_event_inbox
@@ -399,8 +400,9 @@ def _profile_event_lane(
     snapshot: Mapping[str, Any],
     event: Mapping[str, Any],
     runtime_root: str | Path,
-) -> tuple[str, ...]:
-    """Only serialize transport work; actual ingress re-reads its authority."""
+    capture_mention: Callable[[Mapping[str, object]], dict[str, object]] | None,
+) -> tuple[tuple[str, ...], dict[str, object] | None]:
+    """Capture a live target before joining the ordinary reply lane."""
 
     private_config = dict(snapshot.get("private_profiles") or {}).get(profile)
     if isinstance(private_config, Mapping) and _native_profile_event(snapshot, profile, event):
@@ -417,7 +419,7 @@ def _profile_event_lane(
             # Use the same App/chat/topic (or private sender) source shape for
             # scheduling, never a grant or a persisted Session selection. Core
             # still verifies the provider and acquires its fresh source fence.
-            return ("private." + source["source_ref"],)
+            return ("private." + source["source_ref"],), None
 
     scoped, _status = _profile_event_route(
         profile=profile, snapshot=snapshot, event=event
@@ -440,10 +442,16 @@ def _profile_event_lane(
                     route.get("app_ref"), route.get("target_ref"),
                     route.get("topic_root_message_id"),
                 )
+            capture = (
+                capture_mention(route) if capture_mention is not None
+                and _route_native_mention(routed_event, route, targets) else None
+            )
+            if capture and capture.get("target_turn_id"):
+                return ("steer." + source, "steer.session." + str(route["session_id"])), capture
             if route.get("session_id"):
-                return (source, "session." + str(route["session_id"]))
-            return (source,)
-    return ("unmatched." + _opaque_digest(profile, event.get("chat_id")),)
+                return (source, "session." + str(route["session_id"])), capture
+            return (source,), capture
+    return ("unmatched." + _opaque_digest(profile, event.get("chat_id")),), None
 
 
 def poll_lark_goal_topic_profile_once(
@@ -457,6 +465,7 @@ def poll_lark_goal_topic_profile_once(
     reply_runner: CommandRunner = _default_simple_runner,
     proposal_deliverer: ProposalDeliverer | None = None,
     private_admitter: Callable[[str, dict[str, object]], Mapping[str, object]] | None = None,
+    mention_capture: Mapping[str, object] | None = None,
 ) -> dict[str, Any]:
     """Consume one bounded event batch for an App and reuse Inbox reply/ACK."""
 
@@ -527,6 +536,7 @@ def poll_lark_goal_topic_profile_once(
                 reply_runner=reply_runner,
                 provider_runner=provider_runner,
                 proposal_deliverer=proposal_deliverer,
+                mention_capture=mention_capture,
             )
         except Exception:
             event_statuses.append("processing_failed")
@@ -558,6 +568,7 @@ def stream_lark_goal_topic_profile(
     health_sink: HealthSink | None = None,
     proposal_deliverer: ProposalDeliverer | None = None,
     private_admitter: Callable[[str, dict[str, object]], Mapping[str, object]] | None = None,
+    capture_mention: Callable[[Mapping[str, object]], dict[str, object]] | None = None,
     review_callback_handler: ReviewCallbackHandler | None = None,
 ) -> dict[str, Any]:
     """Keep one bounded long-lived CLI consumer attached between messages."""
@@ -669,7 +680,8 @@ def stop_consumer() -> None:
     def handle_event(line: str) -> None:
         nonlocal event_count, replied_count, provider_ready
 
-        event = _event_payloads(line)[0]
+        envelope = json.loads(line)
+        event = envelope["event"]
         message_id = str(event.get("message_id") or "")
         record_event_timing(profile, message_id, "dispatch_started")
         current_snapshot = snapshot_provider()
@@ -679,15 +691,16 @@ def handle_event(line: str) -> None:
             snapshot=current_snapshot,
             runtime_root=runtime_root,
             answer=answer,
-            consume_runner=lambda _args, payload=line: {
+            consume_runner=lambda _args: {
                 "returncode": 0,
-                "stdout": payload,
+                "stdout": json.dumps(event, ensure_ascii=False),
                 "stderr": "",
             },
             provider_runner=provider_runner,
             reply_runner=reply_runner,
             proposal_deliverer=proposal_deliverer,
             private_admitter=private_admitter,
+            mention_capture=envelope["mention_capture"],
         )
         record_event_timing(profile, message_id, "dispatch_returned")
         with result_lock:
@@ -733,7 +746,7 @@ def handle_event(line: str) -> None:
                     flush=True,
                 )
 
-    dispatch = ProfileEventDispatch(handle_event, stop)
+    dispatch = ProfileEventDispatch(handle_event, stop, reserve_priority=capture_mention is not None)
     watcher.start()
     stream_failed = True
     try:
@@ -766,14 +779,16 @@ def handle_event(line: str) -> None:
             for event in _event_payloads(line):
                 message_id = str(event.get("message_id") or "")
                 record_event_timing(profile, message_id, "received")
-                lane = _profile_event_lane(
+                lane, mention_capture = _profile_event_lane(
                     profile=profile,
                     snapshot=snapshot_provider(),
                     event=event,
                     runtime_root=runtime_root,
+                    capture_mention=capture_mention,
                 )
                 record_event_timing(profile, message_id, "lane_ready")
-                if not dispatch.submit(lane, json.dumps(event, ensure_ascii=False)):
+                envelope = json.dumps({"event": event, "mention_capture": mention_capture}, ensure_ascii=False)
+                if not dispatch.submit(lane, envelope, priority=bool((mention_capture or {}).get("target_turn_id"))):
                     break
         stream_failed = False
     finally:
@@ -835,6 +850,73 @@ def handle_event(line: str) -> None:
     }
 
 
+def _bound_topic_session(route: Mapping[str, object], runtime_controller):
+    session_id = str(route.get("session_id") or "")
+    if not session_id:
+        return None
+    session = runtime_controller.store.load_session(session_id)
+    manager = route.get("conversation_kind") == "manager"
+    agent_id = manager_turn_executor(runtime_controller) if manager else str(route.get("agent_id") or "codex")
+    goal_id = str(route.get("goal_id") or "")
+    channel_id = str(route.get("manager_channel_id") or "") if manager else f"goal.{goal_id}"
+    if (
+        session is None or session.get("status") == "closed"
+        or (not manager and session.get("goal_id") != goal_id)
+        or session.get("agent_id") != agent_id or session.get("channel_id") != channel_id
+    ):
+        return None
+    return session
+
+
+def capture_topic_mention(route: Mapping[str, object], runtime_controller) -> dict[str, object]:
+    """Read the bound Session/Turn without opening or resuming an executor."""
+    session = _bound_topic_session(route, runtime_controller)
+    return {
+        "connection_id": route.get("connection_id"),
+        "session_id": route.get("session_id"),
+        "binding_matches": session is not None,
+        "target_turn_id": (session or {}).get("active_turn_id") or None,
+    }
+
+
+def _native_topic_mention(event: Mapping[str, object], identity: Mapping[str, object]) -> bool:
+    if event.get("sender_type") != "user" or not event.get("sender_id") or event.get("historical_context_only"):
+        return False
+    expected = {str(identity.get(key) or "") for key in ("bot_app_id", "bot_open_id")} - {""}
+    mentions = event.get("mentions")
+    return bool(expected and isinstance(mentions, list) and any(
+        isinstance(item, Mapping) and lark_provider_mention_identities(item).intersection(expected)
+        for item in mentions
+    ))
+
+
+def _route_native_mention(event: Mapping[str, object], route: Mapping[str, object],
+                          target_payload: Mapping[str, object]) -> bool:
+    if route.get("ingress_mode") not in {"live_steering", "session_queue", "async_inbox"}:
+        return False
+    if (route.get("conversation_kind") == "manager"
+            and route.get("authority_mode") != ManagerAuthorityMode.TURN_AUTHORIZED.value):
+        return False
+    target = goal_channel_target_for_name(target_payload, str(route.get("target_ref") or "")) or {}
+    identity = target.get("identity")
+    return isinstance(identity, Mapping) and _native_topic_mention(event, identity)
+
+
+def _mention_reply(route: Mapping[str, object], decision: Mapping[str, object]) -> dict[str, object]:
+    receipt = _session_turn_effect(route)
+    if decision["action"] == "delivered":
+        return {"response_text": "已送达当前回合。", "effect_receipt": receipt}
+    receipt["effect_kind"] = ExternalEffectKind.NO_FOLLOW_UP.value
+    if decision["action"] == "unresolved":
+        return {"response_text": "送达状态待核实。原消息保留在 Lark。", "effect_receipt": receipt}
+    text = {
+        "queue_steering_attachments_unavailable": "本轮调整仅支持文字。原消息和附件保留在 Lark。",
+        "live_steering_turn_mismatch": "接收时的回合已结束或切换。原消息保留在 Lark。",
+        "live_steering_turn_not_started": "接收时的回合尚未就绪。原消息保留在 Lark。",
+    }.get(str(decision.get("error_code")), "当前会话无法调整本轮。原消息保留在 Lark。")
+    return {"response_text": text, "effect_receipt": receipt}
+
+
 def answer_lark_goal_topic(
     *,
     route: Mapping[str, Any],
@@ -878,17 +960,50 @@ def answer_lark_goal_topic(
         route.get("message_id"),
         route.get("topic_root_message_id"),
     )
+    native_mention = route.get("native_mention") is True
+    if native_mention:
+        from ...capabilities.native_chat.queued_admission import deliver_native_mention
+
+        capture = route.get("mention_capture")
+        capture = capture if isinstance(capture, Mapping) else capture_topic_mention(route, runtime_controller)
+        current = _bound_topic_session(route, runtime_controller)
+        binding_matches = (
+            current is not None and capture.get("binding_matches") is True
+            and capture.get("session_id") == session_id
+            and capture.get("connection_id") == route.get("connection_id")
+            and ingress_mode in {"live_steering", "session_queue"}
+        )
+        if manager and binding_matches and current is not None and capture.get("target_turn_id"):
+            from ...chat_manager_context import manager_authorization_scope_is_current
+
+            binding_matches = manager_authorization_scope_is_current(
+                runtime_controller.registry_path, runtime_controller.manager_scope_resolver, current,
+                current.get("manager_authorization_scope_id"),
+                runtime_root=runtime_controller.coordination_runtime_root,
+            )
+        body = text
+        try:
+            payload = json.loads(text)
+        except (TypeError, ValueError):
+            payload = None
+        if isinstance(payload, Mapping) and isinstance(payload.get("text"), str):
+            body = payload["text"]
+        decision = deliver_native_mention(
+            runtime_controller, session_id=session_id, ingress_id=client_turn_id,
+            target_turn_id=capture.get("target_turn_id"), binding_matches=binding_matches,
+            text_only=route.get("native_mention_text_only") is True,
+            may_deliver_new=route.get("native_mention_fresh") is True,
+            steering_text=(reply_context + "\n\n" if reply_context else "") + body,
+            queued_text=message, origin="lark",
+        )
+        if decision["action"] != "enqueue":
+            return _mention_reply(route, decision)
+        ingress_mode = "session_queue"
+
     if ingress_mode in {"live_steering", "session_queue"}:
-        session = runtime_controller.store.load_session(session_id)
-        if (
-            session is None
-            or (not manager and session.get("goal_id") != goal_id)
-            or session.get("agent_id") != agent_id
-            or session.get("channel_id") != expected_channel
-            or session.get("status") == "closed"
-        ):
+        if _bound_topic_session(route, runtime_controller) is None:
             if manager and manager_session_requires_executor_rebind(
-                session,
+                runtime_controller.store.load_session(session_id),
                 expected_channel=expected_channel,
                 agent_id=agent_id,
             ):
@@ -896,9 +1011,7 @@ def answer_lark_goal_topic(
                     "manager_channel_executor_rebind_required",
                     _session_turn_effect(route),
                 )
-            raise RuntimeError(
-                "bound Agent session is unavailable or no longer matches"
-            )
+            raise RuntimeError("bound Agent session is unavailable or no longer matches")
         if ingress_mode == "live_steering":
             turn, _created = runtime_controller.steer_active_turn(
                 session_id=session_id,
@@ -934,6 +1047,10 @@ def answer_lark_goal_topic(
                 objective=str(objective or goal_id),
                 origin="lark",
             )
+            if native_mention:
+                runtime_controller.store.update_ingress_receipt(
+                    session_id, client_turn_id, status="delivered",
+                )
     else:
         # Compatibility path for bindings created before the three ingress modes.
         channel_id = "lark." + _opaque_digest(
@@ -1055,8 +1172,9 @@ def process_lark_goal_topic_event(
     reply_runner: CommandRunner,
     provider_runner: Any | None = None,
     proposal_deliverer: ProposalDeliverer | None = None,
+    mention_capture: Mapping[str, object] | None = None,
 ) -> dict[str, Any]:
-    """Single-flight an addressed manager message through answer, reply and ACK.
+    """Single-flight a manager or live mention through ingress, reply and ACK.
 
     The inbox and delivery receipt make sequential retries safe, but two
     listeners can otherwise both observe an absent receipt and generate two
@@ -1071,10 +1189,20 @@ def process_lark_goal_topic_event(
         runtime_root=runtime_root,
     )
     route = decision.get("route")
-    if (
-        isinstance(route, Mapping)
-        and route.get("conversation_kind") == "manager"
-        and route.get("authority_mode") == ManagerAuthorityMode.TURN_AUTHORIZED.value
+    native_mention = isinstance(route, Mapping) and _route_native_mention(event, route, target_payload)
+    if mention_capture is not None and not native_mention:
+        return {"ok": True, "status": "ignored", "reason": "mention_binding_changed"}
+    if native_mention:
+        route = {
+            **route,
+            "native_mention": True,
+            "native_mention_text_only": event.get("message_type") == "text" and not event.get("attachment_count"),
+            "mention_capture": mention_capture,
+        }
+        decision = {**decision, "route": route}
+    if isinstance(route, Mapping) and (
+        native_mention or (route.get("conversation_kind") == "manager"
+                           and route.get("authority_mode") == ManagerAuthorityMode.TURN_AUTHORIZED.value)
     ):
         message_id = str(route.get("message_id") or "")
         if not MESSAGE_ID_PATTERN.fullmatch(message_id):
@@ -1252,6 +1380,7 @@ def _process_lark_goal_topic_event(
             "goal_id": route["goal_id"],
             "agent_id": route.get("agent_id"),
             "inbox_config_ref": config_ref,
+            **({"reason": "async_inbox_requires_bound_session"} if route.get("native_mention") else {}),
         }
     # Receipt ACK is visible while the synchronous manager is reasoning. The
     # private reaction ledger makes retries idempotent. Manager received ACKs
@@ -1295,6 +1424,11 @@ def _process_lark_goal_topic_event(
     captured_event = next(
         item for item in projection["items"] if item.get("message_id") == message_id
     )
+    if route.get("native_mention") and (
+        not captured_event.get("target_mention_count") or captured_event.get("historical_context_only")
+        or captured_event.get("sender_id") != canonical["sender_id"]
+    ):
+        return {"ok": False, "status": "native_mention_proof_changed", "goal_id": route["goal_id"]}
     route = {
         **route,
         "source_sender_id": str(canonical.get("sender_id") or ""),
@@ -1305,6 +1439,7 @@ def _process_lark_goal_topic_event(
         "root_id": captured_event.get("root_id", ""),
         "thread_id": captured_event.get("thread_id", ""),
         "thread_context": captured_event.get("thread_context"),
+        **({"native_mention_fresh": int(ingest.get("accepted_count") or 0) > 0} if route.get("native_mention") else {}),
         **({"context_materials": context_materials} if context_materials else {}),
     }
     delivery_path: Path | None = None
diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index 0cea1658f9..c8c1814273 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -306,6 +306,7 @@ def answer(
                         effective_route = route
                         if (
                             route.get("conversation_kind") == "manager"
+                            and route.get("native_mention") is not True
                             and self.manager_route_reconciler is not None
                         ):
                             try:
@@ -343,9 +344,10 @@ def answer(
                         else:
                             response_text = answer_result
                             proposal_ids = []
+                        effect_receipt = answer_result.get("effect_receipt") if isinstance(answer_result, Mapping) else None
                         return {
                             "response_text": response_text,
-                            "effect_receipt": session_turn_effect(effective_route),
+                            "effect_receipt": effect_receipt or session_turn_effect(effective_route),
                             "proposal_ids": proposal_ids,
                         }
 
@@ -415,6 +417,7 @@ def handle_review_callback(
                             stop=stop,
                             runtime_root=self.runtime_root,
                             answer=answer,
+                            capture_mention=lambda route: runtime.capture_topic_mention(route, self.runtime_controller),
                             private_admitter=(self._admit_private
                                 if self.private_conversations is not None else None),
                             proposal_deliverer=deliver_proposals,
diff --git a/tests/control_plane_ts/chat_queue_steering.test.ts b/tests/control_plane_ts/chat_queue_steering.test.ts
index 75d7ee327a..21e33f6393 100644
--- a/tests/control_plane_ts/chat_queue_steering.test.ts
+++ b/tests/control_plane_ts/chat_queue_steering.test.ts
@@ -1,6 +1,6 @@
 import assert from "node:assert/strict";
 import test from "node:test";
-import {planQueuedChatMutation, planQueuedChatSteering, selectQueuedChatTurns, QUEUE_STEERING_SCHEMA} from "../../loopx/control_plane/turn_driver/chat_queue_steering.ts";
+import {planNativeMentionIngress, planQueuedChatMutation, planQueuedChatSteering, selectQueuedChatTurns, QUEUE_STEERING_SCHEMA} from "../../loopx/control_plane/turn_driver/chat_queue_steering.ts";
 
 function facts() {
   return {phase: "prepare", session: {status: "busy", active_turn_id: "active"}, active_status: "running",
@@ -96,3 +96,27 @@ test("acknowledgement recovers after target completion while proven rejection re
   assert.equal(planQueuedChatSteering({...f, session: {status: "ready", active_turn_id: null}}).action, "release");
   assert.equal(planQueuedChatSteering({...f, phase: "settle"}).action, "release");
 });
+
+test("native mention admission preserves the observed target and prior ingress", () => {
+  const input = {mentioned: true, binding_matches: true, text_only: true, may_deliver_new: true,
+    target_turn_id: "first", receipt_message_matches: true};
+  assert.deepEqual(planNativeMentionIngress({...input, mentioned: false}), {action: "configured"});
+  assert.deepEqual(planNativeMentionIngress(input), {action: "steer", target_turn_id: "first"});
+  assert.deepEqual(planNativeMentionIngress({...input, target_turn_id: null}), {action: "enqueue"});
+  assert.deepEqual(planNativeMentionIngress({...input, may_deliver_new: false, target_turn_id: "later"}),
+    {action: "unresolved"});
+  assert.deepEqual(planNativeMentionIngress({...input, queued_turn_exists: true}), {action: "enqueue"});
+  assert.equal(planNativeMentionIngress({...input, binding_matches: false}).action, "reject");
+  assert.equal(planNativeMentionIngress({...input, text_only: false}).error_code, "queue_steering_attachments_unavailable");
+  const receipt = {mode: "live_steering", status: "delivered", expected_turn_id: "first", active_turn_id: "first"};
+  assert.deepEqual(planNativeMentionIngress({...input, receipt, may_deliver_new: false, target_turn_id: "later"}),
+    {action: "delivered", target_turn_id: "first"});
+  for (const uncertain of [{...receipt, status: "pending"}, {...receipt, active_turn_id: "later"},
+    {...receipt, status: "failed", error_code: "live_steering_rejected"}]) {
+    assert.deepEqual(planNativeMentionIngress({...input, receipt: uncertain}), {action: "unresolved"});
+  }
+  assert.deepEqual(planNativeMentionIngress({...input, receipt, receipt_message_matches: false}), {action: "unresolved"});
+  assert.equal(planNativeMentionIngress({...input, receipt: {...receipt, status: "failed",
+    error_code: "live_steering_turn_mismatch"}}).action, "reject");
+  assert.equal(planNativeMentionIngress({...input, receipt: {mode: "session_queue"}}).action, "enqueue");
+});
diff --git a/tests/extensions/test_lark_goal_topic_runtime.py b/tests/extensions/test_lark_goal_topic_runtime.py
index 667102e1b6..13dae803aa 100644
--- a/tests/extensions/test_lark_goal_topic_runtime.py
+++ b/tests/extensions/test_lark_goal_topic_runtime.py
@@ -3594,3 +3594,215 @@ def test_message_event_projection_preserves_flat_cli_and_raw_provider_provenance
         for key, value in expected.items():
             assert row[key] == value
         assert row["content"] == "plain rendered text"
+
+
+@pytest.mark.parametrize("manager", [False, True])
+def test_native_mention_reaches_the_running_topic_before_its_ordinary_reply(tmp_path: Path, manager: bool) -> None:
+    """Use the actual listener, Inbox, Chat store, and Codex stdio protocol."""
+    import runpy
+    import stat
+    import time
+
+    from loopx.chat_manager import manager_channel
+    from loopx.chat_runtime import ChatRuntimeController
+    from loopx.chat_store import ChatSessionStore
+    from loopx.extensions.lark.event_inbox import ingest_lark_event_inbox
+    from loopx.extensions.lark.goal_topic_runtime import (
+        answer_lark_goal_topic,
+        capture_topic_mention,
+        process_lark_goal_topic_event,
+        stream_lark_goal_topic_profile,
+    )
+    from loopx.extensions.lark.manager_context import opaque_digest, session_turn_effect
+    from loopx.extensions.lark.manager_routing import authorized_manager_goal_ids
+
+    fixture = runpy.run_path(str(Path(__file__).parents[2] / "examples/loopx-chat-runtime-smoke.py"))
+    release = tmp_path / "release"
+    executable = tmp_path / "codex"
+    fake = fixture["FAKE_CODEX"].replace("import json\n", "import json\nfrom pathlib import Path\n", 1)
+    fake = fake.replace(
+        "        response = 'Visible steered response.",
+        f"        until = time.monotonic() + 10\n"
+        f"        while not Path({str(release)!r}).exists() and time.monotonic() < until:\n"
+        "            time.sleep(0.01)\n"
+        "        response = 'Visible steered response.",
+        1,
+    )
+    executable.write_text(fake, encoding="utf-8")
+    executable.chmod(executable.stat().st_mode | stat.S_IXUSR)
+    registry = {"goals": [{"id": "goal-alpha", "repo": str(tmp_path),
+                           "coordination": {"registered_agents": ["codex"]}}]}
+    registry_path = tmp_path / "registry.json"
+    registry_path.write_text(json.dumps(registry), encoding="utf-8")
+    store = ChatSessionStore(tmp_path / "chat")
+    runtime_root = tmp_path / "runtime"
+    snapshot: dict[str, Any] = {}
+    controller = ChatRuntimeController(
+        store=store, codex_bin=str(executable), registry_path=registry_path if manager else None,
+        manager_scope_resolver=lambda session: authorized_manager_goal_ids(
+            snapshot, session, runtime_root=tmp_path, registry_path=registry_path,
+            executor_runtime_root=runtime_root,
+        ),
+    )
+    session, _ = controller.open_session(
+        goal_id="goal-alpha", agent_id="codex", work_dir=tmp_path,
+        objective="Review delivery", mode="resume_latest",
+        channel_id=manager_channel(provider="lark", audience="mew\0oc_public_fixture") if manager else "goal.goal-alpha",
+    )
+    session_id = session["session_id"]
+    target_path, binding_path = tmp_path / "targets.json", tmp_path / "binding.json"
+    state: dict[str, Any] = {}
+    connected = connect_lark_goal_topic(
+        registry=registry,
+        goal_id="goal-alpha", agent_id="codex", session_id=session_id, target_path=target_path,
+        binding_path=binding_path, app_ref="mew", chat_id="oc_public_fixture", chat_name="Product group",
+        capture_scope="configured_chat_all", ingress_mode="session_queue",
+        conversation_kind="manager" if manager else "goal",
+        turn_trigger="human_messages" if manager else None,
+        runner=_connection_runner(state), cli_bin="fake-lark",
+    )
+    assert connected["ok"], connected
+    target_payload = read_goal_channel_targets(target_path)
+    binding_payload = read_goal_channel_binding(binding_path)
+    topic = binding_for_goal(binding_payload, "goal-alpha")["topic"]["root_message_id"]
+    snapshot.update(target_payload=target_payload, binding_payloads={"goal-alpha": binding_payload})
+    ordinary = {"event_id": "evt_native_initial", "message_id": "om_native_initial",
+                "chat_id": "oc_public_fixture", "root_id": topic, "content": "wait for steer",
+                "sender_id": "ou_fixture_sender", "sender_type": "user", "message_type": "text", "mentions": []}
+    mention_text = "Add the error case.\nKeep both outcomes."
+    mention = {**ordinary, "event_id": "evt_native_mention", "message_id": "om_native_mention",
+               "content": json.dumps({"text": mention_text}),
+               "mentions": [{"name": "Current bot", "id": "cli_public_fixture"}]}
+    attachment = {**mention, "event_id": "evt_native_file", "message_id": "om_native_file",
+                  "message_type": "file", "content": json.dumps({"file_key": "file_fixture"})}
+    wrong_bot = {**mention, "event_id": "evt_other_bot", "message_id": "om_other_bot",
+                 "content": "ordinary input for this connection",
+                 "mentions": [{"name": "Current bot", "id": "cli_other_bot"}]}
+    idle = {**mention, "event_id": "evt_native_idle", "message_id": "om_native_idle",
+            "content": json.dumps({"text": "Start normal input."})}
+    steered_reply = threading.Event()
+    attachment_reply = threading.Event()
+    runner = _reply_runner(state)
+
+    def reply(args):
+        if args[3:6] == ["im", "reactions", "create"]:
+            return {"returncode": 0, "stdout": json.dumps({"ok": True, "data": {"reaction_id": "reaction_fixture"}})}
+        if args[3:6] == ["im", "reactions", "delete"]:
+            return {"returncode": 0, "stdout": json.dumps({"ok": True})}
+        result = runner(args)
+        if "+messages-mget" in args:
+            if state.get("reply_text") == "已送达当前回合。":
+                steered_reply.set()
+            if "本轮调整仅支持文字" in state.get("reply_text", ""):
+                attachment_reply.set()
+        return result
+
+    def answer(route, text):
+        result = answer_lark_goal_topic(
+            route=route, text=text, work_dir=tmp_path, objective="Review delivery", runtime_controller=controller,
+        )
+        return result if isinstance(result, Mapping) else {
+            "response_text": result, "effect_receipt": session_turn_effect(route),
+        }
+
+    initial_id = "lark." + opaque_digest(ordinary["message_id"], topic)
+    mention_id = "lark." + opaque_digest(mention["message_id"], topic)
+    wrong_id = "lark." + opaque_digest(wrong_bot["message_id"], topic)
+    idle_id = "lark." + opaque_digest(idle["message_id"], topic)
+
+    def lines():
+        yield json.dumps(ordinary) + "\n"
+        deadline = time.monotonic() + 8
+        while time.monotonic() < deadline:
+            initial = store.turn_for_client(session_id, initial_id)
+            if initial and initial.get("upstream_turn_id"):
+                break
+            time.sleep(0.02)
+        else:
+            raise AssertionError("the initial Topic input never reached the native executor")
+        try:
+            yield json.dumps(wrong_bot) + "\n"
+            yield json.dumps(attachment) + "\n"
+            assert attachment_reply.wait(5), "the original file needs an explicit retained-message reply"
+            yield json.dumps(mention) + "\n"
+            yield json.dumps(mention) + "\n"
+            assert steered_reply.wait(5), "the listener waited for the ordinary answer before handling the mention"
+            assert store.load_turn(session_id, initial["turn_id"])["status"] == "running"
+            assert store.steering_receipt(session_id, wrong_id) is None
+            assert store.turn_for_client(session_id, wrong_id) is None
+        finally:
+            release.touch()
+        deadline = time.monotonic() + 8
+        while time.monotonic() < deadline:
+            queued = store.turn_for_client(session_id, wrong_id)
+            if queued and queued["status"] == "completed" and not store.load_session(session_id).get("active_turn_id"):
+                break
+            time.sleep(0.02)
+        else:
+            raise AssertionError("the ordinary input did not finish after the active Turn")
+        yield json.dumps(idle) + "\n"
+        yield json.dumps(idle) + "\n"
+        yield "[event] exited — received 7 event(s) in 1s (reason: limit)\n"
+
+    class FinishedConsumer:
+        stdout = lines()
+
+        def poll(self):
+            return 0
+
+        def wait(self, timeout=None):
+            return 0
+
+    try:
+        result = stream_lark_goal_topic_profile(
+            profile="mew", snapshot_provider=lambda: snapshot, stop=threading.Event(), runtime_root=runtime_root,
+            answer=answer, process_factory=lambda _: FinishedConsumer(), reply_runner=reply,
+            provider_runner=lambda args, **kwargs: subprocess.CompletedProcess(
+                args, 0, json.dumps({"ok": True, "data": {"items": [], "has_more": False}}), "",
+            ),
+            capture_mention=lambda route: capture_topic_mention(route, controller),
+        )
+        assert result["ok"] and result["event_count"] == 7 and result["replied_count"] == 5, result
+        initial = store.turn_for_client(session_id, initial_id)
+        assert store.turn_for_client(session_id, mention_id) is None
+        assert store.turn_for_client(session_id, idle_id)["status"] == "completed"
+        assert store.steering_receipt(session_id, idle_id)["mode"] == "session_queue"
+        attachment_id = "lark." + opaque_digest(attachment["message_id"], topic)
+        assert store.turn_for_client(session_id, attachment_id) is None
+        receipt = store.steering_receipt(session_id, mention_id)
+        assert receipt["expected_turn_id"] == receipt["active_turn_id"] == initial["turn_id"]
+        assert receipt["status"] == "delivered"
+        user = [msg for msg in store.messages(session_id) if msg["role"] == "user"]
+        assert [msg["text"] for msg in user if msg["text"] == mention_text] == [mention_text]
+        steered = next(msg for msg in user if msg["text"] == mention_text)
+        assert steered["origin"] == "lark" and steered["turn_id"] == initial["turn_id"]
+        assert len([e for e in store.events_after(session_id, initial["turn_id"], None)
+                    if e["kind"] == "turn.steered"]) == 1
+        late = {**mention, "event_id": "evt_native_late", "message_id": "om_native_late"}
+        ended = process_lark_goal_topic_event(
+            **snapshot, event=late, runtime_root=runtime_root, answer=answer, reply_runner=reply,
+            mention_capture={
+                "connection_id": binding_for_goal(binding_payload, "goal-alpha")["connection_id"],
+                "session_id": session_id, "binding_matches": True, "target_turn_id": initial["turn_id"],
+            },
+        )
+        assert ended["ok"] and "已结束或切换" in state["reply_text"], ended
+        late_id = "lark." + opaque_digest(late["message_id"], topic)
+        assert store.turn_for_client(session_id, late_id) is None
+        assert store.steering_receipt(session_id, late_id)["error_code"] == "live_steering_turn_mismatch"
+        interrupted = {**mention, "event_id": "evt_native_interrupted", "message_id": "om_native_interrupted"}
+        stored = ingest_lark_event_inbox(
+            project=runtime_root, config_path=ended["inbox_config_ref"], execute=True,
+            events=[{**interrupted, "schema_version": "lark_event_inbox_event_v0"}],
+        )
+        assert stored["accepted_count"] == 1
+        recovered = process_lark_goal_topic_event(
+            **snapshot, event=interrupted, runtime_root=runtime_root, answer=answer, reply_runner=reply,
+        )
+        assert recovered["ok"] and "送达状态待核实" in state["reply_text"], recovered
+        interrupted_id = "lark." + opaque_digest(interrupted["message_id"], topic)
+        assert store.turn_for_client(session_id, interrupted_id) is None
+        assert store.steering_receipt(session_id, interrupted_id) is None
+    finally:
+        release.touch()
+        controller.close_session(session_id)