From 3672381d1632a97c2f73315c6f3b790913759318 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 26 Oct 2026 07:20:31 +0800
Subject: [PATCH 1/2] fix(lark): project original App images with verified
 resource receipts

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../app-conversation-and-async-inbox-v0.md    |  20 +-
 ...p-conversation-and-async-inbox-v0.zh-CN.md |  13 +-
 loopx/extensions/lark/inbox_reply.py          |  25 ++-
 loopx/extensions/lark/post_delivery.py        |   6 +-
 .../lark/presentation/markdown_parts.py       |   3 +
 .../lark/presentation/markdown_post.py        |  42 ++--
 .../extensions/lark/private_conversations.py  |  41 +++-
 loopx/extensions/lark/return_files.py         | 121 ++++++++---
 tests/test_lark_app_image_projection.py       | 197 ++++++++++++++++++
 tests/test_lark_post_images.py                |  48 +++++
 10 files changed, 449 insertions(+), 67 deletions(-)
 create mode 100644 tests/test_lark_app_image_projection.py
 create mode 100644 tests/test_lark_post_images.py

diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
index 762413aad8..1c537fc7fe 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -262,12 +262,20 @@ connection to revoke it; rollback must preserve its private delivery journals.
 
 Use **Continue original conversation** or the explicitly selected App steward
 entry above to reach this path. Old ordinary history is retained separately.
-This checkpoint covers text projection, not reply/thread relation mirroring,
-attachment transfer, retroactive insertion into Lark chronology or all-channel
-broadcast. An App attachment produces an explicit unsynchronized notice and a
-partial outbox, never a full-delivery claim. Synthetic real-HTTP/Core journeys
-cover repeated inputs, long text, restart, missing/unknown receipts, revocation
-and provider replacement. Installed provider readback remains a separate gate.
+This checkpoint covers text and bounded App image projection, not reply/thread
+relation mirroring, retroactive insertion into Lark chronology or all-channel
+broadcast. Canonical App images reuse the existing resource uploader and post
+sender, retaining caption and ordered inline images in the original private
+chat. Long captions keep images on the final concrete post only. Every upload
+and send rechecks the original source and current Core message/grant; completion
+also requires independent image-byte readback. Failed verification recovers the
+same message receipt without another upload or send. An upload whose provider
+acknowledgement is unknown, or a started send without a message locator, stays
+unconfirmed after restart. Earlier delivered attachment placeholders retain
+their frozen receipts and partial status; upgrades do not silently replay them.
+Synthetic real-HTTP/Core journeys cover repeated inputs, complete images and
+long captions, restart, missing/unknown receipts, byte mismatch, revocation and
+provider replacement. Installed provider readback remains a separate gate.
 
 The implementation decision follows primary-source comparisons, not a claim
 that shared context already guarantees two-way delivery:
diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
index 30cc8ee7cd..27d1005270 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
@@ -186,10 +186,15 @@ Core 重查 Session、当前绑定与受众，provider 核验原来源。消息
 关闭既有连接即可撤销授权；回滚须保留私有投递 journal。
 
 通过上述 **继续原会话** 或明确选定的 App 管家入口使用该路径；旧普通历史继续
-独立保留。本检查点覆盖文本投影，不覆盖回复/话题关系镜像、附件传输、向 Lark
-历史回插旧时间或全渠道广播。App 附件会产生明确的未同步提示，并保留 partial
-outbox，不声明完整送达。真实 HTTP/Core 配合合成 provider 的旅程覆盖重复输入、
-长正文、重启、回执缺失/未知、撤权及 App 身份替换；安装态 provider 读回另行验收。
+独立保留。本检查点覆盖文本及有界 App 图片投影，不覆盖回复/话题关系镜像、向
+Lark 历史回插旧时间或全渠道广播。Core 中的 App 图片复用既有资源上传与 post
+发送器，正文及有序行内图片留在原私聊；长正文只在最后一个实际分段携带图片。
+每次上传和发送均重新核验原来源、当前 Core 消息与授权，完成还须独立读回图片
+字节。验证失败时沿同一消息回执恢复，不再次上传或发送。上传确认未知，或发送
+已开始但缺少消息定位符时，重启后仍保留未确认。以前已发送的附件占位说明保留
+冻结回执及 partial 状态，升级不自动重放。真实 HTTP/Core 配合合成 provider 的
+旅程覆盖重复输入、完整图片和长正文、重启、回执缺失/未知、字节不符、撤权及
+App 身份替换；安装态 provider 读回另行验收。
 
 实现判断来自一手源码比较，不把共享上下文视为双向送达已经成立：
 
diff --git a/loopx/extensions/lark/inbox_reply.py b/loopx/extensions/lark/inbox_reply.py
index 129baa08fa..dbe0fcd87f 100644
--- a/loopx/extensions/lark/inbox_reply.py
+++ b/loopx/extensions/lark/inbox_reply.py
@@ -331,6 +331,7 @@ def _deliver_lark_inbox_outbound(
     finalize_reactions: bool = True,
     attachment_keys: tuple[str, ...] = (),
     attachment_names: tuple[str, ...] = (),
+    image_keys: tuple[str, ...] = (),
 ) -> dict[str, Any]:
     """Deliver through one inbox-configured bot with exact provider readback.
 
@@ -377,6 +378,9 @@ def _deliver_lark_inbox_outbound(
     if attachment_keys and (not markdown or len(attachment_keys) > 4
                             or any(not re.fullmatch(r"file_[A-Za-z0-9_-]{1,240}", key) for key in attachment_keys)):
         raise ValueError("result files require bounded post attachment keys")
+    if image_keys and (not markdown or len(image_keys) > 4
+                       or any(not re.fullmatch(r"img_[A-Za-z0-9_-]{1,240}", key) for key in image_keys)):
+        raise ValueError("App images require bounded post image keys")
     reply_text = normalize_lark_outbound_text(
         text,
         limit=None if source_event is not None else short_message_limit,
@@ -425,6 +429,7 @@ def _deliver_lark_inbox_outbound(
                 **({"delivery_id": delivery_id} if delivery_id is not None else {}),
                 **({"content_format": "markdown"} if markdown else {}),
                 **({"attachment_keys": attachment_keys} if attachment_keys else {}),
+                **({"image_keys": image_keys} if image_keys else {}),
             },
             ensure_ascii=False,
             sort_keys=True,
@@ -444,7 +449,7 @@ def _deliver_lark_inbox_outbound(
 
     base = ["lark-cli", "--profile", profile]
     content_args = (
-        ["--msg-type", "post", "--content", lark_markdown_post_content(reply_text)]
+        ["--msg-type", "post", "--content", lark_markdown_post_content(reply_text, image_keys=image_keys)]
         if markdown else ["--text", reply_text]
     )
     if attachment_keys:
@@ -542,7 +547,7 @@ def verify_source() -> tuple[bool, bool]:
     provider_preview_verified = bool(
         preview.get("returncode") == 0
         and (
-            lark_markdown_preview_matches(text=reply_text, payload=_json_object(preview.get("stdout")), attachment_keys=attachment_keys)
+            lark_markdown_preview_matches(text=reply_text, payload=_json_object(preview.get("stdout")), attachment_keys=attachment_keys, image_keys=image_keys)
             if markdown else lark_provider_preview_matches_outbound(
                 outbound_text=reply_text, payload=_json_object(preview.get("stdout")),
             )
@@ -594,7 +599,7 @@ def verify_source() -> tuple[bool, bool]:
                         **({"reply_placement": reply_placement} if reply_placement is not None else {}),
                         **({"delivery_id": delivery_id} if delivery_id is not None else {}),
                         "finalize_reactions": finalize_reactions,
-                        "attachment_keys": attachment_keys, "attachment_names": attachment_names})
+                        "attachment_keys": attachment_keys, "attachment_names": attachment_names, "image_keys": image_keys})
     guidance = None
     if before_send is not None:
         # Bind review to destination/profile as well as content and placement.
@@ -747,7 +752,7 @@ def verify_source() -> tuple[bool, bool]:
         and readback_message is not None
         and (reply_placement is None or _private_root_readback_matches(readback_message, chat_id))
         and (lark_markdown_readback_matches(text=reply_text, message=readback_message, attachment_keys=attachment_keys,
-                                          attachment_names=attachment_names)
+                                          attachment_names=attachment_names, image_keys=image_keys)
              if markdown else lark_readback_matches_outbound(
             outbound_text=reply_text,
             message=readback_message,
@@ -829,6 +834,7 @@ def reply_lark_event_inbox(
     finalize_reactions: bool = True,
     attachment_keys: tuple[str, ...] = (),
     attachment_names: tuple[str, ...] = (),
+    image_keys: tuple[str, ...] = (),
 ) -> dict[str, Any]:
     """Reply with the explicit inbox-configured bot and placement policy.
 
@@ -873,6 +879,7 @@ def reply_lark_event_inbox(
         finalize_reactions=finalize_reactions,
         attachment_keys=attachment_keys,
         attachment_names=attachment_names,
+        image_keys=image_keys,
     )
 
     result.setdefault("content_format", "markdown" if content_format == "markdown"
@@ -895,6 +902,7 @@ def verify_lark_inbox_reply(
     finalize_reactions: bool = True,
     attachment_keys: tuple[str, ...] = (),
     attachment_names: tuple[str, ...] = (),
+    image_keys: tuple[str, ...] = (),
     resume_parts: bool = False,
     before_send: Callable[[str], Mapping[str, Any]] | None = None,
 ) -> dict[str, Any]:
@@ -930,7 +938,7 @@ def verify_lark_inbox_reply(
     chat_id = str(reply_config.get("chat_id") or "")
     if batch:
         original = reply_lark_event_inbox(project=project, config_path=config_path,
-            message_id=message_id, text=text, content_format="markdown", attachment_keys=attachment_keys,
+            message_id=message_id, text=text, content_format="markdown", attachment_keys=attachment_keys, image_keys=image_keys,
             reply_placement=reply_placement, delivery_id=delivery_id)
         receipt = original["idempotency_key"]
         return verify_post_parts(receipt=receipt, intent_digest=_intent_digest(profile, chat_id, receipt),
@@ -942,10 +950,10 @@ def verify_lark_inbox_reply(
                 **({"delivery_id": delivery_id} if delivery_id is not None else {}),
                 "before_send": before_send, "finalize_reactions": finalize_reactions,
                 "text": normalize_lark_outbound_text(text, limit=None, preserve_format=True),
-                "attachment_keys": attachment_keys, "attachment_names": attachment_names})
+                "attachment_keys": attachment_keys, "attachment_names": attachment_names, "image_keys": image_keys})
     preview = None
     candidates = [(text, "markdown")]
-    if not attachment_keys:
+    if not attachment_keys and not image_keys:
         # Historical text receipts are read-only compatibility, never a new
         # downgrade from an oversized Markdown reply.
         candidates.append((text, "text"))
@@ -961,6 +969,7 @@ def verify_lark_inbox_reply(
             execute=False,
             runner=runner,
             attachment_keys=attachment_keys,
+            image_keys=image_keys,
         )
         receipt = str(candidate.get("idempotency_key") or "")
         if (
@@ -1037,7 +1046,7 @@ def verify_lark_inbox_reply(
         }
     verified = (
         lark_markdown_readback_matches(text=reply_text, message=message, attachment_keys=attachment_keys,
-                                      attachment_names=attachment_names)
+                                      attachment_names=attachment_names, image_keys=image_keys)
         if markdown
         else lark_readback_matches_outbound(
             outbound_text=reply_text,
diff --git a/loopx/extensions/lark/post_delivery.py b/loopx/extensions/lark/post_delivery.py
index 3799aa3540..56aff757a3 100644
--- a/loopx/extensions/lark/post_delivery.py
+++ b/loopx/extensions/lark/post_delivery.py
@@ -142,6 +142,7 @@ def options(index):
                 "attachment_names": arguments.get("attachment_names", ())
                 if final
                 else (),
+                "image_keys": arguments.get("image_keys", ()) if final else (),
             }
 
         result = deliver_manager_reply_parts(
@@ -161,7 +162,7 @@ def options(index):
                 int(state.get("delivery_part_stall_count") or 0) + 1
             )
             notice_options = options(0)
-            notice_options.update(attachment_keys=(), attachment_names=())
+            notice_options.update(attachment_keys=(), attachment_names=(), image_keys=())
             notice = deliver_stall_notice(
                 delivery_state=state,
                 delivery_path=path,
@@ -261,6 +262,7 @@ def verify_post_parts(*, receipt, intent_digest, attempt, arguments, resume=Fals
             finalize_reactions=False,
             attachment_keys=arguments.get("attachment_keys", ()) if final else (),
             attachment_names=arguments.get("attachment_names", ()) if final else (),
+            image_keys=arguments.get("image_keys", ()) if final else (),
         )
         if not result.get("reply_verified"):
             return result
@@ -286,7 +288,7 @@ def verify_post_parts(*, receipt, intent_digest, attempt, arguments, resume=Fals
 
 
 def attachment_reply(*, project, config_path, attempt, text):
-    """Files belong to the last concrete post, never to the composite locator."""
+    """Resources belong to the last concrete post, never the composite locator."""
     match = POST_PARTS_LOCATOR.fullmatch(str(attempt.get("message_ref") or ""))
     if not match:
         return text, attempt
diff --git a/loopx/extensions/lark/presentation/markdown_parts.py b/loopx/extensions/lark/presentation/markdown_parts.py
index 72d7a6923f..ddd33a0cb6 100644
--- a/loopx/extensions/lark/presentation/markdown_parts.py
+++ b/loopx/extensions/lark/presentation/markdown_parts.py
@@ -39,6 +39,9 @@ def post_request_bytes(text: str, template: Mapping[str, Any]) -> int:
         content.update(
             {key: value for key, value in authored.items() if key != "zh_cn"}
         )
+        locale = authored.get("zh_cn")
+        if isinstance(locale, Mapping) and isinstance(locale.get("content"), list):
+            content["zh_cn"] = {**locale, "content": [content["zh_cn"]["content"][0], *locale["content"][1:]]}
     body = {
         **template,
         "content": json.dumps(content, ensure_ascii=False, separators=(",", ":")),
diff --git a/loopx/extensions/lark/presentation/markdown_post.py b/loopx/extensions/lark/presentation/markdown_post.py
index c56ae912b2..0af7c1737f 100644
--- a/loopx/extensions/lark/presentation/markdown_post.py
+++ b/loopx/extensions/lark/presentation/markdown_post.py
@@ -129,17 +129,19 @@ def _normalize_strong_line(line: str) -> str:
     return "".join(chunks)
 
 
-def lark_markdown_post_content(text: str) -> str:
+def lark_markdown_post_content(text: str, *, image_keys: tuple[str, ...] = ()) -> str:
     """Render authored Markdown without CLI image fetching or link rewriting."""
     text = normalize_lark_markdown_emphasis(text)
     return json.dumps(
-        {"zh_cn": {"content": [[{"tag": "md", "text": text}]]}},
+        {"zh_cn": {"content": [[{"tag": "md", "text": text}],
+                              *[[{"tag": "img", "image_key": key}] for key in image_keys]]}},
         ensure_ascii=False,
         separators=(",", ":"),
     )
 
 
-def _single_markdown_post(value: Any, attachment_keys: tuple[str, ...] = ()) -> str | None:
+def _single_markdown_post(value: Any, attachment_keys: tuple[str, ...] = (),
+                          image_keys: tuple[str, ...] = ()) -> str | None:
     if isinstance(value, str):
         try:
             value = json.loads(value)
@@ -157,7 +159,9 @@ def _single_markdown_post(value: Any, attachment_keys: tuple[str, ...] = ()) ->
     if locale.get("title", "") != "":
         return None
     rows = locale.get("content")
-    if not isinstance(rows, list) or len(rows) != 1:
+    if not isinstance(rows, list) or len(rows) != 1 + len(image_keys):
+        return None
+    if rows[1:] != [[{"tag": "img", "image_key": key}] for key in image_keys]:
         return None
     row = rows[0]
     if not isinstance(row, list) or len(row) != 1:
@@ -170,7 +174,8 @@ def _single_markdown_post(value: Any, attachment_keys: tuple[str, ...] = ()) ->
     )
 
 
-def lark_markdown_preview_matches(*, text: str, payload: Mapping[str, Any], attachment_keys: tuple[str, ...] = ()) -> bool:
+def lark_markdown_preview_matches(*, text: str, payload: Mapping[str, Any],
+                                  attachment_keys: tuple[str, ...] = (), image_keys: tuple[str, ...] = ()) -> bool:
     data = payload.get("data")
     calls = payload.get("api")
     if calls is None and isinstance(data, Mapping):
@@ -182,13 +187,13 @@ def lark_markdown_preview_matches(*, text: str, payload: Mapping[str, Any], atta
     return (
         isinstance(body, Mapping)
         and body.get("msg_type") == "post"
-        and _single_markdown_post(body.get("content"), attachment_keys)
+        and _single_markdown_post(body.get("content"), attachment_keys, image_keys)
         == normalize_lark_markdown_emphasis(text)
     )
 
 
 def _readback_post(message: Mapping[str, Any], attachment_keys: tuple[str, ...],
-                   attachment_names: tuple[str, ...]) -> tuple[str, tuple[str, ...]] | None:
+                   attachment_names: tuple[str, ...], image_keys: tuple[str, ...]) -> tuple[str, tuple[str, ...]] | None:
     if message.get("msg_type", message.get("message_type")) != "post":
         return None
     if message.get("mentions") not in (None, []):
@@ -207,7 +212,7 @@ def _readback_post(message: Mapping[str, Any], attachment_keys: tuple[str, ...],
         if isinstance(actual, Mapping) and attachment_keys:
             files = actual.get("files")
         keys = tuple(file.get("key") for file in files if isinstance(file, Mapping)) if isinstance(files, list) else ()
-        actual = _single_markdown_post(actual, keys)
+        actual = _single_markdown_post(actual, keys, image_keys)
     else:
         if attachment_keys:
             # The CLI renders the post's attachment zone as trailing file tags.
@@ -220,6 +225,18 @@ def _readback_post(message: Mapping[str, Any], attachment_keys: tuple[str, ...],
                 return None
             files = [tag.attrib for tag in tags]
             actual = "\n".join(lines[:-len(attachment_keys)]).rstrip()
+        if image_keys:
+            # The CLI renders each inline image as its own Markdown line.
+            # Consume only the exact ordered resource keys, never a caption
+            # containing a similar image token or a missing/extra image.
+            lines = actual.rstrip().splitlines()
+            images = [re.fullmatch(r"!\[Image\]\((img_[A-Za-z0-9_-]{1,240})\)", line)
+                      for line in lines[-len(image_keys):]]
+            if len(images) != len(image_keys) or any(match is None for match in images):
+                return None
+            if tuple(match[1] for match in images) != image_keys:
+                return None
+            actual = "\n".join(lines[:-len(image_keys)]).rstrip()
     if not isinstance(actual, str) or not isinstance(files, list) or len(files) != len(attachment_keys):
         return None
     keys = []
@@ -242,9 +259,10 @@ def _readback_post(message: Mapping[str, Any], attachment_keys: tuple[str, ...],
 
 def lark_markdown_readback_attachment_keys(*, text: str, message: Mapping[str, Any],
                                          attachment_keys: tuple[str, ...] = (),
-                                         attachment_names: tuple[str, ...] = ()) -> tuple[str, ...] | None:
+                                         attachment_names: tuple[str, ...] = (),
+                                         image_keys: tuple[str, ...] = ()) -> tuple[str, ...] | None:
     """Resolve keys only from the exact post; this alone never verifies files."""
-    parsed = _readback_post(message, attachment_keys, attachment_names)
+    parsed = _readback_post(message, attachment_keys, attachment_names, image_keys)
     if parsed is None:
         return None
     actual, keys = parsed
@@ -255,7 +273,7 @@ def lark_markdown_readback_attachment_keys(*, text: str, message: Mapping[str, A
 
 def lark_markdown_readback_matches(*, text: str, message: Mapping[str, Any],
                                  attachment_keys: tuple[str, ...] = (),
-                                 attachment_names: tuple[str, ...] = ()) -> bool:
+                                 attachment_names: tuple[str, ...] = (), image_keys: tuple[str, ...] = ()) -> bool:
     """Accept the raw post or CLI's md text, never a plain-text lookalike."""
     return lark_markdown_readback_attachment_keys(text=text, message=message,
-        attachment_keys=attachment_keys, attachment_names=attachment_names) is not None
+        attachment_keys=attachment_keys, attachment_names=attachment_names, image_keys=image_keys) is not None
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index 1929d80ae7..1d5e0a7f5b 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -31,6 +31,7 @@
 from .inbox_reactions import mark_lark_event_inbox_processing, mark_lark_event_inbox_received
 from .outbound import present_lark_conversation_text as _presentation_text
 from .private_images import private_message_caption, private_message_images
+from .return_files import uploaded_chat_images, verify_chat_images
 from .private_progress import (ACTIVITY_UPDATE_INTERVAL_SEC, UPDATE_INTERVAL_SEC, legacy_recovery_repairs,
                                presentation_phase, presentation_stream, project_progress)
 
@@ -818,7 +819,7 @@ def _feedback(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], P
     def _deliver(self, path: Path, record: dict[str, Any], phase: str, text: str, *, inbox: Callable[[], Path],
                  edit_only: bool = False, authorization_check: Callable[[], bool] | None = None,
                  reply_placement: str | None = None, delivery_id: str | None = None,
-                 finalize_source: bool = True) -> bool:
+                 finalize_source: bool = True, image_attachments: list[dict[str, Any]] | None = None) -> bool:
         if not text:
             return False
         if reply_placement is not None and (record["event"].get("chat_type") != "p2p"
@@ -838,11 +839,29 @@ def _deliver(self, path: Path, record: dict[str, Any], phase: str, text: str, *,
             text = _presentation_text(text)
         if phase_state.get("text") not in (None, text):
             raise ValueError("delivery content changed after an attempt")
+        if image_attachments and (kind != "web" or reply_placement != "chat_root" or edit_only):
+            raise ValueError("App image projection requires its original private outbox")
         record_event_timing(record["profile"], record["event"]["message_id"], "delivery_started")
         config = inbox()
         record_event_timing(record["profile"], record["event"]["message_id"], "delivery_inbox_ready")
+        def image_authorized() -> bool:
+            if authorization_check is None or not authorization_check():
+                return False
+            name, verified = self._source_authority(record, record["profile"],
+                record["event"]["chat_id"], record["event"]["message_id"])
+            return bool(name) and verified
+
+        image_keys = ()
+        if image_attachments:
+            image_keys = uploaded_chat_images(root=self.runtime_root, profile=record["profile"],
+                provider_ref=record["binding_id"], attachments=image_attachments, private_transport=self,
+                before_upload=image_authorized, verify_only=bool(phase_state.get("started") or phase_state.get("attempt")))
+            if phase_state.get("image_keys", list(image_keys)) != list(image_keys):
+                raise ValueError("App image delivery resources changed after an attempt")
+            phase_state["image_keys"] = list(image_keys)
         kwargs = dict(project=self.runtime_root, config_path=config,
             message_id=record["event"]["message_id"], text=text, runner=self._reply_runner,
+            **({"image_keys": image_keys} if image_keys else {}),
             **({"reply_placement": reply_placement} if reply_placement is not None else {}),
             **({"delivery_id": delivery_id} if delivery_id is not None else {}),
             finalize_reactions=finalize_source and not edit_only and kind not in {"admission", "progress", "periodic_report"} and not (kind == "terminal" and record.get("commission_resources")),
@@ -925,6 +944,15 @@ def attempt(value: Any) -> None:
                                                              if row.get("binding_id") == record["binding_id"]), {}))))
         # A verified intermediate reply is not completion. A final reply with
         # cleanup owed recovers its original attempt; it must never be resent.
+        if image_attachments and result.get("reply_verified") is True and result.get("ok") is True:
+            from .post_delivery import attachment_reply
+            image_text, image_attempt = attachment_reply(project=self.runtime_root, config_path=config,
+                attempt=phase_state["attempt"], text=text)
+            verified = verify_chat_images(profile=record["profile"], message_id=image_attempt["message_ref"],
+                text=image_text, attachments=image_attachments, keys=image_keys, private_transport=self,
+                before_read=image_authorized)
+            result.update(verified)
+            phase_state["images_verified"] = verified.get("reply_verified") is True
         phase_state["verified"] = result.get("reply_verified") is True and result.get("ok") is True
         phase_state["blocker"] = result.get("blocker")
         record["deliveries"][phase] = phase_state
@@ -1218,7 +1246,11 @@ def _reconcile_web_projection(self, path: Path) -> int:
                 pending = next((message for message in messages
                     if not row["deliveries"].get("web." + message["message_id"], {}).get("verified")), None)
                 if pending is None:
-                    row["status"] = "partial" if any(message.get("attachments") for message in messages) else "delivered"
+                    # Older verified placeholders retain their exact attempts.
+                    # They cannot be relabelled as media delivery or resent.
+                    row["status"] = "partial" if any(message.get("attachments") and not
+                        row["deliveries"].get("web." + message["message_id"], {}).get("images_verified")
+                        for message in messages) else "delivered"
                     _atomic_write_json(path, row)
                     return 0
                 def authorized() -> bool:
@@ -1228,14 +1260,13 @@ def authorized() -> bool:
                         return False
                 who = "你" if pending["role"] == "user" else "LoopX"
                 text = f"**{who}（App）**\n\n{pending['text']}"
-                if pending.get("attachments"):
-                    text += "\n\n（此条 App 附件尚未同步，请在原 App 查看。）"
                 delivery_id = hashlib.sha256(json.dumps(
                     [row["binding_id"], row["session_id"], pending["message_id"]],
                     separators=(",", ":")).encode()).hexdigest()
                 delivered = self._deliver(path, row, "web." + pending["message_id"], text,
                     inbox=lambda: self._inbox(row), authorization_check=authorized,
-                    reply_placement="chat_root", delivery_id=delivery_id, finalize_source=False)
+                    reply_placement="chat_root", delivery_id=delivery_id, finalize_source=False,
+                    image_attachments=pending.get("attachments"))
                 if delivered:
                     row.pop("blocker", None)
                 _atomic_write_json(path, row)
diff --git a/loopx/extensions/lark/return_files.py b/loopx/extensions/lark/return_files.py
index fc640b8ad7..bf66b0714c 100644
--- a/loopx/extensions/lark/return_files.py
+++ b/loopx/extensions/lark/return_files.py
@@ -1,8 +1,9 @@
-"""Lark resource IO for explicit, snapshotted bound-owner result files."""
+"""Lark resource IO for snapshotted result files and canonical App images."""
 
 from __future__ import annotations
 
 import hashlib
+import base64
 import json
 import re
 import subprocess
@@ -11,7 +12,8 @@
 
 from .inbox_reply import _default_runner, _json_object, _message
 from .outbound import normalize_lark_outbound_text
-from .presentation.markdown_post import lark_markdown_readback_attachment_keys
+from .presentation.markdown_post import lark_markdown_readback_attachment_keys, lark_markdown_readback_matches
+from ...chat_attachments import normalize_chat_image_attachments
 from ...control_plane.collaboration.inbox import _hash, _read, _root, _write
 from ...control_plane.collaboration.result_files import read_result_file
 from ...file_lock import exclusive_file_lock
@@ -33,40 +35,71 @@ def uploaded_result_files(*, root, profile, provider_ref, attachments, runner=No
     Cached keys are bound to App identity, content and display name. Recovery
     never uploads replacement resources to reconstruct a sent-message intent.
     """
+    return _uploaded_resources(root=root, profile=profile, provider_ref=provider_ref, attachments=attachments,
+        cache_dir="lark-result-files", resource="files", key_field="file_key", key_prefix="file_",
+        read_bytes=lambda attachment: read_result_file(root, attachment),
+        upload_data=lambda attachment: {"file_type": "stream", "file_name": attachment["name"]},
+        runner=runner, private_transport=private_transport, before_upload=before_upload, verify_only=verify_only)
+
+
+def _chat_images(attachments):
+    images = normalize_chat_image_attachments(attachments)
+    for image in images:
+        image["raw"] = base64.b64decode(image.pop("data_url").split(",", 1)[1], validate=True)
+        image["sha256"] = hashlib.sha256(image["raw"]).hexdigest()
+    return images
+
+
+def uploaded_chat_images(*, root, profile, provider_ref, attachments, private_transport,
+                         before_upload, verify_only=False):
+    """Reuse resource IO; never regenerate an unknown upload or sent intent."""
+    return _uploaded_resources(root=root, profile=profile, provider_ref=provider_ref,
+        attachments=_chat_images(attachments), cache_dir="lark-chat-images", resource="images",
+        key_field="image_key", key_prefix="img_", read_bytes=lambda image: image["raw"],
+        upload_data=lambda _image: {"image_type": "message"}, runner=None,
+        private_transport=private_transport, before_upload=before_upload, verify_only=verify_only,
+        fence_upload=True)
+
+
+def _uploaded_resources(*, root, profile, provider_ref, attachments, cache_dir, resource,
+                        key_field, key_prefix, read_bytes, upload_data, runner, private_transport,
+                        before_upload, verify_only, fence_upload=False):
     keys = []
     for attachment in attachments:
         identity = {"profile": profile, "provider_ref": provider_ref,
                     "sha256": attachment["sha256"], "name": attachment["name"], "size": attachment["size"]}
-        cache = _root(root) / "lark-result-files" / (_hash(identity) + ".json")
+        cache = _root(root) / cache_dir / (_hash(identity) + ".json")
         with exclusive_file_lock(cache):
             saved = _read(cache) if cache.exists() else None
             if saved is not None:
-                if saved.get("identity") != identity or not re.fullmatch(r"file_[A-Za-z0-9_-]{1,240}", str(saved.get("file_key", ""))):
-                    raise ValueError("result file upload identity conflict")
-                keys.append(saved["file_key"])
+                if saved.get("identity") != identity or not re.fullmatch(key_prefix + r"[A-Za-z0-9_-]{1,240}", str(saved.get(key_field, ""))):
+                    raise ValueError("resource upload identity conflict or acknowledgement unavailable")
+                keys.append(saved[key_field])
                 continue
             if verify_only:
                 raise ValueError("sent result file resource record unavailable")
-            raw = read_result_file(root, attachment)
+            raw = read_bytes(attachment)
             if not before_upload():
                 raise ValueError("result file return authority revoked")
             with TemporaryDirectory(prefix="loopx-lark-result-") as temporary:
                 cwd = Path(temporary)
                 (cwd / "resource").write_bytes(raw)
-                args = ["lark-cli", "--profile", profile, "im", "files", "create",
-                        "--data", json.dumps({"file_type": "stream", "file_name": attachment["name"]}),
+                args = ["lark-cli", "--profile", profile, "im", resource, "create",
+                        "--data", json.dumps(upload_data(attachment)),
                         "--file", "./resource", "--as", "bot", "--format", "json"]
                 preview = _call([*args, "--dry-run"], cwd, runner=runner, private_transport=private_transport)
                 if preview.get("returncode") != 0:
                     raise ValueError("result file upload preflight unavailable")
                 if not before_upload():
                     raise ValueError("result file return authority revoked")
+                if fence_upload:
+                    _write(cache, {"identity": identity, "upload_started": True})
                 uploaded = _call(args, cwd, runner=runner, private_transport=private_transport)
                 payload = _json_object(uploaded.get("stdout"))
-                key = (payload.get("data") or {}).get("file_key")
-                if uploaded.get("returncode") != 0 or not re.fullmatch(r"file_[A-Za-z0-9_-]{1,240}", str(key or "")):
+                key = (payload.get("data") or {}).get(key_field)
+                if uploaded.get("returncode") != 0 or payload.get("ok") is False or not re.fullmatch(key_prefix + r"[A-Za-z0-9_-]{1,240}", str(key or "")):
                     raise ValueError("result file upload unavailable")
-            _write(cache, {"identity": identity, "file_key": key})
+            _write(cache, {"identity": identity, key_field: key})
             keys.append(key)
     return tuple(keys)
 
@@ -86,22 +119,50 @@ def verify_result_files(*, profile, message_id, text, attachments, keys, runner=
             attachment_keys=keys, attachment_names=tuple(attachment["name"] for attachment in attachments))
         if resources is None:
             return {"reply_verified": False, "verification_performed": True, "blocker": "provider_delivery_mismatch"}
-        for index, (attachment, key) in enumerate(zip(attachments, resources, strict=True)):
-            # A suffix keeps lark-cli from inferring one from remote metadata.
-            # The fixed local name still belongs to this temporary directory.
-            output = f"./resource-{index}.bin"
-            result = _call(["lark-cli", "--profile", profile, "im", "+messages-resources-download",
-                           "--message-id", message_id, "--file-key", key, "--type", "file", "--output", output,
-                           "--as", "bot", "--format", "json"], cwd, runner=runner, private_transport=private_transport)
-            if result.get("returncode") != 0:
-                return {"reply_verified": False, "verification_performed": False, "blocker": "provider_verification_unavailable"}
-            path = cwd / output
-            if path.is_symlink() or not path.is_file() or path.stat().st_size != attachment["size"]:
-                return {"reply_verified": False, "verification_performed": True, "blocker": "provider_delivery_mismatch"}
-            # Reuse the host's bounded, regular-file reader. No remote filename
-            # or provider-returned destination can choose a filesystem address.
-            from ...control_plane.collaboration.result_files import _read_regular
-
-            if hashlib.sha256(_read_regular(path)).hexdigest() != attachment["sha256"]:
-                return {"reply_verified": False, "verification_performed": True, "blocker": "provider_delivery_mismatch"}
+        return _verify_resource_bytes(cwd=cwd, profile=profile, message_id=message_id, attachments=attachments,
+            keys=resources, resource_type="file", runner=runner, private_transport=private_transport)
+
+
+def verify_chat_images(*, profile, message_id, text, attachments, keys, private_transport, before_read):
+    """Require the exact ordered post and independently downloaded image bytes."""
+    with TemporaryDirectory(prefix="loopx-lark-image-readback-") as temporary:
+        cwd = Path(temporary).resolve()
+        if not before_read():
+            return {"reply_verified": False, "verification_performed": False, "blocker": "provider_verification_unavailable"}
+        result = _call(["lark-cli", "--profile", profile, "im", "+messages-mget",
+                        "--message-ids", message_id, "--as", "bot", "--no-reactions", "--format", "json"],
+                       cwd, runner=None, private_transport=private_transport)
+        if result.get("returncode") != 0:
+            return {"reply_verified": False, "verification_performed": False, "blocker": "provider_verification_unavailable"}
+        message = _message(_json_object(result.get("stdout")), message_id)
+        if not lark_markdown_readback_matches(text=normalize_lark_outbound_text(text, limit=None, preserve_format=True),
+                                             message=message or {}, image_keys=keys):
+            return {"reply_verified": False, "verification_performed": True, "blocker": "provider_delivery_mismatch"}
+        return _verify_resource_bytes(cwd=cwd, profile=profile, message_id=message_id,
+            attachments=_chat_images(attachments), keys=keys, resource_type="image", runner=None,
+            private_transport=private_transport, before_read=before_read)
+
+
+def _verify_resource_bytes(*, cwd, profile, message_id, attachments, keys, resource_type, runner,
+                           private_transport, before_read=lambda: True):
+    for index, (attachment, key) in enumerate(zip(attachments, keys, strict=True)):
+        # A suffix keeps lark-cli from inferring one from remote metadata.
+        # The fixed local name still belongs to this temporary directory.
+        output = f"./resource-{index}.bin"
+        if not before_read():
+            return {"reply_verified": False, "verification_performed": False, "blocker": "provider_verification_unavailable"}
+        result = _call(["lark-cli", "--profile", profile, "im", "+messages-resources-download",
+                       "--message-id", message_id, "--file-key", key, "--type", resource_type, "--output", output,
+                       "--as", "bot", "--format", "json"], cwd, runner=runner, private_transport=private_transport)
+        if result.get("returncode") != 0:
+            return {"reply_verified": False, "verification_performed": False, "blocker": "provider_verification_unavailable"}
+        path = cwd / output
+        if path.is_symlink() or not path.is_file() or path.stat().st_size != attachment["size"]:
+            return {"reply_verified": False, "verification_performed": True, "blocker": "provider_delivery_mismatch"}
+        # Reuse the host's bounded, regular-file reader. No remote filename
+        # or provider-returned destination can choose a filesystem address.
+        from ...control_plane.collaboration.result_files import _read_regular
+
+        if hashlib.sha256(_read_regular(path)).hexdigest() != attachment["sha256"]:
+            return {"reply_verified": False, "verification_performed": True, "blocker": "provider_delivery_mismatch"}
     return {"reply_verified": True, "verification_performed": True}
diff --git a/tests/test_lark_app_image_projection.py b/tests/test_lark_app_image_projection.py
new file mode 100644
index 0000000000..b5e46cae86
--- /dev/null
+++ b/tests/test_lark_app_image_projection.py
@@ -0,0 +1,197 @@
+"""App image projection uses the original Core input and provider outbox."""
+import json
+from pathlib import Path
+
+import pytest
+from test_chat_ordinary_project import ordinary  # noqa: F401
+from test_chat_image_attachments import PNG_BYTES, PNG_DATA_URL
+from test_lark_private_conversations import connect, _drain_app_projection
+
+from loopx.extensions.lark.private_conversations import LarkPrivateConversations
+
+
+class ImageIO:
+    def __init__(self, provider):
+        self.provider = provider
+        self.resources = {}
+        self.uploads = 0
+        self.downloads = 0
+        self.corrupt_download = False
+        self.unknown_upload = False
+        self.after_upload = None
+
+    def __call__(self, args, cwd=None, timeout=None):
+        if "images" in args and "create" in args:
+            self.provider.calls.append(list(args))
+            assert json.loads(args[args.index("--data") + 1]) == {"image_type": "message"}
+            assert args[args.index("--as") + 1] == "bot"
+            if "--dry-run" in args:
+                return {"returncode": 0, "stdout": '{"ok":true}'}
+            raw = (Path(cwd) / args[args.index("--file") + 1]).read_bytes()
+            self.uploads += 1
+            key = f"img_uploaded_{self.uploads}"
+            self.resources[key] = raw
+            if self.after_upload:
+                self.after_upload()
+            return {"returncode": 0, "stdout": json.dumps({"ok": True,
+                "data": {} if self.unknown_upload else {"image_key": key}})}
+        if "+messages-resources-download" in args:
+            self.provider.calls.append(list(args))
+            assert args[args.index("--type") + 1] == "image"
+            key = args[args.index("--file-key") + 1]
+            message = self.provider.messages[args[args.index("--message-id") + 1]]
+            assert key in message["body"]["content"]
+            raw = b"corrupt" if self.corrupt_download else self.resources[key]
+            (Path(cwd) / args[args.index("--output") + 1]).write_bytes(raw)
+            self.downloads += 1
+            return {"returncode": 0, "stdout": '{"ok":true}'}
+        return self.provider(args, cwd, timeout)
+
+
+def _journey(fixture, *, caption="Inspect this diagram", count=1):
+    store, runtime, provider, transport = connect(fixture)
+    io = ImageIO(provider)
+    transport.runner = io
+    source = provider.event("notes-app", "app-image-source", "Original source")
+    transport.admit("notes-app", source)
+    original = transport.core.pending()[0]
+    sid = original["session_id"]
+    assert runtime.wait_for_turn(session_id=sid, turn_id=original["turn_id"], timeout_sec=10)["status"] == "completed"
+    transport.reconcile()
+    before = len(provider.writes)
+    attachments = [{"id": f"app-image-{i}", "name": f"diagram-{i}.png",
+                    "mime_type": "image/png", "size": len(PNG_BYTES), "data_url": PNG_DATA_URL}
+                   for i in range(count)]
+    status, result = fixture[3](f"/api/chat/sessions/{sid}/turns", {
+        "message": caption, "attachments": attachments, "client_turn_id": "app-original-image"})
+    assert status == 202
+    assert runtime.wait_for_turn(session_id=sid, turn_id=result["turn_id"], timeout_sec=10)["status"] == "completed"
+    transport._web_projection_scan_at = 0  # The next scheduled discovery, without a wall-clock sleep.
+    return store, runtime, provider, transport, io, sid, before
+
+
+def _projection(transport):
+    return json.loads(next((transport.root / "web-projections").glob("*.json")).read_text())
+
+
+def _restart(runtime, transport, io):
+    return LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
+                                   runner=io, cli_bin="lark-cli")
+
+
+def test_app_images_reach_same_original_chat_in_order_once_and_survive_restart(ordinary):  # noqa: F811
+    store, runtime, provider, transport, io, sid, before = _journey(ordinary, count=2)
+    requests = ordinary[4].read_text()
+    try:
+        _drain_app_projection(transport, lambda: len(provider.writes) >= before + 2)
+        transport.reconcile()
+        assert io.uploads == 2 and io.downloads == 2
+        messages = [m for ref, m in provider.messages.items() if ref.startswith("om_out_")][before:]
+        rows = json.loads(messages[0]["body"]["content"])["zh_cn"]["content"]
+        assert rows[0] == [{"tag": "md", "text": "**你（App）**\n\nInspect this diagram"}]
+        assert rows[1:] == [[{"tag": "img", "image_key": key}] for key in io.resources]
+        assert messages[0]["chat_id"] == "oc_notes_app" and not messages[0].get("parent_id")
+        assert _projection(transport)["status"] == "delivered"
+        assert any(p.get("images_verified") for p in _projection(transport)["deliveries"].values())
+        written = list(provider.writes)
+        restarted = _restart(runtime, transport, io)
+        for _ in range(3):
+            restarted.reconcile()
+        assert provider.writes == written and io.uploads == 2
+        assert ordinary[4].read_text() == requests and len(store.list_sessions()) == 1
+        assert len([m for m in store.messages(sid) if m.get("origin") == "web"]) == 1
+    finally:
+        runtime.close()
+
+
+def test_image_byte_readback_failure_recovers_same_receipt_without_send_or_upload(ordinary):  # noqa: F811
+    _, runtime, provider, transport, io, _, before = _journey(ordinary)
+    io.corrupt_download = True
+    try:
+        transport.reconcile()
+        assert len(provider.writes) == before + 1
+        assert _projection(transport)["status"] == "pending"
+        sent = list(provider.writes)
+        transport.reconcile()
+        assert provider.writes == sent and io.uploads == 1
+        io.corrupt_download = False
+        restarted = _restart(runtime, transport, io)
+        _drain_app_projection(restarted, lambda: len(provider.writes) == before + 2)
+        restarted.reconcile()
+        assert _projection(restarted)["status"] == "delivered" and io.uploads == 1
+    finally:
+        runtime.close()
+
+
+@pytest.mark.parametrize("effect", ["upload", "send"])
+def test_unknown_media_effect_never_reuploads_or_resends_after_restart(ordinary, effect):  # noqa: F811
+    _, runtime, provider, transport, io, _, before = _journey(ordinary)
+    io.unknown_upload = effect == "upload"
+    provider.lose_next_reply = effect == "send"
+    try:
+        transport.reconcile()
+        assert io.uploads == 1
+        sent = list(provider.writes)
+        assert len(sent) == before + (effect == "send")
+        restarted = _restart(runtime, transport, io)
+        for _ in range(3):
+            restarted.reconcile()
+        assert provider.writes == sent and io.uploads == 1
+        assert _projection(restarted)["status"] == "pending"
+    finally:
+        runtime.close()
+
+
+def test_revoked_image_source_after_upload_cannot_send_or_continue(ordinary):  # noqa: F811
+    _, runtime, provider, transport, io, _, before = _journey(ordinary)
+    binding = transport._binding("notes-app")
+    io.after_upload = lambda: transport.bindings.disconnect(binding["binding_id"],
+        expected_revision=transport.bindings.read()["revision"])
+    try:
+        transport.reconcile()
+        assert io.uploads == 1 and len(provider.writes) == before
+        restarted = _restart(runtime, transport, io)
+        restarted.reconcile()
+        assert io.uploads == 1 and len(provider.writes) == before
+    finally:
+        runtime.close()
+
+
+def test_long_app_caption_puts_images_only_in_last_concrete_post(ordinary):  # noqa: F811
+    caption = "Full caption and evidence.\n" * 2100 + "IMAGE CAPTION END"
+    _, runtime, provider, transport, io, _, before = _journey(ordinary, caption=caption)
+    try:
+        _drain_app_projection(transport, lambda: any("IMAGE CAPTION END" in text for _, text in provider.writes[before:]))
+        transport.reconcile()
+        image_messages = [m for m in provider.messages.values() if '"tag":"img"' in m.get("body", {}).get("content", "")]
+        assert len(image_messages) == 1 and "IMAGE CAPTION END" in image_messages[0]["body"]["content"]
+        assert io.uploads == 1 and io.downloads == 1
+        _drain_app_projection(transport, lambda: _projection(transport)["status"] == "delivered")
+        requests = ordinary[4].read_text()
+        sent = list(provider.writes)
+        _restart(runtime, transport, io).reconcile()
+        assert provider.writes == sent and ordinary[4].read_text() == requests
+    finally:
+        runtime.close()
+
+
+def test_old_verified_placeholder_remains_partial_without_supplemental_send(ordinary):  # noqa: F811
+    store, runtime, provider, transport, io, sid, _ = _journey(ordinary)
+    try:
+        path = transport._web_projection_paths()[0]
+        row = json.loads(path.read_text())
+        message = next(m for m in store.messages(sid) if m.get("origin") == "web")
+        phase = "web." + message["message_id"]
+        # Freeze an actual historical text-only send through the unchanged
+        # sender. Upgrading cannot turn that receipt into image delivery.
+        text = "**你（App）**\n\n" + message["text"] + "\n\n（此条 App 附件尚未同步，请在原 App 查看。）"
+        assert transport._deliver(path, row, phase, text, inbox=lambda: transport._inbox(row),
+            authorization_check=lambda: True, reply_placement="chat_root", delivery_id="legacy-image-placeholder",
+            finalize_source=False)
+        sent_placeholder = provider.writes[-1]
+        restarted = _restart(runtime, transport, io)
+        _drain_app_projection(restarted, lambda: _projection(restarted)["status"] == "partial")
+        assert provider.writes.count(sent_placeholder) == 1 and io.uploads == 0
+        assert _projection(restarted)["deliveries"][phase]["text"] == text
+    finally:
+        runtime.close()
diff --git a/tests/test_lark_post_images.py b/tests/test_lark_post_images.py
new file mode 100644
index 0000000000..9c591d2e14
--- /dev/null
+++ b/tests/test_lark_post_images.py
@@ -0,0 +1,48 @@
+"""Exact post image content survives preview, readback and multipart sizing."""
+import json
+
+import pytest
+
+from loopx.extensions.lark.presentation.markdown_post import (
+    lark_markdown_post_content, lark_markdown_preview_matches, lark_markdown_readback_matches,
+)
+from loopx.extensions.lark.presentation.markdown_parts import post_request_bytes
+
+TEXT = "**Image caption**\n\n```python\nprint('keep indentation')\n```"
+KEYS = ("img_first", "img_second")
+
+
+def test_exact_images_match_raw_post_and_real_cli_rendering():
+    content = lark_markdown_post_content(TEXT, image_keys=KEYS)
+    assert lark_markdown_preview_matches(text=TEXT, payload={"api": [{"body": {
+        "msg_type": "post", "content": content}}]}, image_keys=KEYS)
+    assert lark_markdown_readback_matches(text=TEXT, message={"msg_type": "post", "body": {"content": content}}, image_keys=KEYS)
+    rendered = TEXT + "\n\n![Image](img_first)\n![Image](img_second)"
+    assert lark_markdown_readback_matches(text=TEXT, message={"msg_type": "post", "content": rendered}, image_keys=KEYS)
+    assert not lark_markdown_readback_matches(text=TEXT, message={"msg_type": "post", "content": rendered})
+
+
+@pytest.mark.parametrize("changed", [(), ("img_first",), ("img_second", "img_first"),
+    ("img_other", "img_second"), (*KEYS, "img_extra")])
+def test_readback_rejects_missing_reordered_replaced_or_extra_images(changed):
+    content = lark_markdown_post_content(TEXT, image_keys=changed)
+    for actual in [{"body": {"content": content}},
+                   {"content": TEXT + "\n" + "\n".join(f"![Image]({key})" for key in changed)}]:
+        assert not lark_markdown_readback_matches(text=TEXT, message={"msg_type": "post", **actual}, image_keys=KEYS)
+
+
+def test_hidden_image_node_fields_or_caption_lookalikes_cannot_verify():
+    content = json.loads(lark_markdown_post_content(TEXT, image_keys=KEYS))
+    content["zh_cn"]["content"][1][0]["extra"] = "changed"
+    assert not lark_markdown_readback_matches(text=TEXT,
+        message={"msg_type": "post", "body": {"content": content}}, image_keys=KEYS)
+    assert not lark_markdown_readback_matches(text=TEXT,
+        message={"msg_type": "text", "content": TEXT + "\n![Image](img_first)\n![Image](img_second)"}, image_keys=KEYS)
+
+
+def test_multipart_envelope_budget_keeps_image_rows_for_mapping_and_string_templates():
+    content = json.loads(lark_markdown_post_content(TEXT, image_keys=KEYS))
+    plain = {"msg_type": "post", "content": lark_markdown_post_content(TEXT)}
+    image = {"msg_type": "post", "content": json.dumps(content, ensure_ascii=False, separators=(",", ":"))}
+    assert post_request_bytes(TEXT, image) > post_request_bytes(TEXT, plain)
+    assert post_request_bytes(TEXT, {**image, "content": content}) == post_request_bytes(TEXT, image)

From 7d7e0ff2862d32452f25bf9bceac56611d8d0d91 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 26 Oct 2026 13:01:12 +0800
Subject: [PATCH 2/2] fix(lark): recheck original authority after send
 preflight

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../app-conversation-and-async-inbox-v0.md    |  4 +-
 ...p-conversation-and-async-inbox-v0.zh-CN.md |  4 +-
 .../extensions/lark/private_conversations.py  | 32 ++++----
 tests/test_lark_app_image_projection.py       | 73 ++++++++++++++++++-
 4 files changed, 96 insertions(+), 17 deletions(-)

diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
index 1c537fc7fe..759c1e2c0e 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -267,7 +267,9 @@ relation mirroring, retroactive insertion into Lark chronology or all-channel
 broadcast. Canonical App images reuse the existing resource uploader and post
 sender, retaining caption and ordered inline images in the original private
 chat. Long captions keep images on the final concrete post only. Every upload
-and send rechecks the original source and current Core message/grant; completion
+and send rechecks the original source and current Core message/grant. Sends and
+edits check again after provider preview, before recording the write intent;
+each continued or resumed rich part uses that same boundary. Completion
 also requires independent image-byte readback. Failed verification recovers the
 same message receipt without another upload or send. An upload whose provider
 acknowledgement is unknown, or a started send without a message locator, stays
diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
index 27d1005270..6520954980 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
@@ -189,7 +189,9 @@ Core 重查 Session、当前绑定与受众，provider 核验原来源。消息
 独立保留。本检查点覆盖文本及有界 App 图片投影，不覆盖回复/话题关系镜像、向
 Lark 历史回插旧时间或全渠道广播。Core 中的 App 图片复用既有资源上传与 post
 发送器，正文及有序行内图片留在原私聊；长正文只在最后一个实际分段携带图片。
-每次上传和发送均重新核验原来源、当前 Core 消息与授权，完成还须独立读回图片
+每次上传和发送均重新核验原来源、当前 Core 消息与授权。发送和编辑在 provider
+预检返回后、记录写入意图前再次检查；每个继续或恢复的富文本分段复用该边界。
+完成还须独立读回图片
 字节。验证失败时沿同一消息回执恢复，不再次上传或发送。上传确认未知，或发送
 已开始但缺少消息定位符时，重启后仍保留未确认。以前已发送的附件占位说明保留
 冻结回执及 partial 状态，升级不自动重放。真实 HTTP/Core 配合合成 provider 的
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index 1d5e0a7f5b..e5cf66079f 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -881,20 +881,29 @@ def image_authorized() -> bool:
             return False
         if edit and reply_placement is not None:
             raise ValueError("a new private message cannot replace an existing source reply")
-        if edit:
-            def before_edit(_intent: str) -> dict[str, bool]:
-                phase_state.update(text=text, started=True)
-                record["deliveries"][phase] = phase_state
-                _atomic_write_json(path, record)
-                return {"continue_delivery": True}
 
+        def before_send(_intent: str) -> dict[str, bool]:
+            # Provider preview is external IO. Recheck the original source and
+            # current Core grant afterwards, including every resumed rich part.
+            if authorization_check is not None and not authorization_check():
+                return {"continue_delivery": False}
+            name, verified = self._source_authority(record, record["profile"],
+                record["event"]["chat_id"], record["event"]["message_id"])
+            if not name or not verified or (authorization_check is not None and not authorization_check()):
+                return {"continue_delivery": False}
+            phase_state.update(text=text, started=True)
+            record["deliveries"][phase] = phase_state
+            _atomic_write_json(path, record)
+            return {"continue_delivery": True}
+
+        if edit:
             def edit_attempt(value: Any) -> None:
                 phase_state["attempt"] = dict(value)
                 record["deliveries"][phase] = phase_state
                 _atomic_write_json(path, record)
 
             result = update_lark_inbox_reply(**kwargs, previous_text=edit["text"], attempt=edit["attempt"],
-                before_send=before_edit, delivery_attempt_recorder=edit_attempt,
+                before_send=before_send, delivery_attempt_recorder=edit_attempt,
                 require_previous_text_match=edit_only)
             if not edit_only and result.get("blocker") == "provider_update_edit_limit":
                 # A confirmed edit rejection plus fresh non-matching readback
@@ -920,18 +929,13 @@ def edit_attempt(value: Any) -> None:
                 _atomic_write_json(path, record)
                 return self._deliver(path, record, phase, text, inbox=inbox)
         elif phase_state.get("attempt"):
-            result = verify_lark_inbox_reply(**kwargs, attempt=phase_state["attempt"], resume_parts=True)
+            result = verify_lark_inbox_reply(**kwargs, attempt=phase_state["attempt"], resume_parts=True,
+                                            before_send=before_send)
         elif phase_state.get("started"):
             # A crash between provider write and its receipt is ambiguous. Keep
             # the original source recoverable; never resend blindly.
             return False
         else:
-            def before_send(_intent: str) -> dict[str, bool]:
-                phase_state.update(text=text, started=True)
-                record["deliveries"][phase] = phase_state
-                _atomic_write_json(path, record)
-                return {"continue_delivery": True}
-
             def attempt(value: Any) -> None:
                 phase_state["attempt"] = dict(value)
                 _atomic_write_json(path, record)
diff --git a/tests/test_lark_app_image_projection.py b/tests/test_lark_app_image_projection.py
index b5e46cae86..973e48659d 100644
--- a/tests/test_lark_app_image_projection.py
+++ b/tests/test_lark_app_image_projection.py
@@ -19,6 +19,7 @@ def __init__(self, provider):
         self.corrupt_download = False
         self.unknown_upload = False
         self.after_upload = None
+        self.after_send_preview = None
 
     def __call__(self, args, cwd=None, timeout=None):
         if "images" in args and "create" in args:
@@ -45,7 +46,10 @@ def __call__(self, args, cwd=None, timeout=None):
             (Path(cwd) / args[args.index("--output") + 1]).write_bytes(raw)
             self.downloads += 1
             return {"returncode": 0, "stdout": '{"ok":true}'}
-        return self.provider(args, cwd, timeout)
+        result = self.provider(args, cwd, timeout)
+        if "+messages-send" in args and "--dry-run" in args and self.after_send_preview:
+            self.after_send_preview()
+        return result
 
 
 def _journey(fixture, *, caption="Inspect this diagram", count=1):
@@ -157,6 +161,42 @@ def test_revoked_image_source_after_upload_cannot_send_or_continue(ordinary):  #
         runtime.close()
 
 
+@pytest.mark.parametrize("long_caption,preview_number,allowed_prefix", [
+    (False, 1, 0), (True, 1, 0), (True, 2, 0), (True, 3, 1),
+])
+def test_disconnect_during_send_preview_stops_every_remaining_part(
+        ordinary, long_caption, preview_number, allowed_prefix):  # noqa: F811
+    caption = "Caption and evidence.\n" * 2100 if long_caption else "Inspect this diagram"
+    _, runtime, provider, transport, io, _, before = _journey(ordinary, caption=caption)
+    binding = transport._binding("notes-app")
+    previews = 0
+
+    def revoke():
+        nonlocal previews
+        previews += 1
+        if previews == preview_number:
+            transport.bindings.disconnect(binding["binding_id"],
+                expected_revision=transport.bindings.read()["revision"])
+
+    io.after_send_preview = revoke
+    try:
+        transport.reconcile()
+        assert previews >= preview_number
+        assert io.uploads == 1
+        assert len(provider.writes) == before + allowed_prefix
+        assert not any('"tag":"img"' in m.get("body", {}).get("content", "")
+                       for m in provider.messages.values())
+        row = _projection(transport)
+        assert row["status"] != "delivered"
+        # No provider write was attempted at the revoked boundary. Earlier
+        # confirmed parts stay frozen; restart cannot upload or send the rest.
+        attempts = list(provider.writes)
+        _restart(runtime, transport, io).reconcile()
+        assert provider.writes == attempts and io.uploads == 1
+    finally:
+        runtime.close()
+
+
 def test_long_app_caption_puts_images_only_in_last_concrete_post(ordinary):  # noqa: F811
     caption = "Full caption and evidence.\n" * 2100 + "IMAGE CAPTION END"
     _, runtime, provider, transport, io, _, before = _journey(ordinary, caption=caption)
@@ -175,6 +215,37 @@ def test_long_app_caption_puts_images_only_in_last_concrete_post(ordinary):  # n
         runtime.close()
 
 
+def test_resumed_parts_recheck_disconnect_after_preview_without_replaying_prefix(ordinary):  # noqa: F811
+    _, runtime, provider, transport, io, _, before = _journey(
+        ordinary, caption="Caption and evidence.\n" * 2100)
+    binding = transport._binding("notes-app")
+    provider.verify_replies = False
+    try:
+        transport.reconcile()
+        assert len(provider.writes) == before + 1 and io.uploads == 1
+        assert _projection(transport)["status"] == "pending"
+        provider.verify_replies = True
+        previews = 0
+
+        def revoke():
+            nonlocal previews
+            previews += 1
+            transport.bindings.disconnect(binding["binding_id"],
+                expected_revision=transport.bindings.read()["revision"])
+
+        io.after_send_preview = revoke
+        sent = list(provider.writes)
+        restarted = _restart(runtime, transport, io)
+        restarted.reconcile()
+        assert previews == 1
+        assert provider.writes == sent and io.uploads == 1
+        assert _projection(restarted)["status"] != "delivered"
+        restarted.reconcile()
+        assert provider.writes == sent and io.uploads == 1
+    finally:
+        runtime.close()
+
+
 def test_old_verified_placeholder_remains_partial_without_supplemental_send(ordinary):  # noqa: F811
     store, runtime, provider, transport, io, sid, _ = _journey(ordinary)
     try: