loopx-agent · Nov 7, 2026 · Nov 8, 2026 · Nov 8, 2026 · Nov 8, 2026 · Nov 8, 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

@@ -1374,6 +1374,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)
@@ -1465,21 +1466,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,

@@ -628,6 +628,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");

@@ -1,14 +1,16 @@
 from __future__ import annotations
 
 import hashlib
+import heapq
 import json
 import os
 import re
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterator, Mapping, Sequence
 from contextlib import nullcontext
 from datetime import UTC, datetime
 from pathlib import Path, PurePosixPath
-from typing import Any
+from time import time_ns
+from typing import Any, NotRequired, TypedDict
 
 from ...control_plane.work_items.operator_inbox import (
     CAPTURE_SCOPES,
@@ -597,6 +599,152 @@ def lark_event_mentions_bot(
     )
 
 
+
+class LarkTopicInput(TypedDict):
+    order: int
+    provider_ref: str
+    route_ref: str
+    lane: tuple[str, ...]
+    event: dict[str, object]
+    mention_capture: dict[str, object] | None
+    live_turn_capture: NotRequired[dict[str, object]]
+    started: bool
+
+
+class LarkTopicInboxIntake:
+    """Private live input in the same Topic Inbox, before reply dispatch.
+
+    This keeps the exact provider envelope separate from the compact context
+    projection. The existing App listener lease owns capture, and its existing
+    source lane owns dispatch. An ambiguous attempt stays retained for readback.
+    """
+
+    def __init__(self, *, project: Path, config_path: Path, provider_ref: str) -> None:
+        config = load_lark_event_inbox_config(project=project, config_path=config_path)
+        if not config["enabled"]:
+            raise ValueError("Lark Topic intake requires an enabled Inbox")
+        self._inbox: Path = config["inbox_path"]
+        self._provider_ref = provider_ref
+        reply = config["reply"]
+        configured_provider = str(reply["bot_app_id"] or reply["sender_profile"])
+        configured_ref = hashlib.sha256(configured_provider.encode("utf-8")).hexdigest()[:32]
+        self._source_matches = configured_ref == provider_ref
+        self._directory = self._inbox / ".state" / "topic-intake"
+        self._last_order = 0
+        if self._source_matches:
+            self._last_order = max(
+                (packet["order"] for _, packet in self._inputs()), default=0
+            )
+
+    @property
+    def has_retained_input(self) -> bool:
+        return self._last_order > 0
+
+    @staticmethod
+    def _read(path: Path) -> LarkTopicInput | None:
+        try:
+            row = json.loads(path.read_text(encoding="utf-8"))
+        except FileNotFoundError:
+            return None
+        if not isinstance(row, dict):
+            raise ValueError("invalid Lark Topic intake record")
+        order = row.get("order")
+        event = row.get("event")
+        provider_ref = row.get("provider_ref")
+        route_ref = row.get("route_ref")
+        lane = row.get("lane")
+        mention = row.get("mention_capture")
+        live_turn = row.get("live_turn_capture")
+        started = row.get("started")
+        if (
+            type(order) is not int or order <= 0
+            or not isinstance(event, dict) or event.get("message_id") != path.stem
+            or not isinstance(provider_ref, str) or not re.fullmatch(r"[a-f0-9]{32}", provider_ref)
+            or not isinstance(route_ref, str) or not re.fullmatch(r"[a-f0-9]{32}", route_ref)
+            or not isinstance(lane, list) or not 1 <= len(lane) <= 2
+            or not all(isinstance(key, str) and 0 < len(key) <= 256 for key in lane)
+            or (mention is not None and not isinstance(mention, dict))
+            or (live_turn is not None and not isinstance(live_turn, dict))
+            or type(started) is not bool
+        ):
+            raise ValueError("invalid Lark Topic intake record")
+        packet: LarkTopicInput = {
+            "order": order, "event": event, "provider_ref": provider_ref,
+            "route_ref": route_ref, "lane": tuple(lane),
+            "mention_capture": mention, "started": started,
+        }
+        if live_turn is not None:
+            packet["live_turn_capture"] = live_turn
+        return packet
+
+    def _inputs(self) -> Iterator[tuple[Path, LarkTopicInput]]:
+        for path in self._directory.glob("*.json"):
+            if not MESSAGE_ID_PATTERN.fullmatch(path.stem):
+                continue
+            packet = self._read(path)
+            if packet is not None and packet["provider_ref"] == self._provider_ref:
+                yield path, packet
+
+    def retains(self, message_id: str) -> bool:
+        if not MESSAGE_ID_PATTERN.fullmatch(message_id):
+            return False
+        packet = self._read(self._directory / f"{message_id}.json")
+        return packet is not None and packet["provider_ref"] == self._provider_ref
+
+    def capture(
+        self, *, event: Mapping[str, object], route_ref: str, lane: tuple[str, ...],
+        mention_capture: dict[str, object] | None,
+        live_turn_capture: dict[str, object] | None = None,
+    ) -> bool:
+        if not self._source_matches:
+            raise ValueError("Lark Topic intake belongs to a different provider")
+        message_id = str(event.get("message_id") or "")
+        if not MESSAGE_ID_PATTERN.fullmatch(message_id):
+            raise ValueError("invalid Lark Topic input message id")
+        path = self._directory / f"{message_id}.json"
+        if path.is_file():
+            # Provider redelivery cannot overwrite the original body, route or
+            # pending/ambiguous attempt. The caller still wakes the same lane.
+            return False
+        self._last_order = max(self._last_order + 1, time_ns())
+        packet: LarkTopicInput = {
+            "order": self._last_order, "provider_ref": self._provider_ref,
+            "route_ref": route_ref, "lane": lane, "event": dict(event),
+            "mention_capture": mention_capture, "started": False,
+        }
+        if live_turn_capture is not None:
+            packet["live_turn_capture"] = live_turn_capture
+        self._directory.mkdir(parents=True, exist_ok=True, mode=0o700)
+        write_private_json_atomic(path, packet)
+        return True
+
+    def pending(
+        self, lane: tuple[str, ...] | None = None, *, limit: int | None = None,
+    ) -> list[Path]:
+        # Keep only references in memory while a reply waits on a Turn.
+        pending = (
+            (packet["order"], path)
+            for path, packet in self._inputs()
+            if not packet["started"] and (lane is None or packet["lane"] == lane)
+        )
+        ordered = sorted(pending) if limit is None else heapq.nsmallest(limit, pending)
+        return [path for _, path in ordered]
+
+    def read(self, path: Path) -> LarkTopicInput | None:
+        packet = self._read(path)
+        return packet if packet is not None and packet["provider_ref"] == self._provider_ref else None
+
+    def begin(self, path: Path, packet: LarkTopicInput) -> None:
+        # Persist the attempt fence before any Chat/model/provider effect.
+        write_private_json_atomic(path, {**packet, "started": True})
+
+    def finish(self, path: Path) -> None:
+        # The existing processed receipt is authoritative even after cleanup.
+        # If reply/ACK was ambiguous, retain the exact source instead of replay.
+        if path.stem in _load_processed(self._inbox / "processed.json"):
+            path.unlink(missing_ok=True)
+
+
 def ingest_lark_event_inbox(
     *,
     project: str | Path,