fix(lark): retain verified thread identity for App quotes

apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py

@@ -32,6 +32,8 @@ def __init__(self, root, *, steward=False, peer=False):
         self.root, self.verified = root, True
         self.steward, self.peer = steward, peer
         self.peer_provider = None
+        default_compact_reply = "0" if os.environ.get("LOOPX_REPLY_BEFORE") == "1" else "1"
+        self.compact_reply = os.environ.get("LOOPX_REPLY_COMPACT_EVENT", default_compact_reply) == "1"
         if peer:
             self.transport_ref = "synthetic-peer"
 
@@ -42,7 +44,7 @@ def observe(self):
             from loopx.extensions.lark.conversation_identity import identity_ref
             operator = identity_ref(provider, self.reply_event()["sender_id"])
         return {"transport_ref": self.transport_ref, "provider_ref": provider,
-                "operator_ref": operator, "verified": self.verified}
+                "operator_ref": operator, "verified": self.verified, "bot_app_id": "cli_fixture"}
 
     def health_snapshot(self):
         return {"status": "listening"}
@@ -102,24 +104,30 @@ def start(self, server):
             event = self.reply_event()
             parent = "原要求：保留已有结果和失败记录。Original requirement: retain results and failed attempts."
             body = "请比较这次修改，并保留原话题中的限制。Compare this change against the original constraints."
-            raw = {**event, "source_conversation_id": event["chat_id"], "reply_context": {
-                "message_id": event["parent_id"], "conversation_id": event["chat_id"], "content": parent},
-                "thread_context": {"conversation_id": event["chat_id"], "root_message_id": event["root_id"],
-                    "thread_id": event["thread_id"], "messages": [
-                        {"message_id": event["root_id"], "conversation_id": event["chat_id"], "thread_id": event["thread_id"],
-                         "position": -1, "content": "### 话题背景 / Topic background\n\n" + "完整上下文保持可读。Keep the original context readable. " * 35,
-                         "sender": {"id": "ou_fixture_owner", "kind": "user"}},
-                        {"message_id": event["parent_id"], "conversation_id": event["chat_id"], "thread_id": event["thread_id"],
-                         "position": 0, "content": parent, "sender": {"id": "cli_fixture", "kind": "app"}},
-                        {"message_id": event["message_id"], "conversation_id": event["chat_id"], "thread_id": event["thread_id"],
-                         "position": 1, "content": body}]}}
-            frozen = conversation_reply_context(raw)["context_text"] + "\n\nCurrent user message:\n" + body
             request = identity_ref(self.observe()["provider_ref"], event["message_id"])
+            record = {"event": event, "profile": self.transport_ref, "request_ref": request}
+            if self.compact_reply:
+                selected = bindings.resolve(binding_id=binding["binding_id"], **source, include_observation=True)
+                frozen = self.capture_compact_reply(runtime, record, selected, body, parent)
+            else:
+                raw = {**event, "source_conversation_id": event["chat_id"], "reply_context": {
+                    "message_id": event["parent_id"], "conversation_id": event["chat_id"], "content": parent},
+                    "thread_context": {"conversation_id": event["chat_id"], "root_message_id": event["root_id"],
+                        "thread_id": event["thread_id"], "messages": [
+                            {"message_id": event["root_id"], "conversation_id": event["chat_id"], "thread_id": event["thread_id"],
+                             "position": -1, "content": self.reply_background,
+                             "sender": {"id": "ou_fixture_owner", "kind": "user"}},
+                            {"message_id": event["parent_id"], "conversation_id": event["chat_id"], "thread_id": event["thread_id"],
+                             "position": 0, "content": parent, "sender": {"id": "cli_fixture", "kind": "app"}},
+                            {"message_id": event["message_id"], "conversation_id": event["chat_id"], "thread_id": event["thread_id"],
+                             "position": 1, "content": body}]}}
+                frozen = conversation_reply_context(raw)["context_text"] + "\n\nCurrent user message:\n" + body
+                record.update(reply_context_message=frozen, reply_context_source_text=body)
             turn, _ = server.chat_store.create_turn(sid, client_turn_id=f"external-{request}", message=frozen, origin="lark")
             correlation = {"request_ref": request, "binding_id": binding["binding_id"], "source": source,
                            "session_id": sid, "turn_id": turn["turn_id"]}
             for directory, value in (("external-requests", {**correlation, "status": "accepted", "command": None, "message": frozen}),
-                    ("lark-private-deliveries", {**correlation, "status": "delivered", "event": event, "reply_context_message": frozen, "reply_context_source_text": body})):
+                    ("lark-private-deliveries", {**record, **correlation, "status": "delivered"})):
                 _atomic_write_json(server.chat_store.root / directory / f"{request}.json", value)
             server.chat_store.update_turn(sid, turn["turn_id"], status="completed")
             server.chat_store.release_active_turn(sid, turn["turn_id"])
@@ -225,11 +233,45 @@ def unavailable(handler):
                 print(json.dumps(result), flush=True)
         threading.Thread(target=control, daemon=True).start()
 
-    @staticmethod
-    def reply_event():
-        return {"message_id": "om_fixture_reply", "parent_id": "om_fixture_parent", "root_id": "om_fixture_root",
-                "thread_id": "omt_fixture_topic", "chat_id": "oc_fixture_private", "chat_type": "p2p",
-                "sender_type": "user", "sender_id": "ou_fixture_owner"}
+    reply_background = ("### 话题背景 / Topic background\n\n"
+                        + "完整上下文保持可读。Keep the original context readable. " * 35)
+
+    def reply_event(self):
+        parent = "om_fixture_parent"
+        return {"message_id": "om_fixture_reply", "parent_id": parent,
+                "root_id": parent if self.compact_reply else "om_fixture_root",
+                **({} if self.compact_reply else {"thread_id": "omt_fixture_topic"}),
+                "chat_id": "oc_fixture_private", "chat_type": "p2p",
+                "sender_type": "user", "sender_id": "ou_fixture_owner", "message_type": "text"}
+
+    def capture_compact_reply(self, runtime, record, selected, body, parent):
+        from loopx.extensions.lark.private_conversations import LarkPrivateConversations
+
+        event = record["event"]
+        common = {"chat_id": event["chat_id"], "msg_type": "text", "thread_id": "omt_fixture_topic"}
+        current = {**common, "message_id": event["message_id"], "content": body,
+                   "parent_id": event["parent_id"], "root_id": event["root_id"],
+                   "sender": {"id": event["sender_id"], "sender_type": "user"}, "thread_message_position": 1}
+        earlier = {**common, "message_id": "om_fixture_earlier", "content": self.reply_background,
+                   "sender": current["sender"], "thread_message_position": 0}
+        quoted = {**common, "message_id": event["parent_id"], "content": parent,
+                  "sender": {"id": "cli_fixture", "sender_type": "app"}, "thread_message_position": -1,
+                  "thread_replies": [earlier, current]}
+
+        def read(args, *_):
+            if args[3:] == ["api", "GET", f"/open-apis/im/v1/messages/{event['message_id']}", "--as", "bot"]:
+                payload = {"code": 0, "data": {"items": [current]}}
+            else:
+                assert args[3:7] == ["im", "+messages-mget", "--message-ids", event["parent_id"]], args
+                payload = {"ok": True, "data": {"items": [quoted]}}
+            return {"returncode": 0, "stdout": json.dumps(payload), "stderr": ""}
+
+        transport = LarkPrivateConversations(controller=runtime, runtime_root=runtime.store.root.parent,
+            runner=read, cli_bin="lark-cli")
+        # Exercise the real provider-backed freeze before seeding Core history.
+        # Opening the resulting page still does not require any model Turns.
+        return transport._reply_context_message(transport.root / f"{record['request_ref']}.json",
+            record, selected=selected, source_message=current, text=body)
 
 
 def main():

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

@@ -162,6 +162,12 @@ remain unavailable. There is no fallback to ordinary steward/project history,
 Session creation, executor change or model reconfiguration on this entry.
 Group audiences retain their existing boundary.
 
+For a private Lark reply, the adapter keeps the thread ID from the verified
+provider context when the incoming event omits it. This lets the App show the
+original body, exact quote and verified thread excerpt together. Redelivery
+can omit that ID and adopt the same request. A conflicting supplied ID still
+fails source verification.
+
 An owner-private steward connection also offers **Use for App steward**. Core
 stores this explicit entry preference alongside the binding set, without
 changing any audience, managed identity or execution grant. Multiple stewards

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

@@ -115,6 +115,10 @@ Lark listener 观测。该本地 hook 只读取缓存，不发起网络请求。
 不回退到普通管家/项目历史，不新建替代 Session，也不在这里切换执行器或模型。
 群受众保留原有边界。
 
+飞书私聊回复的事件省略话题 ID 时，适配器从已核验的原消息上下文保留该 ID。
+App 因此可以一起呈现原正文、精确引用和已核验的话题节选。重复到达时省略
+该 ID 仍可沿用原请求，明确给出不同 ID 时继续拒绝。
+
 已有本人私聊管家连接还提供 **用于 App 管家**。Core 在原 binding 集合中保存
 明确的入口选择，不改变受众、受管身份或执行授权。多个管家不按列表顺序或最近
 活跃时间竞争。点击本机 App 管家聊天时，仅在所选来源中发现当前已有 Session，

loopx/extensions/lark/private_conversations.py

@@ -635,6 +635,13 @@ def read(args: Sequence[str], **_kwargs: Any) -> subprocess.CompletedProcess[str
         projected = conversation_reply_context({**context, "source_conversation_id": event["chat_id"]})
         prefix = str(projected["context_text"])
         message = f"{prefix}\n\nCurrent user message:\n{text}" if prefix else text
+        # A compact private event can omit its thread ID. Retain the verified
+        # identity for redelivery even when the provider has no thread excerpt.
+        thread_id = context.get("thread_id")
+        if (event["chat_type"] == "p2p" and event.get("parent_id") and not event.get("thread_id")
+                and context.get("reply_context_verified") is True
+                and isinstance(thread_id, str) and 0 < len(thread_id) <= 200):
+            record["event"] = {**event, "thread_id": thread_id}
         record.update(reply_context_message=message, reply_context_source_text=text)
         _atomic_write_json(path, record)
         return message
@@ -682,8 +689,15 @@ def admit(self, profile: str, event: dict[str, Any], *,
                     return {"status": "history_answer_held"}
                 # event_id can change on redelivery; canonical message content
                 # and audience cannot change underneath the stable message id.
-                if any(record["event"].get(key) != event.get(key) for key in
-                       ["message_id", "sender_id", "chat_id", "content", "message_type", "chat_type", "sender_type", "root_id", "parent_id", "thread_id"]):
+                # A compact private redelivery may omit a thread ID already
+                # verified and frozen with the original reply context.
+                saved_event = record["event"]
+                omitted_private_thread = (event["chat_type"] == "p2p" and not event.get("thread_id")
+                                          and "reply_context_message" in record)
+                if any(saved_event.get(key) != event.get(key) for key in (
+                        "message_id", "sender_id", "chat_id", "content", "message_type",
+                        "chat_type", "sender_type", "root_id", "parent_id")) or (
+                        not omitted_private_thread and saved_event.get("thread_id") != event.get("thread_id")):
                     return {"status": "source_conflict"}
             else:
                 record = {"schema_version": "lark_private_chat_delivery_v0", "request_ref": request,

tests/test_lark_private_conversations.py

@@ -4390,6 +4390,93 @@ def observe(*args, **kwargs):
         runtime.close()
 
 
+@pytest.mark.parametrize("captured_thread", [False, True])
+def test_private_reply_keeps_the_exact_quote_when_thread_id_is_optional(ordinary, captured_thread):  # noqa: F811
+    store, runtime, provider, transport = connect(ordinary)
+    profile = "steward-app"
+    parent = provider.event(profile, "private-quote-parent", "Original quote\n\n**Second paragraph**")
+    reply = provider.event(profile, "private-quote-reply", "My question\n\n**Keep its formatting**")
+    parent_id, reply_id = parent["message_id"], reply["message_id"]
+    thread_id = "omt_private_quote"
+    reply.update(parent_id=parent_id, root_id=parent_id, thread_id=thread_id if captured_thread else None)
+    child = provider.messages[reply_id]
+    child.update(parent_id=parent_id, root_id=parent_id, thread_id=thread_id, thread_message_position=0)
+    provider.messages[parent_id].update(thread_id=thread_id, thread_message_position=-1, thread_replies=[child])
+    try:
+        assert transport.admit(profile, reply)["status"] == "durably_accepted"
+        request, = transport.core.pending()
+        sid, tid = request["session_id"], request["turn_id"]
+        transport = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
+            runner=provider, cli_bin="lark-cli")
+        redelivery = {**reply, "event_id": "redelivered-private-quote", "thread_id": None}
+        assert transport.admit(profile, redelivery)["status"] == "durably_accepted"
+        assert transport.admit(profile, {**redelivery, "thread_id": "omt_other"})["status"] == "source_conflict"
+        runtime.wait_for_turn(session_id=sid, turn_id=tid, timeout_sec=10)
+        original, = [row for row in store.messages(sid) if row["role"] == "user"]
+        status, snapshot = ordinary[3](f"/api/chat/sessions/{sid}?message_limit=5&turn_id={tid}")
+        assert status == 200
+        visible = next(row for row in snapshot["messages"] if row["message_id"] == original["message_id"])
+        context = visible["reply_context"]
+        assert context["source_text"] == reply["content"]
+        assert context["parent"] == {"message_id": parent_id, "content": parent["content"], "complete": True}
+        assert context["thread"]["thread_id"] == thread_id
+        assert context["thread"]["before_message_id"] == reply_id
+        assert visible["text"] == original["text"] == request["message"]
+        transport.reconcile()
+        assert provider.topic_writes == [(reply["chat_id"], parent_id, "Runtime response.")]
+    finally:
+        runtime.close()
+
+
+@pytest.mark.parametrize("event_shape", ["compact", "full", "provider_without_thread"])
+def test_nested_private_reply_deduplicates_when_thread_history_is_unavailable(ordinary, event_shape):  # noqa: F811
+    store, runtime, provider, transport = connect(ordinary)
+    profile = "steward-app"
+    root = provider.event(profile, "nested-root", "Earlier topic")
+    parent = provider.event(profile, "nested-parent", "Exact quoted answer\n\n**Second paragraph**")
+    reply = provider.event(profile, "nested-reply", "My question\n\n**Keep its formatting**")
+    root_id, parent_id, reply_id = (row["message_id"] for row in (root, parent, reply))
+    thread_id = "omt_nested_child"
+    reply.update(parent_id=parent_id, root_id=root_id)
+    provider.messages[parent_id].update(parent_id=root_id, root_id=root_id, thread_id="omt_parent",
+        sender={"id": provider.profile_apps[profile], "sender_type": "app"})
+    child = provider.messages[reply_id]
+    child.update(parent_id=parent_id, root_id=root_id)
+    if event_shape != "provider_without_thread":
+        child.update(thread_id=thread_id, thread_message_position=-1)
+    if event_shape == "full":
+        reply["thread_id"] = thread_id
+    try:
+        assert transport.admit(profile, reply)["status"] == "durably_accepted"
+        request, = transport.core.pending()
+        sid, tid = request["session_id"], request["turn_id"]
+        transport = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
+            runner=provider, cli_bin="lark-cli")
+        full = {**reply, "event_id": "nested-full-redelivery", "thread_id": thread_id}
+        expected = "source_conflict" if event_shape == "provider_without_thread" else "durably_accepted"
+        assert transport.admit(profile, full)["status"] == expected
+        compact = {**reply, "event_id": "nested-compact-redelivery"}
+        compact.pop("thread_id", None)
+        assert transport.admit(profile, compact)["status"] == "durably_accepted"
+        assert transport.admit(profile, {**full, "thread_id": "omt_parent"})["status"] == "source_conflict"
+        assert transport.admit(profile, {**full, "parent_id": root_id})["status"] == "source_conflict"
+        assert runtime.wait_for_turn(session_id=sid, turn_id=tid, timeout_sec=10)["status"] == "completed"
+        original, = [row for row in store.messages(sid) if row["role"] == "user"]
+        status, snapshot = ordinary[3](f"/api/chat/sessions/{sid}?message_limit=5&turn_id={tid}")
+        assert status == 200
+        visible = next(row for row in snapshot["messages"] if row["message_id"] == original["message_id"])
+        context = visible["reply_context"]
+        assert context["source_text"] == reply["content"]
+        assert context["parent"] == {"message_id": parent_id, "content": parent["content"], "complete": True}
+        assert "thread" not in context
+        assert visible["text"] == original["text"] == request["message"]
+        transport.reconcile()
+        replies = [args for args in provider.calls if "+messages-reply" in args and "--dry-run" not in args]
+        assert sum(args[args.index("--message-id") + 1] == reply_id for args in replies) == 1
+    finally:
+        runtime.close()
+
+
 def test_group_admission_source_reads_overlap_but_keep_rendered_model_input(ordinary, monkeypatch):  # noqa: F811
     _, runtime, provider, transport = connect_group(ordinary)
     try: