fix(chat): show unsupported Lark media in the original App history

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

@@ -336,6 +336,14 @@ 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 bound App history also displays a read-only notice for an original Lark
+file, audio or video that Core recorded as unsupported. It verifies the exact
+Session, current local owner binding and both original receipts before showing
+the notice. Its source time participates in history pagination; late rows
+invalidate any affected prefix, and a changed notice refreshes the App's
+transcript even when no Turn runs. The notice carries no resource bytes or
+credentials, creates no Turn, and does not resend the original reply.
+
 The implementation decision follows primary-source comparisons, not a claim
 that shared context already guarantees two-way delivery:
 

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

@@ -237,6 +237,11 @@ Lark 历史回插旧时间或全渠道广播。Core 中的 App 图片复用既
 旅程覆盖重复输入、完整图片和长正文、重启、回执缺失/未知、字节不符、撤权及
 App 身份替换；安装态 provider 读回另行验收。
 
+绑定的 App 历史也显示 Core 已明确判为不支持的原飞书文件、音频或视频的只读
+提示。展示前核对原 Session、当前本地本人绑定和两份原回执；原来源时间参与
+历史分页，迟到记录会使受影响的旧游标失效；即使没有新 Turn，提示变化也会触发
+App 刷新。提示不携带资源字节或凭据，不创建 Turn，也不重发原回复。
+
 实现判断来自一手源码比较，不把共享上下文视为双向送达已经成立：
 
 - [OpenClaw 历史增量](https://github.com/openclaw/openclaw/blob/b3ba4d622638ac23f64c0cec415bf5bfe81f5cf5/src/gateway/server-methods/chat-history-delta.ts)

loopx/canary/module_metric_baseline.json

@@ -44,7 +44,7 @@
     "loopx/chat_server.py": {
       "any_count": 25,
       "dict_any_count": 0,
-      "lines": 2046
+      "lines": 2058
     },
     "loopx/chat_store.py": {
       "lines": 2141
@@ -84,7 +84,7 @@
       "lines": 2760
     },
     "loopx/extensions/lark/private_conversations.py": {
-      "lines": 2024
+      "lines": 2191
     },
     "loopx/heartbeat_prompt.py": {
       "any_count": 17,

loopx/capabilities/manager_context/roundtrip.py

@@ -341,7 +341,8 @@ def candidate_paths():
 
 
 def project_chat_session_snapshot(root, store, session_id, *, registry,
-                                  message_limit=None, message_cursor=None, turn_id=None):
+                                  message_limit=None, message_cursor=None, turn_id=None,
+                                  project_history_messages=None):
     """Project return delivery state into one existing Chat snapshot."""
 
     snapshot = (store.session_snapshot(session_id, strict_messages=True) if message_limit is not None
@@ -355,6 +356,10 @@ def project_chat_session_snapshot(root, store, session_id, *, registry,
         # OriginalRequest already displays only this Turn. Select it before
         # paging/projection, retaining every message and the strict journal read.
         snapshot["messages"] = [row for row in snapshot["messages"] if row.get("turn_id") == turn_id]
+    elif project_history_messages is not None:
+        # Extension-owned, presentation-only rows must join the frozen page
+        # prefix. A selected Turn only reads its canonical messages.
+        snapshot["messages"] = project_history_messages(snapshot["messages"])
     from .chat_history import page_chat_messages
     page = (page_chat_messages(session_id, snapshot["messages"], limit=message_limit,
                               cursor=message_cursor, turn_id=turn_id) if message_limit is not None else None)

loopx/chat_server.py

@@ -60,7 +60,8 @@
 from .capabilities.native_chat.model_settings import normalize_agent_reasoning_effort, normalize_agent_model_id
 from .extensions.lark.private_conversation_api import PrivateConversationRequestMixin, PRIVATE_CONVERSATIONS_PATH
 from .extensions.lark.conversation_identity import observe_lark_conversation_identity
-from .extensions.lark.private_conversations import LarkPrivateConversations, project_private_reply_history
+from .extensions.lark.private_conversations import (LarkPrivateConversations, PrivateUnsupportedMediaHistory,
+    project_private_reply_history)
 from .chat_loopx_mode import DelegationWakeService, handle_loopx_request
 from .capabilities.manager_context.roundtrip import (
     ReturnService,
@@ -480,6 +481,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.runtime_identity = release_runtime_identity()
         super().__init__(*args, **kwargs)
         self.completed_todo_pages = CompletedTodoPages()
+        self.private_media_history = PrivateUnsupportedMediaHistory()
 
     def server_close(self) -> None:
         background_api.close_chat_background_services(self)
@@ -935,17 +937,24 @@ def _session_snapshot(self, session_id: str) -> None:
             # execution. Ordinary Chat refresh keeps its existing recovery path.
             if "turn_id" not in query:
                 self.server.runtime_controller.recover_exited_turn(session_id)
+            store = self.server.chat_store
+            session = store.load_session(session_id) or {}
+            media = self.server.private_media_history
+            contexts = getattr(self.server.runtime_controller, "project_contexts", None)
+            bindings = getattr(contexts, "conversation_bindings", None)
+            markers = [] if "turn_id" in query else media.read(store, bindings, session)
             snapshot = cast(Callable[..., dict[str, object]], project_chat_session_snapshot)
             payload = snapshot(
                 self.server.runtime_controller.coordination_runtime_root,
                 self.server.chat_store, session_id,
                 registry=self.server.registry_path, message_limit=message_limit,
                 message_cursor=query.get("message_cursor", [None])[0],
-                turn_id=query.get("turn_id", [None])[0])
-            # The HTTP composition root joins provider history onto the selected page.
-            payload["messages"] = project_private_reply_history(
-                self.server.chat_store, self.server.chat_store.load_session(session_id) or {},
-                payload["messages"])
+                turn_id=query.get("turn_id", [None])[0],
+                project_history_messages=(lambda messages: media.insert(messages, markers)) if markers else None)
+            if markers:
+                payload["session"] = media.summary(payload["session"], markers)
+            # Reply context is a view of rows already selected by the page.
+            payload["messages"] = project_private_reply_history(store, session, payload["messages"])
             self._send_json(payload)
         except KeyError:
             self._send_error("chat session was not found", status=404)
@@ -1021,6 +1030,10 @@ def _list_sessions(self) -> None:
                 if bindings is None:
                     raise ValueError("conversation bindings unavailable")
                 snapshot = bindings.local_session_snapshot(query["conversation_binding_id"][0])
+                media = self.server.private_media_history
+                store = self.server.chat_store
+                snapshot["sessions"] = [media.summary(row, media.read(store, bindings,
+                    store.load_session(row["session_id"]) or {})) for row in snapshot["sessions"]]
             except (ValueError, OSError, KeyError):
                 # A failed provider/store read may contain private diagnostics.
                 # Refuse this selector; never fall back to an unfiltered list.

loopx/extensions/lark/private_conversations.py

@@ -12,11 +12,15 @@
 import time
 from collections.abc import Callable, Mapping, Sequence
 from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone
 from pathlib import Path
+from threading import Lock
 from typing import Any
 
 from ...capabilities.native_chat.external_conversations import ChatExternalConversations, QUOTA_BACKLOG_BATCH_SIZE
-from ...chat_store import TERMINAL_TURN_STATES, _atomic_write_json, _read_json
+from ...capabilities.native_chat.conversation_bindings import ChatConversationBindings
+from ...capabilities.native_chat.file_discovery import FileChatDiscovery, file_change_fingerprint
+from ...chat_store import CHAT_MESSAGE_SCHEMA_VERSION, ChatSessionStore, TERMINAL_TURN_STATES, _atomic_write_json, _read_json
 from ...control_plane.collaboration import conversation_reply_context
 from ...control_plane.effect_runtime import effect_runtime_request_scope
 from ...file_lock import exclusive_file_lock
@@ -43,6 +47,169 @@
 # Reserve edits for a final/closing notice; continue long Turns on a new draft.
 PROGRESS_EDIT_BUDGET = 12
 CHAT_CONTROL_COMMANDS = {"/status": "status", "/help": "help", "/new": "new", "/stop": "stop"}
+_UNSUPPORTED_MEDIA_LABELS = {"audio": "音频", "file": "文件", "media": "视频"}
+
+
+def _unsupported_media_label(row: Mapping[str, Any]) -> str | None:
+    event = row.get("event")
+    kind = event.get("message_type") if isinstance(event, Mapping) else None
+    if (isinstance(kind, str) and row.get("schema_version") == "lark_private_chat_delivery_v0"
+            and row.get("status") in ("command_completed", "delivered")
+            and row.get("response_code") == "unsupported_attachment" and "turn_id" in row and row["turn_id"] is None):
+        return _UNSUPPORTED_MEDIA_LABELS.get(kind)
+    return None
+
+
+def _history_timestamp(value: Any) -> datetime:
+    if not isinstance(value, str):
+        raise ValueError("missing history time")
+    stamp = datetime.fromisoformat(value)
+    if stamp.tzinfo is None:
+        raise ValueError("unqualified history time")
+    return stamp.astimezone(timezone.utc)
+
+
+class PrivateUnsupportedMediaHistory:
+    """Read Lark presentation history without constructing a transport worker."""
+
+    def __init__(self) -> None:
+        self._discovery = FileChatDiscovery()
+        self._location_lock = Lock()
+        self._location_cache: tuple[Path, tuple[int, ...], dict[str, list[Path]]] | None = None
+
+    def _locations(self, root: Path, sid: str) -> list[Path]:
+        complete = True
+
+        def read_receipt(path: Path) -> dict[str, Any]:
+            try:
+                return _read_json(path)
+            except UnicodeDecodeError:
+                return {}
+
+        def locate(_path: Path, row: dict[str, Any]) -> dict[str, Any] | None:
+            nonlocal complete
+            # Keep valid locators, but retry unreadable receipts on the next scan.
+            if row.get("schema_version") != "lark_private_chat_delivery_v0":
+                complete = False
+                return None
+            return {"media_session": row.get("session_id") if _unsupported_media_label(row) else None}
+
+        with self._location_lock:
+            try:
+                stamp = file_change_fingerprint(root)
+            except OSError:
+                return []
+            cached = self._location_cache
+            if stamp is not None and cached is not None and cached[:2] == (root, stamp):
+                return cached[2].get(sid, [])
+            paths = (path for path in root.glob("*.json") if re.fullmatch(r"[a-f0-9]{24}", path.stem))
+            locations: dict[str, list[Path]] = {}
+            for path, view in self._discovery.rows(root, paths, read=read_receipt, project=locate):
+                session_id = view.get("media_session")
+                if isinstance(session_id, str):
+                    locations.setdefault(session_id, []).append(path)
+            # All production journal publications are atomic replacements.
+            # Cache only the locator list, never a selected receipt or grant.
+            try:
+                if complete and stamp is not None and file_change_fingerprint(root) == stamp:
+                    self._location_cache = (root, stamp, locations)
+            except OSError:
+                pass
+            return locations.get(sid, [])
+
+    def read(self, store: ChatSessionStore, bindings: ChatConversationBindings | None,
+             session: dict[str, Any]) -> list[dict[str, Any]]:
+        """Show original, non-executed Lark media without replay or provider IO."""
+        saved = session.get("steward_context") or session.get("project_context") or {}
+        if (bindings is None or saved.get("audience") != "bound_owner"
+                or session.get("session_mode") != "managed_runtime"):
+            return []
+        sid = session["session_id"]
+        try:
+            state = bindings.read()
+            matches = [row for row in state["bindings"] if row["binding_id"] == saved.get("binding_id")]
+        except (KeyError, TypeError, ValueError, OSError):
+            return []
+        if len(matches) != 1:
+            return []
+        binding, = matches
+        prefix = "manager.external.native" if binding.get("context_kind") == "steward" else "project.external"
+        if (binding.get("enabled") is not True or binding.get("audience", "owner") != "owner"
+                or any(not saved.get(k) or binding.get(k) != saved[k]
+                       for k in ("provider_ref", "operator_ref", "project_ref", "grant"))
+                or not saved.get("source_ref")
+                or session.get("channel_id") != f"{prefix}.{binding['binding_id']}.{saved['source_ref']}"):
+            return []
+
+        markers = []
+        for path in self._locations(store.root / "lark-private-deliveries", sid):
+            try:
+                # The locator is only a hint. Join the two exact receipts and
+                # recompute the private principal; don't trust a cached grant.
+                record = _read_json(path)
+                label = _unsupported_media_label(record)
+                event = record["event"]
+                source = lark_private_source(provider_ref=saved["provider_ref"], event=event)
+                native = _read_json(store.root / "external-requests" / path.name)
+                expected = {"request_ref": path.stem, "binding_id": binding["binding_id"],
+                            "session_id": sid, "source": source, "response_code": "unsupported_attachment"}
+                if (not label or record.get("profile") != binding["transport_ref"]
+                        or source["private_human_message"] is not True or source["source_ref"] != saved["source_ref"]
+                        or source["sender_ref"] != saved["operator_ref"]
+                        or identity_ref(saved["provider_ref"], event["message_id"]) != path.stem
+                        or any(row.get(k) != v for row in (record, native) for k, v in expected.items())
+                        or native.get("schema_version") != "loopx_chat_external_request_v0"
+                        or native.get("origin") not in (None, "lark")
+                        or native.get("status") != "command_completed" or native.get("command") != "unsupported"
+                        or "turn_id" not in native or native["turn_id"] is not None):
+                    continue
+                epoch = str(event.get("create_time") or "")
+                stamp = (datetime.fromtimestamp(int(epoch) / 1000, timezone.utc)
+                         if re.fullmatch(r"[0-9]{13}", epoch) else _history_timestamp(native.get("created_at")))
+                mid = f"lark.unsupported.{path.stem}"
+                markers.append({"schema_version": CHAT_MESSAGE_SCHEMA_VERSION,
+                    "message_id": mid, "turn_id": None, "role": "error", "origin": "lark",
+                    "text": f"飞书{label} · 未提交执行。请在原飞书会话查看。",
+                    "created_at": stamp.isoformat(timespec="microseconds")})
+            except (KeyError, ValueError, TypeError, OSError):
+                continue
+
+        try:
+            if not markers or bindings.read() != state:
+                return []
+        except (ValueError, OSError):
+            return []
+        return sorted(markers, key=lambda row: (row["created_at"], row["message_id"]))
+
+    @staticmethod
+    def summary(summary: dict[str, Any], markers: list[dict[str, Any]]) -> dict[str, Any]:
+        """Invalidate an App's quiet history cache even when no Turn was created."""
+        revision = summary.get("transcript_revision")
+        if not revision or not markers:
+            return summary
+        digest = hashlib.sha256(json.dumps(markers, sort_keys=True, ensure_ascii=False,
+                                          separators=(",", ":")).encode("utf-8")).hexdigest()
+        return {**summary, "transcript_revision": f"{revision}.lark.{digest}"}
+
+    @staticmethod
+    def insert(messages: list[dict[str, Any]], markers: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        if not markers:
+            return messages
+        # Keep canonical append order and insert presentation rows by original
+        # time before paging. A late insertion must invalidate the old prefix.
+        ids = {row.get("message_id") for row in messages}
+        pending = [(_history_timestamp(row["created_at"]), row) for row in reversed(markers)
+                   if row["message_id"] not in ids]
+        projected = []
+        for message in messages:
+            try:
+                stamp = _history_timestamp(message.get("created_at"))
+            except ValueError:
+                stamp = None
+            while pending and stamp is not None and pending[-1][0] < stamp:
+                projected.append(pending.pop()[1])
+            projected.append(message)
+        return projected + [row[1] for row in reversed(pending)]
 
 
 class _ControlReadFallback(Exception):