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..759c1e2c0e 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,22 @@ 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. 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
+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..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
@@ -186,10 +186,17 @@ Core 重查 Session、当前绑定与受众，provider 核验原来源。消息
 关闭既有连接即可撤销授权；回滚须保留私有投递 journal。
 
 通过上述 **继续原会话** 或明确选定的 App 管家入口使用该路径；旧普通历史继续
-独立保留。本检查点覆盖文本投影，不覆盖回复/话题关系镜像、附件传输、向 Lark
-历史回插旧时间或全渠道广播。App 附件会产生明确的未同步提示，并保留 partial
-outbox，不声明完整送达。真实 HTTP/Core 配合合成 provider 的旅程覆盖重复输入、
-长正文、重启、回执缺失/未知、撤权及 App 身份替换；安装态 provider 读回另行验收。
+独立保留。本检查点覆盖文本及有界 App 图片投影，不覆盖回复/话题关系镜像、向
+Lark 历史回插旧时间或全渠道广播。Core 中的 App 图片复用既有资源上传与 post
+发送器，正文及有序行内图片留在原私聊；长正文只在最后一个实际分段携带图片。
+每次上传和发送均重新核验原来源、当前 Core 消息与授权。发送和编辑在 provider
+预检返回后、记录写入意图前再次检查；每个继续或恢复的富文本分段复用该边界。
+完成还须独立读回图片
+字节。验证失败时沿同一消息回执恢复，不再次上传或发送。上传确认未知，或发送
+已开始但缺少消息定位符时，重启后仍保留未确认。以前已发送的附件占位说明保留
+冻结回执及 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..e5cf66079f 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")),
@@ -862,20 +881,29 @@ def _deliver(self, path: Path, record: dict[str, Any], phase: str, text: str, *,
             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
@@ -901,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)
@@ -925,6 +948,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 +1250,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 +1264,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..973e48659d
--- /dev/null
+++ b/tests/test_lark_app_image_projection.py
@@ -0,0 +1,268 @@
+"""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
+        self.after_send_preview = 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}'}
+        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):
+    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()
+
+
+@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)
+    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_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:
+        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)