fix(lark): share fresh source verification within reply operations

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

@@ -69,6 +69,17 @@ Future callback repeatedly reschedules unfinished work. File/Core and native
 protocol fixtures qualify this scheduling path separately from installed service,
 provider latency, complete stop acknowledgement and first-visible Web results.
 
+Native Lark sends and receipt recovery use one operation-bound source verifier
+for the current App, exact profile/chat/message and Core grant, as existing
+reply edits already do. The observed Bot name must match the Inbox metadata;
+a rename still holds the first send while metadata refreshes. The verifier
+joins source reads and rechecks the exact binding before returning. A lost
+readback retains the same attempt and re-verifies without resending. No identity
+or grant is cached across operations, and legacy membership-only callbacks
+retain separate sender authentication. File/Core fixtures qualify this transport
+transaction, including late revocation; they do not qualify installed provider
+latency, the complete stop SLA or first-visible Web results.
+
 The packaged settings journey, source revocation/recovery, duplicate events,
 native queue/stop and two-App isolation have synthetic-provider regression and
 browser coverage. A native Codex source canary separately qualifies distinct

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

@@ -49,6 +49,14 @@ canonical journal，并重新核验来源与 Core。扫描前清除提示，避
 反复重排尚未完成的工作。File/Core 与原生协议 fixture 单独验证这一调度路径，
 不替代安装态服务、provider 延迟、完整停止回执及 Web 首次可见结果的验收。
 
+原生 Lark 新发送和回执恢复在单次操作内，由同一来源校验核验当前 App、确切
+profile/chat/message 及 Core grant，与既有回复编辑路径一致。观测到的 Bot 名称
+必须匹配 Inbox 元数据；改名时仍先停止发送，刷新元数据后再继续。校验等待全部
+来源读取完成，并在返回前重读确切绑定。丢失读回保留同一 attempt，重新校验但不
+重复发送。操作之间不缓存身份或 grant；旧 membership-only callback 继续独立验证
+发送身份。File/Core fixture 验证这一传输事务及读取期间撤权，不认证安装态 provider
+延迟、完整停止 SLA 或 Web 首次可见结果。
+
 源码 canary 已验证打包页面的桌面/窄屏、撤权拒绝与恢复、慢会话期间另一 App 完成、
 后续消息持久排队、exact stop，以及重启后原会话恢复和已确认回复不重复发送。
 这些是合成 provider/协议验收；真实原生 Codex 另行验证独立线程与上下文隔离。

loopx/extensions/lark/inbox_reply.py

@@ -33,6 +33,10 @@
 )
 
 CommandRunner = Callable[[Sequence[str]], Mapping[str, Any]]
+# Internal transport callback: return the freshly Core-verified App name and
+# exact source verdict separately. A changed source is not an identity mismatch.
+# Neither observation is reused by a later operation.
+SourceAuthorityVerifier = Callable[[str, str, str], tuple[str | None, bool]]
 BOT_IDENTITY_VERIFY_ATTEMPTS = 3
 
 
@@ -227,6 +231,30 @@ def _bot_identity_verified(
     return False
 
 
+def _verified_sender_and_source(
+    *, runner: CommandRunner, base: Sequence[str], profile: str, chat_id: str,
+    message_id: str, expected_name: str,
+    source_membership_verifier: Callable[[], bool] | None,
+    source_authority_verifier: SourceAuthorityVerifier | None,
+) -> tuple[bool, bool]:
+    if source_authority_verifier is not None:
+        if source_membership_verifier is not None or not message_id:
+            raise ValueError("one source-bound verification owner is required")
+        observation = source_authority_verifier(profile, chat_id, message_id)
+        if not isinstance(observation, tuple) or len(observation) != 2:
+            return False, False
+        name, source_verified = observation
+        verified = isinstance(name, str) and bool(name)
+        return verified and name == expected_name, source_verified is True
+    identity = _bot_identity_verified(runner=runner, base=base, expected_name=expected_name)
+    if not identity:
+        return False, False
+    membership = ({"returncode": 0 if source_membership_verifier() else 1}
+                  if source_membership_verifier is not None else _call(runner,
+                      [*base, "im", "chats", "get", "--chat-id", chat_id, "--as", "bot", "--format", "json"]))
+    return True, membership.get("returncode") == 0
+
+
 def _result(
     *,
     status: str,
@@ -285,6 +313,7 @@ def _deliver_lark_inbox_outbound(
     provider_preflight: bool = False,
     runner: CommandRunner = _default_runner,
     source_membership_verifier: Callable[[], bool] | None = None,
+    source_authority_verifier: SourceAuthorityVerifier | None = None,
     before_send: Callable[[str], Mapping[str, Any]] | None = None,
     delivery_attempt_recorder: Callable[[Mapping[str, str | None]], None] | None = None,
     short_message_limit: int | None = DEFAULT_LARK_TEXT_LIMIT,
@@ -394,10 +423,11 @@ def _deliver_lark_inbox_outbound(
         )
 
     base = ["lark-cli", "--profile", profile]
-    identity_verified = _bot_identity_verified(
-        runner=runner,
-        base=base,
-        expected_name=str(reply_config["bot_display_name"]),
+    identity_verified, membership_verified = _verified_sender_and_source(
+        runner=runner, base=base, profile=profile, chat_id=chat_id,
+        message_id=source_message_id, expected_name=str(reply_config["bot_display_name"]),
+        source_membership_verifier=source_membership_verifier,
+        source_authority_verifier=source_authority_verifier,
     )
     if not identity_verified:
         return _result(
@@ -409,25 +439,7 @@ def _deliver_lark_inbox_outbound(
             format_preflight_passed=True,
         )
 
-    if source_membership_verifier is not None:
-        membership = {"returncode": 0 if source_membership_verifier() else 1}
-    else:
-        membership = _call(
-            runner,
-            base
-            + [
-                "im",
-                "chats",
-                "get",
-                "--chat-id",
-                chat_id,
-                "--as",
-                "bot",
-                "--format",
-                "json",
-            ],
-        )
-    if membership.get("returncode") != 0:
+    if not membership_verified:
         return _result(
             status="gate_required",
             ok=False,
@@ -759,6 +771,7 @@ def reply_lark_event_inbox(
     provider_preflight: bool = False,
     runner: CommandRunner = _default_runner,
     source_membership_verifier: Callable[[], bool] | None = None,
+    source_authority_verifier: SourceAuthorityVerifier | None = None,
     before_send: Callable[[str], Mapping[str, Any]] | None = None,
     delivery_attempt_recorder: Callable[[Mapping[str, str | None]], None] | None = None,
     short_message_limit: int | None = DEFAULT_LARK_TEXT_LIMIT,
@@ -774,6 +787,13 @@ def reply_lark_event_inbox(
 
     Intermediate admission/progress replies pass ``finalize_reactions=False``;
     their verified delivery does not settle the source's processing lifecycle.
+
+    Native Core callers may supply ``source_authority_verifier`` instead of
+    the legacy membership callback. It must freshly verify the exact supplied
+    profile/chat/message and Core grants, returning the observed Bot name and
+    source verification result separately.
+    The sender consumes that result once; retries invoke the callback again.
+    Ordinary/legacy callers retain independent identity and membership checks.
     """
 
     result = _deliver_lark_inbox_outbound(
@@ -786,6 +806,7 @@ def reply_lark_event_inbox(
         provider_preflight=provider_preflight,
         runner=runner,
         source_membership_verifier=source_membership_verifier,
+        source_authority_verifier=source_authority_verifier,
         before_send=before_send,
         delivery_attempt_recorder=delivery_attempt_recorder,
         short_message_limit=short_message_limit,
@@ -808,6 +829,7 @@ def verify_lark_inbox_reply(
     attempt: Mapping[str, Any],
     runner: CommandRunner = _default_runner,
     source_membership_verifier: Callable[[], bool] | None = None,
+    source_authority_verifier: SourceAuthorityVerifier | None = None,
     finalize_reactions: bool = True,
     attachment_keys: tuple[str, ...] = (),
     attachment_names: tuple[str, ...] = (),
@@ -864,36 +886,13 @@ def verify_lark_inbox_reply(
             "blocker": "provider_delivery_intent_conflict",
         }
     base = ["lark-cli", "--profile", profile]
-    if not _bot_identity_verified(
-        runner=runner,
-        base=base,
+    identity, membership = _verified_sender_and_source(
+        runner=runner, base=base, profile=profile, chat_id=chat_id, message_id=message_id,
         expected_name=str(reply_config.get("bot_display_name") or ""),
-    ):
-        return {
-            "ok": False,
-            "verification_performed": False,
-            "reply_verified": False,
-            "blocker": "provider_verification_unavailable",
-        }
-    if source_membership_verifier is not None:
-        membership = {"returncode": 0 if source_membership_verifier() else 1}
-    else:
-        membership = _call(
-            runner,
-            [
-                *base,
-                "im",
-                "chats",
-                "get",
-                "--chat-id",
-                chat_id,
-                "--as",
-                "bot",
-                "--format",
-                "json",
-            ],
-        )
-    if membership.get("returncode") != 0:
+        source_membership_verifier=source_membership_verifier,
+        source_authority_verifier=source_authority_verifier,
+    )
+    if not identity or not membership:
         return {
             "ok": False,
             "verification_performed": False,
@@ -987,7 +986,8 @@ def update_lark_inbox_reply(
     *, project: str | Path, config_path: str | Path, message_id: str,
     text: str, previous_text: str, attempt: Mapping[str, Any],
     runner: CommandRunner = _default_runner,
-    source_membership_verifier: Callable[[], bool],
+    source_membership_verifier: Callable[[], bool] | None = None,
+    source_authority_verifier: SourceAuthorityVerifier | None = None,
     before_send: Callable[[str], Any],
     delivery_attempt_recorder: Callable[[Any], None],
     finalize_reactions: bool = False,
@@ -1026,7 +1026,16 @@ def update_lark_inbox_reply(
     body = {"msg_type": "post", "content": content}
     if len(json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) > LARK_POST_REQUEST_MAX_BYTES:
         return {"ok": False, "reply_verified": False, "blocker": "provider_update_too_large"}
-    if not source_membership_verifier():
+    if source_authority_verifier is not None:
+        identity, membership = _verified_sender_and_source(
+            runner=runner, base=base, profile=profile, chat_id=chat_id, message_id=message_id,
+            expected_name=str(reply.get("bot_display_name") or ""),
+            source_membership_verifier=source_membership_verifier,
+            source_authority_verifier=source_authority_verifier)
+        source_verified = identity and membership
+    else:
+        source_verified = source_membership_verifier is not None and source_membership_verifier()
+    if not source_verified:
         return {"ok": False, "reply_verified": False, "blocker": "provider_verification_unavailable"}
     ref = str(attempt["message_ref"])
     read_args = [*base, "im", "+messages-mget", "--message-ids", ref, "--as", "bot", "--no-reactions", "--format", "json"]

loopx/extensions/lark/private_conversations.py

@@ -141,7 +141,8 @@ def _completed_stop_root(self, record: Mapping[str, Any], binding: Mapping[str,
         return None
 
     def _source_message(self, record: dict[str, Any], *, selected: dict[str, Any] | None = None,
-                        render_content: bool = True) -> Mapping[str, Any] | None:
+                        render_content: bool = True,
+                        sender_identity: dict[str, str] | None = None) -> Mapping[str, Any] | None:
         """Read the exact source under this App, then recheck the Core audience.
 
         A p2p source read proves the destination without requiring group-member
@@ -174,7 +175,8 @@ def _source_message(self, record: dict[str, Any], *, selected: dict[str, Any] |
                     if current != read_binding or current != selected["binding"]:
                         return None  # Revocation/replacement while reads ran is not a delivery grant.
                 else:
-                    selected = self.bindings.resolve(binding_id=record["binding_id"], **record["source"])
+                    selected = self.bindings.resolve(binding_id=record["binding_id"],
+                                                     include_observation=sender_identity is not None, **record["source"])
             native_path = self.core.root / f"{record['request_ref']}.json"
             native = _read_json(native_path) if native_path.exists() else {}
             if native.get("agent_target"):
@@ -188,6 +190,8 @@ def rendered() -> Mapping[str, Any]:
             # Resolve has just verified this App for this invocation. Reuse
             # that observation for mention checks, never across verifications.
             observation = selected.get("observation")
+            if sender_identity is not None and observation is not None:
+                sender_identity["bot_display_name"] = str(observation.get("bot_display_name") or "")
             stop_root: Mapping[str, Any] | None = None
             completed_stop = False
             if group:
@@ -302,13 +306,34 @@ def sources() -> tuple[Mapping[str, Any] | None, Mapping[str, Any] | None]:
                         message["_command_input"] = self._group_text_command_input(message, observation)
                         if not message["_command_input"]:
                             return None  # A display rendering cannot supply missing canonical text.
+                # Outbound source reads must join before consuming this
+                # operation's proof. Admission retains Core's fenced rejection
+                # and its existing response when a binding changes mid-read.
+                if not render_content:
+                    current = next((row for row in self.bindings.read()["bindings"]
+                                    if row.get("binding_id") == record["binding_id"]), None)
+                    if current != selected["binding"]:
+                        return None
+                if sender_identity is not None:
+                    if observation is None:
+                        return None
                 return message
         except (KeyError, ValueError, OSError):
             pass
         return None
 
-    def _source_verified(self, record: dict[str, Any]) -> bool:
-        return self._source_message(record, render_content=False) is not None
+    def _source_verified(self, record: dict[str, Any], *, sender_identity: dict[str, str] | None = None) -> bool:
+        return self._source_message(record, render_content=False, sender_identity=sender_identity) is not None
+
+    def _source_authority(self, record: dict[str, Any], profile: str, chat_id: str,
+                          message_id: str) -> tuple[str | None, bool]:
+        """Verify one exact outbound operation; retain no identity or grant."""
+        if (profile != record["profile"] or chat_id != record["event"]["chat_id"]
+                or message_id != record["event"]["message_id"]):
+            return None, False
+        identity: dict[str, str] = {}
+        verified = self._source_verified(record, sender_identity=identity)
+        return identity.get("bot_display_name"), verified
 
     @staticmethod
     def _group_text_command_input(message: Mapping[str, Any], observation: Mapping[str, Any]) -> str | None:
@@ -569,7 +594,7 @@ def _deliver(self, path: Path, record: dict[str, Any], phase: str, text: str, *,
         kwargs = dict(project=self.runtime_root, config_path=config,
             message_id=record["event"]["message_id"], text=text, runner=self._reply_runner,
             finalize_reactions=phase not in {"admission", "progress"} and not (phase == "terminal" and record.get("commission_resources")),
-            source_membership_verifier=lambda: self._source_verified(record))
+            source_authority_verifier=lambda profile, chat, message: self._source_authority(record, profile, chat, message))
         progress = record["deliveries"].get("progress") or {}
         stream = record.get("stream") or {}
         edit = phase_state.get("edit_from")
@@ -665,7 +690,8 @@ def before_send(_intent: str) -> dict[str, bool]:
             message_id=record["event"]["message_id"], text=text,
             previous_text=_presentation_text(stream.get("confirmed_text") or progress["text"]),
             attempt=stream.get("confirmed_attempt") or progress["attempt"],
-            runner=self._reply_runner, source_membership_verifier=lambda: self._source_verified(record),
+            runner=self._reply_runner,
+            source_authority_verifier=lambda profile, chat, message: self._source_authority(record, profile, chat, message),
             before_send=before_send, delivery_attempt_recorder=lambda value: confirmed.update(value))
         stream["blocker"] = result.get("blocker")
         if result.get("blocker") == "provider_update_edit_limit":