From bb0612086f7d61b04678ebacd9ea0119009c9682 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 09:44:10 +0800 Subject: [PATCH 1/2] fix(chat): show unsupported Lark media in the original App history Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 8 + ...p-conversation-and-async-inbox-v0.zh-CN.md | 5 + loopx/canary/module_metric_baseline.json | 4 +- .../capabilities/manager_context/roundtrip.py | 7 +- loopx/chat_server.py | 25 ++- .../extensions/lark/private_conversations.py | 159 +++++++++++++++++- tests/test_lark_private_conversations.py | 68 ++++++++ 7 files changed, 266 insertions(+), 10 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 1ab737b14a..96285a82b6 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/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: 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 a84dadcc21..32378f6433 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 @@ -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) diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index 3352c6fa3c..a66f4c7f1f 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/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": 2181 }, "loopx/heartbeat_prompt.py": { "any_count": 17, diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index 41db7b6d09..f581566abc 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/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) diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 50929860f4..59aa8e4312 100644 --- a/loopx/chat_server.py +++ b/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. diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 8224ad7f89..07c928f967 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/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,159 @@ # 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]: + def locate(_path: Path, row: dict[str, Any]) -> dict[str, Any] | None: + # An unreadable journal must not seed an unchanged-file observation. + if row.get("schema_version") != "lark_private_chat_delivery_v0": + 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_json, 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 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): diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 6f6172db56..4f5a25f97c 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -4993,6 +4993,74 @@ def test_disconnect_failure_returns_original_retry_evidence_without_manual_statu runtime.close() +def test_bound_app_history_keeps_original_unsupported_media_without_a_turn(ordinary): # noqa: F811 + store, runtime, provider, transport = connect(ordinary) + profile = "steward-app" + try: + assert transport.admit(profile, provider.event(profile, "original-text", "Original text."))["status"] == "durably_accepted" + request, = transport.core.pending() + sid, tid = request["session_id"], request["turn_id"] + runtime.wait_for_turn(session_id=sid, turn_id=tid, timeout_sec=10) + transport.reconcile() + def snapshot(query=""): + return ordinary[3](f"/api/chat/sessions/{sid}{query}") + + status, first = snapshot("?message_limit=1") + assert status == 200 + old_cursor = first["message_page"]["next_cursor"] + assert old_cursor + original_messages = store.messages(sid) + model_requests = ordinary[-3].read_bytes() + + event = provider.event(profile, "late-unsupported-audio", "private-resource-token", kind="audio") + event["create_time"] = "1700000000123" + provider.messages[event["message_id"]]["create_time"] = event["create_time"] + assert transport.admit(profile, event)["status"] == "command_recorded" + assert transport.reconcile() == 1 + sent = list(provider.writes) + assert "未提交执行" in sent[-1][1] + assert transport.admit(profile, {**event, "event_id": "duplicate-audio"})["status"] == "command_recorded" + transport.reconcile() + assert provider.writes == sent + original_notice = next(row for row in transport.core.pending() if row.get("command") == "unsupported") + assert original_notice["session_id"] == sid and original_notice["turn_id"] is None + + binding_id = original_notice["binding_id"] + listed_status, listed = ordinary[3](f"/api/chat/sessions?conversation_binding_id={binding_id}") + assert listed_status == 200 + listed_revision = listed["sessions"][0]["transcript_revision"] + assert listed_revision != first["session"]["transcript_revision"] + provider_calls = list(provider.calls) + status, page = snapshot("?message_limit=1") + assert status == 200 and page["session"]["transcript_revision"] == listed_revision + notice, = page["messages"] + assert notice["role"] == "error" and notice["origin"] == "lark" and notice["turn_id"] is None + assert notice["created_at"] == "2023-11-14T22:13:20.123000+00:00" + assert "飞书音频" in notice["text"] and "未提交执行" in notice["text"] + assert "private-resource-token" not in json.dumps(page) + assert snapshot("?message_limit=1&message_cursor=" + old_cursor)[0] == 400 + status, tail = snapshot("?message_limit=10&message_cursor=" + page["message_page"]["next_cursor"]) + assert status == 200 and [r["message_id"] for r in tail["messages"]] == [r["message_id"] for r in original_messages] + status, turn = snapshot(f"?message_limit=10&turn_id={tid}") + assert status == 200 and all(row["message_id"] != notice["message_id"] for row in turn["messages"]) + assert provider.calls == provider_calls and store.messages(sid) == original_messages + assert ordinary[-3].read_bytes() == model_requests + + # The same transport source cannot be surfaced under another native + # correlation, nor after the original grant is disconnected. + native_path = transport.core.root / f"{original_notice['request_ref']}.json" + native_path.write_text(json.dumps({**original_notice, "session_id": "other-session"})) + assert notice["message_id"] not in {row["message_id"] for row in snapshot()[1]["messages"]} + native_path.write_text(json.dumps(original_notice)) + assert notice["message_id"] in {row["message_id"] for row in snapshot()[1]["messages"]} + current = transport.bindings.read() + binding = next(row for row in current["bindings"] if row["transport_ref"] == profile) + transport.bindings.disconnect(binding["binding_id"], expected_revision=current["revision"]) + assert notice["message_id"] not in {row["message_id"] for row in snapshot()[1]["messages"]} + finally: + runtime.close() + + def test_source_rejection_unsupported_file_notice_and_ambiguous_reply_readback(ordinary): # noqa: F811 _, runtime, provider, transport = connect(ordinary) try: From 9766cad9f055dfcae09e9ad66e663bc2b1a32ddd Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 10:31:19 +0800 Subject: [PATCH 2/2] fix(chat): isolate unreadable Lark media receipts Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/canary/module_metric_baseline.json | 2 +- .../extensions/lark/private_conversations.py | 16 +++++- tests/test_lark_private_conversations.py | 55 +++++++++++++++++++ 3 files changed, 69 insertions(+), 4 deletions(-) diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index a66f4c7f1f..5723b8f7cd 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -84,7 +84,7 @@ "lines": 2760 }, "loopx/extensions/lark/private_conversations.py": { - "lines": 2181 + "lines": 2191 }, "loopx/heartbeat_prompt.py": { "any_count": 17, diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 07c928f967..0c2c789c72 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -78,9 +78,19 @@ def __init__(self) -> None: 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: - # An unreadable journal must not seed an unchanged-file observation. + 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} @@ -94,14 +104,14 @@ def locate(_path: Path, row: dict[str, Any]) -> dict[str, Any] | None: 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_json, project=locate): + 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 stamp is not None and file_change_fingerprint(root) == stamp: + if complete and stamp is not None and file_change_fingerprint(root) == stamp: self._location_cache = (root, stamp, locations) except OSError: pass diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 4f5a25f97c..810eef37db 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -5061,6 +5061,61 @@ def snapshot(query=""): runtime.close() +def test_bound_app_history_isolates_unreadable_media_receipts_and_recovers(ordinary): # noqa: F811 + store, runtime, provider, transport = connect(ordinary) + request = ordinary[3] + histories = {} + try: + for profile, kind in (("steward-app", "audio"), ("notes-app", "file")): + event = provider.event(profile, f"{profile}-message", f"Original {profile} text") + assert transport.admit(profile, event)["status"] == "durably_accepted" + original = next(row for row in transport.core.pending() if row.get("message") == event["content"]) + sid = original["session_id"] + runtime.wait_for_turn(session_id=sid, turn_id=original["turn_id"], timeout_sec=10) + media = provider.event(profile, f"{profile}-unsupported", "", kind=kind) + assert transport.admit(profile, media)["status"] == "command_recorded" + record = next(row for row in transport.core.pending() + if row.get("session_id") == sid and row.get("command") == "unsupported") + histories[profile] = (sid, record["binding_id"], record["request_ref"]) + + def view(profile): + sid, binding, _ = histories[profile] + snapshot_status, snapshot = request(f"/api/chat/sessions/{sid}?message_limit=10") + index_status, index = request(f"/api/chat/sessions?conversation_binding_id={binding}") + assert (snapshot_status, index_status) == (200, 200) + assert snapshot["session"]["transcript_revision"] == index["sessions"][0]["transcript_revision"] + return snapshot["messages"], snapshot["session"]["transcript_revision"] + + good = view("steward-app") + foreign = view("notes-app") + assert sum("飞书音频" in row["text"] for row in good[0]) == 1 + assert sum("飞书文件" in row["text"] for row in foreign[0]) == 1 + assert not any("飞书文件" in row["text"] for row in good[0]) + bad_path = transport.root / f"{histories['notes-app'][2]}.json" + original_bytes = bad_path.read_bytes() + temporary = bad_path.with_suffix(".corrupt") + temporary.write_bytes(bytes([0xff, 0xfe])) + temporary.replace(bad_path) + + # A broken optional row must neither poison another binding nor erase + # the affected binding's canonical text. Its own notice stays absent. + assert view("steward-app") == good + damaged, damaged_revision = view("notes-app") + assert damaged == store.messages(histories["notes-app"][0]) + assert damaged_revision != foreign[1] + assert bad_path.read_bytes() == bytes([0xff, 0xfe]) + + # Recover even without a directory generation change after the bad + # read. An incomplete scan cannot be retained as complete. + bad_path.write_bytes(original_bytes) + assert view("notes-app") == foreign + assert view("steward-app") == good + assert not provider.writes + assert sum(json.loads(line)["method"] == "turn/start" for line in ordinary[-3].read_text().splitlines()) == 2 + finally: + runtime.close() + + def test_source_rejection_unsupported_file_notice_and_ambiguous_reply_readback(ordinary): # noqa: F811 _, runtime, provider, transport = connect(ordinary) try: