fix(lark): keep admissions moving during background stop

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -118,6 +118,12 @@ later answer. Core freezes the original Session and canonical candidate Turn
 identities in the existing request journal **before** native inventory or
 transport resume can fail. Native observation may only narrow that frozen set;
 a retry cannot add later completed/active Turns or switch to a newer Session.
+The existing Lark listener returns after this durable freeze; its existing reply
+workers recover the prepared request to observe and cancel native background
+commands. Slow or unavailable native control therefore does not hold inbound
+admission. This is scheduling only: it introduces no queue or completion state,
+does not defer an active Turn's exact interrupt, and sends no successful stop
+reply until native control and the original delivery checks have succeeded.
 A recovered legacy request without a recorded target has unknown original scope;
 Core rejects retargeting and preserves it unconfirmed. A fresh stop is a separate
 user request, never an automatic replacement for the unknown one.

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -81,6 +81,14 @@ listener、队列或模型 runner。一个 App 仍只有一个 binding owner，
 编辑和回执恢复复用现有投递路径，使用 `source_thread` 落点。解绑或原消息不可读后，
 即使执行完成，也不得继续发送结果。
 
+普通 Codex 话题空闲时，`/stop` 也覆盖同一原生 Session 中已完成 Turn 遗留的后台命令。
+Core 先在原请求日志固定 Session 与候选 Turn，原 listener 随即继续受理消息；现有
+回复工作池再恢复这条 prepared 请求，读取原生命令清单并取消原目标。恢复只能缩小
+固定候选，不能追停后来的 Turn 或新 Session。慢速或失败的原生控制不阻塞入站，
+也不提前回复“已停止”；没有固定目标的旧请求保持未知，活跃 Turn 仍沿原确切中断路径。
+这只调整现有 IO 调度，不新增队列、消费者或完成状态；实际安装、停止时延与完整社区
+准出仍需分别验收。
+
 群上下文强制使用 `workspace_only`，不受本人私聊 `host_default` 影响。只保留选定
 工作区的读写 grant，不提供个人 portfolio、直连 Agent 选择、全局 skills、MCP、
 shell profile 或继承的账号环境。原生宿主必须确认确切 permissions profile 和工作区

loopx/capabilities/native_chat/external_conversations.py

@@ -91,7 +91,7 @@ def _automation_cadence_observation(
     def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str,
               message: str, command: str | None = None,
               attachments: list[dict[str, Any]] | None = None,
-              origin: str = "lark") -> dict[str, Any]:
+              origin: str = "lark", defer_background_stop: bool = False) -> dict[str, Any]:
         import re
         if origin not in {"lark", "external"}:
             raise ValueError("unsupported external conversation origin")
@@ -106,36 +106,52 @@ def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str,
             if not isinstance(ref, str) or not re.fullmatch(r"[a-f0-9]{24}", ref):
                 raise ValueError("invalid external conversation source reference")
         path = self.root / f"{request_ref}.json"
-        with exclusive_file_lock(self.root / "source-fences" / f"{binding_id}.{source['source_ref']}.json", operation="route_external_chat_request"), exclusive_file_lock(path, operation="admit_external_chat_request"):
-            # Resolve after waiting for both fences, including exact replay.
-            # A lock-external provider probe cannot authorize the write and
-            # would duplicate this fresh authority check on every request.
-            selected = self.bindings.resolve(binding_id=binding_id, **source)
-            expected = {"binding_id": binding_id, "source": source, "message": message, "command": command,
-                        "attachments": normalize_chat_image_attachments(attachments) or None, "origin": origin}
-            recovering_request = path.exists()
-            if recovering_request:
-                row = _read_json(path)
-                # Existing Lark journals predate persisted transport origin.
-                if any((row.get(key, "lark") if key == "origin" else row.get(key)) != value
-                       for key, value in expected.items()):
-                    raise ValueError("external request identity was reused with different content")
-                if row.get("status") != "prepared":
-                    return row
-            else:
-                row = {"schema_version": "loopx_chat_external_request_v0", "request_ref": request_ref,
-                       **expected, "status": "prepared", "session_id": None, "turn_id": None,
-                       "created_at": datetime.now(timezone.utc).isoformat()}
-                _atomic_write_json(path, row)
+        # A duplicate waits on its own request before taking the route fence;
+        # it must not hold up other messages while native cancellation runs.
+        with exclusive_file_lock(path, operation="admit_external_chat_request"):
+            with exclusive_file_lock(self.root / "source-fences" / f"{binding_id}.{source['source_ref']}.json", operation="route_external_chat_request"):
+                # Resolve after waiting for both fences, including exact replay.
+                # A lock-external provider probe cannot authorize the write and
+                # would duplicate this fresh authority check on every request.
+                selected = self.bindings.resolve(binding_id=binding_id, **source)
+                expected = {"binding_id": binding_id, "source": source, "message": message, "command": command,
+                            "attachments": normalize_chat_image_attachments(attachments) or None, "origin": origin}
+                recovering_request = path.exists()
+                if recovering_request:
+                    row = _read_json(path)
+                    # Existing Lark journals predate persisted transport origin.
+                    if any((row.get(key, "lark") if key == "origin" else row.get(key)) != value
+                           for key, value in expected.items()):
+                        raise ValueError("external request identity was reused with different content")
+                    if row.get("status") != "prepared":
+                        return row
+                else:
+                    row = {"schema_version": "loopx_chat_external_request_v0", "request_ref": request_ref,
+                           **expected, "status": "prepared", "session_id": None, "turn_id": None,
+                           "created_at": datetime.now(timezone.utc).isoformat()}
+                    _atomic_write_json(path, row)
+                try:
+                    row = self._admit_prepared(path, row, selected, recovering_request=recovering_request)
+                except ValueError:
+                    self._reject_prepared(path, row)
+                    raise
+            # Native inventory/cancellation needs the frozen request fence,
+            # not the route fence. New messages retain their own fresh checks.
+            if (defer_background_stop or row["status"] != "prepared"
+                    or row.get("command") != "stop" or not row.get("background_turn_ids")):
+                return row
             try:
-                return self._admit_prepared(path, row, selected, recovering_request=recovering_request)
+                return self._finish_background_stop(path, row, selected)
             except ValueError:
-                # Invalid commands and definitive grant denials are terminal
-                # admission outcomes; recovery must not retry them forever.
-                row.update(status="rejected", response="原授权或命令格式不可用，本条未进入执行队列；请用 /agents 或 /project 查看可用入口。")
-                _atomic_write_json(path, row)
+                self._reject_prepared(path, row)
                 raise
 
+    @staticmethod
+    def _reject_prepared(path: Path, row: dict[str, Any]) -> None:
+        # Definitive command/grant denials must not retry indefinitely.
+        row.update(status="rejected", response="原授权或命令格式不可用，本条未进入执行队列；请用 /agents 或 /project 查看可用入口。")
+        _atomic_write_json(path, row)
+
     def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, Any],
                         *, recovering_request: bool = False) -> dict[str, Any]:
         controller = self.controller
@@ -264,17 +280,7 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
                 row["background_turn_ids"] = plan.get("background_turn_ids", [])
             _atomic_write_json(path, row)
             if operation == "stop" and row["background_turn_ids"]:
-                from .codex_background_control import completed_background_turns, interrupt_completed_background_turns
-                row["background_turn_ids"] = completed_background_turns(
-                    controller, row["session_id"], row["background_turn_ids"])
-                _atomic_write_json(path, row)
-                # Reuse Core's response projection after narrowing. No live
-                # terminals is idle; failed inventory never reaches this point.
-                plan = effect_runtime_result("collaboration.conversation.request", {
-                    "request": {key: value for key, value in row.items() if key != "attachments"},
-                    "current_session": current, "binding": selected["binding"], "agent_target": target})
-                if row["background_turn_ids"]:
-                    interrupt_completed_background_turns(controller, row["session_id"], row["background_turn_ids"])
+                return row  # Finish outside the route fence, under this request.
             if operation == "stop" and row["session_id"] and row["turn_id"]:
                 controller.interrupt_turn(session_id=row["session_id"], turn_id=row["turn_id"])
             elif operation == "new" and row["session_id"]:
@@ -304,6 +310,26 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
         _atomic_write_json(path, row)
         return row
 
+    def _finish_background_stop(self, path: Path, row: dict[str, Any], selected: dict[str, Any]) -> dict[str, Any]:
+        from .codex_background_control import completed_background_turns, interrupt_completed_background_turns
+        from ...control_plane.effect_runtime import effect_runtime_result
+
+        controller = self.controller
+        row["background_turn_ids"] = completed_background_turns(
+            controller, row["session_id"], row["background_turn_ids"])
+        _atomic_write_json(path, row)
+        # Reuse the typed response after narrowing. No live terminals is idle;
+        # failed inventory never reaches this point. Later work is ineligible.
+        plan = effect_runtime_result("collaboration.conversation.request", {
+            "request": {key: value for key, value in row.items() if key != "attachments"},
+            "current_session": controller.store.load_session(row["session_id"]),
+            "binding": selected["binding"], "agent_target": row.get("agent_target")})
+        if row["background_turn_ids"]:
+            interrupt_completed_background_turns(controller, row["session_id"], row["background_turn_ids"])
+        row.update(status="command_completed", response_code=plan["response_code"])
+        _atomic_write_json(path, row)
+        return row
+
     def _record_steward_ingress(self, row: dict[str, Any], selected: dict[str, Any],
                                session: dict[str, Any], client_turn_id: str) -> None:
         if selected["binding"]["context_kind"] != "steward":

loopx/extensions/lark/goal_topic_runtime_service.py

@@ -5,6 +5,7 @@
 from collections.abc import Callable, Mapping
 from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime, timezone
+from functools import partial
 import hashlib
 import logging
 from pathlib import Path
@@ -311,7 +312,8 @@ def handle_review_callback(
                             stop=stop,
                             runtime_root=self.runtime_root,
                             answer=answer,
-                            private_admitter=(self.private_conversations.admit if self.private_conversations is not None else None),
+                            private_admitter=(partial(self.private_conversations.admit, defer_background_stop=True)
+                                if self.private_conversations is not None else None),
                             proposal_deliverer=deliver_proposals,
                             review_callback_handler=(
                                 handle_review_callback

loopx/extensions/lark/private_conversations.py

@@ -332,7 +332,8 @@ def _inbox(self, record: dict[str, Any], *, observation: dict[str, Any] | None =
             raise ValueError("private source could not be captured")
         return path
 
-    def admit(self, profile: str, event: dict[str, Any]) -> dict[str, Any]:
+    def admit(self, profile: str, event: dict[str, Any], *,
+              defer_background_stop: bool = False) -> dict[str, Any]:
         try:
             binding = self._binding(profile)
             source = lark_private_source(provider_ref=binding["provider_ref"], event=event)
@@ -425,7 +426,8 @@ def admit(self, profile: str, event: dict[str, Any]) -> dict[str, Any]:
                 command = "unsupported"
             try:
                 admitted = self.core.admit(binding_id=binding["binding_id"], source=source,
-                    request_ref=request, message=text, command=command, attachments=attachments)
+                    request_ref=request, message=text, command=command, attachments=attachments,
+                    defer_background_stop=defer_background_stop)
             except ValueError:
                 record.update(status="rejected", response=("群助手仅处理当前项目和话题，不能选择个人 Agent 会话；/status 查看状态，/help 查看用法。" if binding.get("audience") == "group" else "操作或原授权不可用；Agent 请先用 /agents 查看确切命令，/project 返回项目对话。新委托请使用 /delegate --tokens N 具体目标，确认或取消请使用原预览中的完整命令。"))
                 _atomic_write_json(path, record)
@@ -436,7 +438,10 @@ def admit(self, profile: str, event: dict[str, Any]) -> dict[str, Any]:
                 record.update(status="rejected", response="队列已满，本条没有被受理；请稍后重试。")
                 _atomic_write_json(path, record)
                 return {"status": "queue_rejected"}
-            record.update(status=admitted["status"], session_id=admitted["session_id"],
+            # Keep the existing recoverable capture until Core has performed
+            # the frozen stop. Older transport readers can recover it too.
+            record.update(status="captured" if admitted["status"] == "prepared" else admitted["status"],
+                          session_id=admitted["session_id"],
                           turn_id=admitted["turn_id"], response=admitted.get("response"), response_code=admitted.get("response_code"))
             _atomic_write_json(path, record)
             status = "durably_accepted" if admitted["status"] == "accepted" else (
@@ -765,6 +770,23 @@ def reconcile_request(self, path: Path) -> int:
             # read its result before retrying admission from a stale snapshot.
             with exclusive_file_lock(path, operation="capture_private_chat_request"):
                 row = _read_json(path)
+                native_path = self.core.root / f"{row['request_ref']}.json"
+                if row["status"] == "captured" and native_path.exists():
+                    native = self.core.read_request(row["request_ref"])
+                    if (native.get("command") == "stop" and native.get("target_recorded") is True
+                            and all(native.get(key) == row.get(key)
+                                    for key in ("request_ref", "binding_id", "source"))):
+                        # This source was already admitted and target-frozen.
+                        # Core revalidates recovery; delivery below re-reads the
+                        # provider source. No second content admission is needed.
+                        self.core.recover(request_ref=row["request_ref"])
+                        native = self.core.read_request(row["request_ref"])
+                        if native["status"] == "prepared":
+                            return 0
+                        if native["status"] == "command_completed":
+                            row.update(**{key: native.get(key) for key in
+                                ("status", "session_id", "turn_id", "response", "response_code")})
+                            _atomic_write_json(path, row)
             if row["status"] == "captured":
                 self.admit(row["profile"], row["event"])
                 row = _read_json(path)