From 8e47f04a1767205452b1a906d55d842a6d8f4899 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 25 Oct 2026 20:31:27 +0800 Subject: [PATCH] fix(lark): deliver App messages from original private conversations Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 40 +++++ ...p-conversation-and-async-inbox-v0.zh-CN.md | 30 ++++ loopx/extensions/lark/inbox_reply.py | 13 +- loopx/extensions/lark/post_delivery.py | 9 ++ .../extensions/lark/private_conversations.py | 134 ++++++++++++++++- tests/test_lark_private_conversations.py | 140 ++++++++++++++++++ 6 files changed, 361 insertions(+), 5 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 9cb3b6d31b..70f1bea4cc 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -154,6 +154,46 @@ Packaged browser evidence qualifies the local UI path separately. Live Lark return delivery, installed service adoption, split-root composition and Goal-less project peer handoff remain open; this checkpoint does not close them. +The existing Lark reply workers now also project committed App-origin user and +answer messages from that exact owner-private Session. Each provider outbox +retains the canonical message identity, frozen text and delivery attempt. The +original Lark input supplies address provenance only: projection never readmits +it or starts model work. Before every write and receipt recovery, Core rechecks +the Session, current binding and audience, and the provider verifies the original +source. Posts identify the App speaker, stay in canonical append order within +that outbox and retain complete long text through the existing rich-post parts. +Distinct canonical messages and parts get distinct provider idempotency keys, +including when their text is identical; callers without that identity retain +their existing receipt digest. Known attempts are read back before continuation; +a started write without a provider locator stays unconfirmed after restart. +Revocation or changed identity prevents further delivery. Disable the existing +connection to revoke it; rollback must preserve its private delivery journals. + +Use **Continue original conversation** above to reach this path. Ordinary App +steward history is still a separate entry; it is not silently merged or retargeted. +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. + +The implementation decision follows primary-source comparisons, not a claim +that shared context already guarantees two-way delivery: + +- [OpenClaw history delta](https://github.com/openclaw/openclaw/blob/b3ba4d622638ac23f64c0cec415bf5bfe81f5cf5/src/gateway/server-methods/chat-history-delta.ts) + resets incomplete/scoped projections; its + [send retry owner](https://github.com/openclaw/openclaw/blob/b3ba4d622638ac23f64c0cec415bf5bfe81f5cf5/src/gateway/server-methods/chat-send-retry.ts) + reconciles already-started execution. These inform recovery, not a Lark bridge. +- [nanobot unified session keys](https://github.com/HKUDS/nanobot/blob/9e9e22d3319678178dba05a76f0ea61064b78ee8/nanobot/session/keys.py) + and [last-destination routing](https://github.com/HKUDS/nanobot/blob/9e9e22d3319678178dba05a76f0ea61064b78ee8/nanobot/agent/loop.py) + share context; a last destination does not replicate a transcript to every surface. +- [mautrix message mappings](https://github.com/mautrix/go/blob/a4237e5b9f8f0e9b35c81d5310615eb70c202cea/bridgev2/database/message.go) + and [portal echo handling](https://github.com/mautrix/go/blob/a4237e5b9f8f0e9b35c81d5310615eb70c202cea/bridgev2/portal.go) + motivate stable transaction/part identities and echo suppression. A Matrix + homeserver/bridge would add separate infrastructure and identity ownership. + LoopX borrows the design, not MPL-2.0 code; the other two snapshots are MIT. + Synthetic product previews: [desktop](../../assets/personal-workspace/private-project-conversations.png), [narrow](../../assets/personal-workspace/private-project-conversations-narrow.png), [revoked workspace](../../assets/personal-workspace/private-project-workspace-revoked.png). 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 c5b9811f96..58196fedea 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 @@ -109,6 +109,36 @@ Web 来源，通过已有 Core 队列及新鲜来源准入。连接列表独立 真实 Lark 结果回传、安装态服务采用、分离运行根组合及无 Goal 项目的 peer 交办 仍未关闭，不由此检查点代替。 +现有 Lark 回复 worker 现在也投递该本人私聊确切 Session 中已提交的 App 用户消息和 +回答。每份 provider outbox 保留 canonical 消息身份、冻结正文和发送尝试。原 Lark +输入只提供地址来源证据；投影不会重新受理它或启动模型工作。每次发送和恢复回执前， +Core 重查 Session、当前绑定与受众，provider 核验原来源。消息标明 App 发言者，在 +该 outbox 内按 canonical 追加顺序发送，长正文沿用既有富文本分段，保留完整内容。 +即使正文相同，不同 canonical 消息及分段也使用不同 provider 幂等键；未提供该身份 +的既有调用者保持原回执 digest。恢复先读回已知发送，再继续未发送部分；已经开始但 +没有 provider locator 的写入，重启后仍保留未确认。撤权或身份变化阻止后续投递。 +关闭既有连接即可撤销授权；回滚须保留私有投递 journal。 + +通过上述 **继续原会话** 使用该路径。普通 App 管家历史仍是独立入口，不会被静默 +合并或重定向。本检查点覆盖文本投影，不覆盖回复/话题关系镜像、附件传输、向 Lark +历史回插旧时间或全渠道广播。App 附件会产生明确的未同步提示，并保留 partial +outbox，不声明完整送达。真实 HTTP/Core 配合合成 provider 的旅程覆盖重复输入、 +长正文、重启、回执缺失/未知、撤权及 App 身份替换；安装态 provider 读回另行验收。 + +实现判断来自一手源码比较，不把共享上下文视为双向送达已经成立： + +- [OpenClaw 历史增量](https://github.com/openclaw/openclaw/blob/b3ba4d622638ac23f64c0cec415bf5bfe81f5cf5/src/gateway/server-methods/chat-history-delta.ts) + 对不完整/受限投影要求重置；其 + [发送重试 owner](https://github.com/openclaw/openclaw/blob/b3ba4d622638ac23f64c0cec415bf5bfe81f5cf5/src/gateway/server-methods/chat-send-retry.ts) + 对已经开始的执行先对账。这些机制用于恢复设计，不代表已有 Lark 桥。 +- [nanobot 统一会话键](https://github.com/HKUDS/nanobot/blob/9e9e22d3319678178dba05a76f0ea61064b78ee8/nanobot/session/keys.py) + 和[最后目的地路由](https://github.com/HKUDS/nanobot/blob/9e9e22d3319678178dba05a76f0ea61064b78ee8/nanobot/agent/loop.py) + 可以共享上下文，但最后目的地不等于向每个界面复制完整记录。 +- [mautrix 消息映射](https://github.com/mautrix/go/blob/a4237e5b9f8f0e9b35c81d5310615eb70c202cea/bridgev2/database/message.go) + 及 [portal 回声处理](https://github.com/mautrix/go/blob/a4237e5b9f8f0e9b35c81d5310615eb70c202cea/bridgev2/portal.go) + 支持稳定事务/分段身份和回声抑制的设计判断。采用 Matrix homeserver/桥还会增加 + 独立基础设施与身份 owner；LoopX 借鉴设计，不复制 MPL-2.0 代码，另两个快照为 MIT。 + ![合成私聊工作区设置](../../assets/personal-workspace/private-project-conversations.png) ![窄屏私聊设置](../../assets/personal-workspace/private-project-conversations-narrow.png) ![工作区撤权读回](../../assets/personal-workspace/private-project-workspace-revoked.png) diff --git a/loopx/extensions/lark/inbox_reply.py b/loopx/extensions/lark/inbox_reply.py index 189507e762..129baa08fa 100644 --- a/loopx/extensions/lark/inbox_reply.py +++ b/loopx/extensions/lark/inbox_reply.py @@ -318,6 +318,7 @@ def _deliver_lark_inbox_outbound( text: str, content_format: str = "text", reply_placement: str | None = None, + delivery_id: str | None = None, execute: bool = False, provider_preflight: bool = False, runner: CommandRunner = _default_runner, @@ -412,12 +413,16 @@ def _deliver_lark_inbox_outbound( ) else "source_thread" ) + if delivery_id is not None and (not isinstance(delivery_id, str) + or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}", delivery_id)): + raise ValueError("invalid native delivery identity") digest = hashlib.sha256( json.dumps( { "message_id": source_message_id, "placement": placement, "text": reply_text, + **({"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 {}), }, @@ -587,6 +592,7 @@ def verify_source() -> tuple[bool, bool]: "source_membership_verifier": source_membership_verifier, "source_authority_verifier": source_authority_verifier, **({"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}) guidance = None @@ -810,6 +816,7 @@ def reply_lark_event_inbox( text: str, content_format: str = "text", reply_placement: str | None = None, + delivery_id: str | None = None, execute: bool = False, provider_preflight: bool = False, runner: CommandRunner = _default_runner, @@ -853,6 +860,7 @@ def reply_lark_event_inbox( text=text, content_format=content_format, reply_placement=reply_placement, + delivery_id=delivery_id, execute=execute, provider_preflight=provider_preflight, runner=runner, @@ -878,6 +886,7 @@ def verify_lark_inbox_reply( config_path: str | Path, message_id: str, reply_placement: str | None = None, + delivery_id: str | None = None, text: str, attempt: Mapping[str, Any], runner: CommandRunner = _default_runner, @@ -922,7 +931,7 @@ def verify_lark_inbox_reply( 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, - reply_placement=reply_placement) + 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), attempt=attempt, resume=resume_parts, arguments={"project": project, @@ -930,6 +939,7 @@ def verify_lark_inbox_reply( "source_membership_verifier": source_membership_verifier, "source_authority_verifier": source_authority_verifier, **({"reply_placement": reply_placement} if reply_placement is not None else {}), + **({"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}) @@ -947,6 +957,7 @@ def verify_lark_inbox_reply( text=candidate_text, content_format=content_format, reply_placement=reply_placement, + delivery_id=delivery_id, execute=False, runner=runner, attachment_keys=attachment_keys, diff --git a/loopx/extensions/lark/post_delivery.py b/loopx/extensions/lark/post_delivery.py index c4d0b67691..3799aa3540 100644 --- a/loopx/extensions/lark/post_delivery.py +++ b/loopx/extensions/lark/post_delivery.py @@ -7,6 +7,7 @@ from __future__ import annotations import json +import hashlib import re from pathlib import Path from typing import Any @@ -19,6 +20,12 @@ POST_PARTS_LOCATOR = re.compile(r"^lark-parts:([a-f0-9]{64})$") +def _part_identity(arguments, index): + identity = arguments.get("delivery_id") + return ({"delivery_id": hashlib.sha256(f"{identity}:part:{index}".encode()).hexdigest()} + if identity is not None else {}) + + def _path(project, config_path, intent_digest): config = load_lark_event_inbox_config(project=project, config_path=config_path) return ( @@ -124,6 +131,7 @@ def options(index): "source_authority_verifier": arguments.get("source_authority_verifier"), **({"reply_placement": arguments["reply_placement"]} if arguments.get("reply_placement") is not None else {}), + **_part_identity(arguments, index), "before_send": (lambda _part_intent: outer_hook(intent_digest)) if outer_hook else None, @@ -249,6 +257,7 @@ def verify_post_parts(*, receipt, intent_digest, attempt, arguments, resume=Fals source_authority_verifier=arguments.get("source_authority_verifier"), **({"reply_placement": arguments["reply_placement"]} if arguments.get("reply_placement") is not None else {}), + **_part_identity(arguments, index), finalize_reactions=False, attachment_keys=arguments.get("attachment_keys", ()) if final else (), attachment_names=arguments.get("attachment_names", ()) if final else (), diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 4fc3363371..1929d80ae7 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -53,6 +53,8 @@ def __init__(self, *, controller: Any, runtime_root: Path, runner: Any, cli_bin: self.reaction_feedback = reaction_feedback self.root = controller.store.root / "lark-private-deliveries" self._context_recovery_probe_at: dict[Path, float] = {} + self._web_projection_scan_at = 0.0 + self._web_projection_snapshots: dict[str, tuple[int, int]] = {} from .owner_attention import LarkOwnerAttention self.owner_attention = LarkOwnerAttention(self) from .private_ingress import PrivateIngressObserver @@ -97,7 +99,7 @@ def profiles(self) -> dict[str, dict[str, str]]: def health(self) -> dict[str, dict[str, int]]: health: dict[str, dict[str, int]] = {} - for path in self.root.glob("*.json"): + for path in [*self.root.glob("*.json"), *self.root.glob("web-projections/*.json")]: row = _read_json(path) entry = health.setdefault(row["binding_id"], {"pending_count": 0, "recovery_count": 0}) if row["status"] != "delivered": @@ -815,7 +817,8 @@ 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) -> bool: + reply_placement: str | None = None, delivery_id: str | None = None, + finalize_source: bool = True) -> bool: if not text: return False if reply_placement is not None and (record["event"].get("chat_type") != "p2p" @@ -841,7 +844,8 @@ def _deliver(self, path: Path, record: dict[str, Any], phase: str, text: str, *, kwargs = dict(project=self.runtime_root, config_path=config, message_id=record["event"]["message_id"], text=text, runner=self._reply_runner, **({"reply_placement": reply_placement} if reply_placement is not None else {}), - finalize_reactions=not edit_only and kind not in {"admission", "progress", "periodic_report"} and not (kind == "terminal" and record.get("commission_resources")), + **({"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")), source_authority_verifier=lambda profile, chat, message: ( self._source_authority(record, profile, chat, message) if authorization_check is None or authorization_check() else (None, False))) @@ -1119,7 +1123,127 @@ def pending_delivery_paths(self) -> list[Path]: self._context_recovery_probe_at[path] = time.monotonic() + 60 if legacy_recovery_repairs(row) or self.core.context_recovery_candidate(row["request_ref"]): paths.append(path) - return paths + return [*paths, *self._web_projection_paths()] + + def _web_projection_paths(self) -> list[Path]: + """Discover committed App messages, using the existing reply workers. + + These files are provider outboxes, not input requests or model work. + The original Core transcript and grant are re-read before each effect. + File observations only avoid rereading unchanged local Chat JSONL; they + never authorize a send, acknowledge input or advance a delivery cursor. + """ + root = self.root / "web-projections" + if time.monotonic() >= self._web_projection_scan_at: + self._web_projection_scan_at = time.monotonic() + 5 + enabled = {row["binding_id"] for row in self.bindings.read()["bindings"] + if row.get("enabled") is True and row.get("audience", "owner") == "owner"} + sources: dict[str, dict[str, Any]] = {} + for path in sorted(self.root.glob("*.json")): + row = _read_json(path) + if (row.get("binding_id") in enabled and row.get("session_id") + and row.get("event", {}).get("chat_type") == "p2p" + and not row.get("agent_target") and row.get("status") in {"accepted", "delivered"}): + sources.setdefault(row["session_id"], row) + store = self.core.controller.store + for sid, source in sources.items(): + try: + stat = (store.sessions_root / sid / "messages.jsonl").stat() + snapshot = (stat.st_mtime_ns, stat.st_size) + if self._web_projection_snapshots.get(sid) == snapshot: + continue + messages = store.messages(sid) + web_turns = {message["turn_id"] for message in messages + if message.get("role") == "user" and message.get("origin") == "web"} + selected = [message for message in messages if message.get("turn_id") in web_turns + and message.get("role") in {"user", "agent"}] + if selected: + key = identity_ref(source["binding_id"], sid) + path = root / f"{key}.json" + with exclusive_file_lock(path, operation="discover_web_conversation_projection"): + row = _read_json(path) if path.exists() else { + "schema_version": "lark_private_web_projection_v0", + "projection_id": key, "request_ref": source["request_ref"], + "binding_id": source["binding_id"], "profile": source["profile"], + "source": source["source"], "event": source["event"], + "session_id": sid, "deliveries": {}, "status": "pending"} + if any(not row["deliveries"].get("web." + message["message_id"], {}).get("verified") + for message in selected): + row["status"] = "pending" + _atomic_write_json(path, row) + self._web_projection_snapshots[sid] = snapshot + except (OSError, ValueError, KeyError): + # An unreadable source is unknown. Keep any prior outbox; + # it retains fresh source and canonical-message checks. + continue + return [path for path in sorted(root.glob("*.json")) if _read_json(path).get("status") == "pending"] + + def _web_projection_messages(self, row: dict[str, Any]) -> list[dict[str, Any]]: + store = self.core.controller.store + source = _read_json(self.root / f"{row['request_ref']}.json") + native = self.core.read_request(row["request_ref"]) + if (any(row.get(key) != source.get(key) for key in + ("request_ref", "binding_id", "profile", "source", "event", "session_id")) + or any(native.get(key) != row.get(key) for key in ("binding_id", "source", "session_id")) + or source.get("agent_target") or row["event"].get("chat_type") != "p2p"): + raise ValueError("original App projection source changed") + session = store.load_session(row["session_id"]) + if session is None: + raise ValueError("original App projection Session unavailable") + saved = session.get("steward_context") or session.get("project_context") or {} + if (saved.get("binding_id") != row["binding_id"] or saved.get("audience") != "bound_owner" + or saved.get("source_ref") != row["source"]["source_ref"]): + raise ValueError("original App projection audience changed") + with self.bindings.input_context(saved, origin="web", session=session, store=store): + messages = store.messages(session["session_id"]) + web_turns = {message["turn_id"] for message in messages + if message.get("role") == "user" and message.get("origin") == "web"} + selected = [] + for message in messages: + if message.get("turn_id") not in web_turns or message.get("role") not in {"user", "agent"}: + continue + turn = store.load_turn(session["session_id"], message["turn_id"]) + if not turn or turn.get("origin") != "web" or turn.get("session_id") != session["session_id"]: + raise ValueError("canonical App input identity changed") + selected.append(message) + return selected + + def _reconcile_web_projection(self, path: Path) -> int: + # The existing worker pool serializes this outbox with a durable fence. + # Recovery reads the same provider attempt; it never starts a Core Turn. + with exclusive_file_lock(path, operation="deliver_web_conversation_projection"): + row = _read_json(path) + try: + messages = self._web_projection_messages(row) + 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" + _atomic_write_json(path, row) + return 0 + def authorized() -> bool: + try: + return pending in self._web_projection_messages(row) + except (OSError, ValueError, KeyError): + 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) + if delivered: + row.pop("blocker", None) + _atomic_write_json(path, row) + return int(delivered) + except (OSError, ValueError, KeyError): + row["blocker"] = "original_web_projection_unavailable" + _atomic_write_json(path, row) + return 0 def reconcile(self) -> int: return sum(self.reconcile_request(path) for path in self.pending_delivery_paths()) @@ -1191,6 +1315,8 @@ def _received_during_recovery(self, path: Path) -> None: self._feedback(path, record, inbox=lambda: self._inbox(record, observation=selected["observation"])) def reconcile_request(self, path: Path) -> int: + if path.parent == self.root / "web-projections": + return self._reconcile_web_projection(path) # Repair only this persisted source. Core still owns admission, recovery # and the shared Session queue/fence; transport workers never run models. row = _read_json(path) diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 50ddd40888..af5335c4db 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -1,5 +1,6 @@ """Public synthetic provider/native Core journeys, never private credentials.""" import json +import re import time import http.client import threading @@ -22,6 +23,8 @@ def __init__(self): self.calls = [] self.writes = [] self.topic_writes = [] + self.delivery_ids = {} + self.lose_next_reply = False self.verify_replies = True self.reactions = {} self.reaction_creates = [] @@ -100,7 +103,12 @@ def __call__(self, args, cwd=None, timeout=None): if "--dry-run" in args: data = {"ok": True, "api": [{"body": {"content": content, "msg_type": kind}}]} else: + delivery_id = (profile, args[args.index("--idempotency-key") + 1]) + if delivery_id in self.delivery_ids: + return {"returncode": 0, "stdout": json.dumps({"ok": True, + "data": {"message_id": self.delivery_ids[delivery_id]}}), "stderr": ""} ref = f"om_out_{len(self.writes)}" + self.delivery_ids[delivery_id] = ref self.writes.append((profile, text)) target = self.messages[args[args.index("--message-id") + 1]] if "+messages-reply" in args else None self.messages[ref] = {"message_id": ref, "msg_type": kind, "body": {"content": content}, @@ -111,6 +119,9 @@ def __call__(self, args, cwd=None, timeout=None): root = target.get("root_id") or target["message_id"] self.messages[ref].update(root_id=root, parent_id=target["message_id"], thread_id=f"omt_{root}") self.topic_writes.append((target["chat_id"], root, text)) + if self.lose_next_reply: + self.lose_next_reply = False + raise OSError("synthetic lost provider response after write") data = {"ok": True, "data": {"message_id": ref}} elif "reactions" in args: ref = args[args.index("--message-id") + 1] @@ -171,6 +182,135 @@ def connect_group(fixture): return store, runtime, provider, transport +def _app_projection_journey(fixture): + store, runtime, provider, transport = connect(fixture) + event = provider.event("notes-app", "original-app-chat", "Original source conversation") + transport.admit("notes-app", event) + original = next(row for row in transport.core.pending() if row["message"] == event["content"]) + runtime.wait_for_turn(session_id=original["session_id"], turn_id=original["turn_id"], timeout_sec=10) + transport.reconcile() + before = len(provider.writes) + turns = [] + for client in ["app-input-one", "app-input-two"]: + status, result = fixture[3](f"/api/chat/sessions/{original['session_id']}/turns", + {"message": "Same App input", "client_turn_id": client}) + assert status == 202 + assert runtime.wait_for_turn(session_id=original["session_id"], + turn_id=result["turn_id"], timeout_sec=10)["status"] == "completed" + turns.append(result["turn_id"]) + return store, runtime, provider, transport, original, turns, before + + +def _drain_app_projection(transport, predicate): + deadline = time.monotonic() + 10 + while not predicate(): + transport.reconcile() + assert time.monotonic() < deadline, transport.health() + time.sleep(.02) + + +def test_app_messages_and_answers_reach_the_original_chat_once_without_new_execution(ordinary): # noqa: F811 + store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary) + sid = original["session_id"] + canonical = [message for message in store.messages(sid) if message.get("turn_id") in turns] + requests_before = ordinary[4].read_text() + try: + _drain_app_projection(transport, lambda: len(provider.writes) >= before + len(canonical)) + projected = provider.writes[before:] + assert len(projected) == len(canonical) + # Repeated content remains two distinct messages, in canonical order. + for (profile, text), message in zip(projected, canonical): + assert profile == "notes-app" and message["text"] in text + assert ("你（App）" if message["role"] == "user" else "LoopX（App）") in text + restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root, + runner=provider, cli_bin="lark-cli") + for _ in range(3): + restarted.reconcile() + assert provider.writes[before:] == projected + assert ordinary[4].read_text() == requests_before + assert len(store.list_sessions()) == 1 + # Provider echoes are Bot-authored, never a second human input. + assert len([m for m in store.messages(sid) if m.get("origin") == "web"]) == 2 + finally: + runtime.close() + + +def test_app_projection_recovers_original_receipt_and_revocation_blocks_remaining_messages(ordinary): # noqa: F811 + store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary) + provider.verify_replies = False + try: + _drain_app_projection(transport, lambda: len(provider.writes) > before) + attempted = list(provider.writes) + transport.reconcile() + assert provider.writes == attempted + provider.verify_replies = True + transport.reconcile() + # A verified old receipt may advance; the remaining canonical messages + # retain the same grant and are denied after revocation. + binding = transport._binding("notes-app") + transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"]) + revoked_writes = list(provider.writes) + for _ in range(3): + transport.reconcile() + assert provider.writes == revoked_writes + assert len(store.list_sessions()) == 1 + finally: + runtime.close() + + +def test_app_long_return_keeps_all_parts_and_does_not_repeat_identical_parts(ordinary): # noqa: F811 + store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary) + text = ("A repeated output paragraph with its complete evidence.\n" * 4000) + "UNIQUE FINAL END" + store.append_message(original["session_id"], role="agent", text=text, + turn_id=turns[-1], message_id="synthetic-long-result") + try: + _drain_app_projection(transport, lambda: any("UNIQUE FINAL END" in t for _, t in provider.writes[before:])) + # Each canonical character survives transport splitting; an identical + # paragraph in different parts cannot reuse the same provider UUID. + parts = [t for _, t in provider.writes[before:] + if "A repeated output paragraph" in t or "UNIQUE FINAL END" in t] + assert len(parts) > 1 + combined = "".join(re.sub(r"^（\d+/\d+）\n\n", "", part) for part in parts) + assert combined == "**LoopX（App）**\n\n" + text + sent = list(provider.writes) + transport.reconcile() + assert provider.writes == sent + finally: + runtime.close() + + +def test_app_projection_does_not_retry_a_write_without_a_provider_locator(ordinary): # noqa: F811 + store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary) + provider.lose_next_reply = True + try: + _drain_app_projection(transport, lambda: len(provider.writes) > before) + attempted = list(provider.writes) + requests = ordinary[4].read_text() + restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root, + runner=provider, cli_bin="lark-cli") + for _ in range(3): + restarted.reconcile() + assert provider.writes == attempted + assert ordinary[4].read_text() == requests + assert restarted.health()[original["binding_id"]]["recovery_count"] == 1 + finally: + runtime.close() + + +def test_app_projection_refuses_a_replaced_provider_identity(ordinary): # noqa: F811 + store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary) + provider.profile_apps["notes-app"] = "cli_replaced_app" + try: + restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root, + runner=provider, cli_bin="lark-cli") + for _ in range(3): + restarted.reconcile() + assert len(provider.writes) == before + assert len(store.list_sessions()) == 1 + finally: + runtime.close() + + def finish_group_turn(runtime, transport, *, message): row = next(row for row in transport.core.pending() if row["message"] == message or row["message"].endswith(f"\n\nCurrent user message:\n{message}"))