diff --git a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py b/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py index 58a0014b05..f033b80a74 100644 --- a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py +++ b/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(): 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 927c48b270..8d712a409e 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/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 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 47ea747e5e..9850e534f4 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 @@ -115,6 +115,10 @@ Lark listener 观测。该本地 hook 只读取缓存，不发起网络请求。 不回退到普通管家/项目历史，不新建替代 Session，也不在这里切换执行器或模型。 群受众保留原有边界。 +飞书私聊回复的事件省略话题 ID 时，适配器从已核验的原消息上下文保留该 ID。 +App 因此可以一起呈现原正文、精确引用和已核验的话题节选。重复到达时省略 +该 ID 仍可沿用原请求，明确给出不同 ID 时继续拒绝。 + 已有本人私聊管家连接还提供 **用于 App 管家**。Core 在原 binding 集合中保存 明确的入口选择，不改变受众、受管身份或执行授权。多个管家不按列表顺序或最近 活跃时间竞争。点击本机 App 管家聊天时，仅在所选来源中发现当前已有 Session， diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 494f7980a5..2e3656483b 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/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, diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 312f0b183c..3fc1a1a8c2 100644 --- a/tests/test_lark_private_conversations.py +++ b/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: