loopx-agent · Nov 7, 2026 · loopx-agent · Nov 7, 2026
@@ -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": "查看飞书事件配置",

@@ -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

@@ -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:

@@ -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,

@@ -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)],

@@ -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");

@@ -21,15 +21,17 @@ 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")
         self._handle = handle
         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)