From c848749a2c521f7f6b5f7ef082721ebaada5344d Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:03:21 +0800 Subject: [PATCH] fix(lark): keep queued notices faithful to native state Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 3 + ...p-conversation-and-async-inbox-v0.zh-CN.md | 2 + .../extensions/lark/private_conversations.py | 4 +- tests/test_lark_private_conversations.py | 18 ++++- tests/test_lark_private_feedback.py | 70 ++++++++++++++++++- 5 files changed, 93 insertions(+), 4 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 f8ba9a47ba..3c80b67888 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -52,6 +52,9 @@ A lost admission correlation cannot move a request to a newer Session. Admission feedback and final delivery use separate durable provider intents; an ambiguous write is read back without blind resend. Unsupported media receives an explicit notice. The workspace grant conveys no Goal, portfolio or peer execution authority. +Lark queued notices describe the observed queue state without claiming that an +earlier request is running: the first Turn can also await its native claim. +Recovery retains an already persisted notice's exact text and provider intent. The packaged settings journey, source revocation/recovery, duplicate events, native queue/stop and two-App isolation have synthetic-provider regression and 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 c925bc7bd7..952115093c 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 @@ -37,6 +37,8 @@ Lark 页面选择一个非默认 App、当前可用工作区和宿主 executor Turn FIFO 和确切 stop/new/status 目标。传输层保存独立的受理与最终投递意图，通过 provider 读回确认回复；发生没有 receipt 的不确定写入时不盲目重发。两 App 复用 现有服务及按 App 获取的 listener lease，不创建第二套队列、模型 runner 或隐藏 Goal。 +飞书排队提示只表达已观测的队列状态，不断言已有前一条请求正在运行：首个 Turn +也可能等待原生 claim。恢复继续使用已持久化提示的原文与 provider intent。 源码 canary 已验证打包页面的桌面/窄屏、撤权拒绝与恢复、慢会话期间另一 App 完成、 后续消息持久排队、exact stop，以及重启后原会话恢复和已确认回复不重复发送。 diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index c35b54ba5f..5de8edec44 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -747,7 +747,9 @@ def inbox() -> Path: if native.get("agent_target"): admission = "已收到，等待原 Agent 宿主处理。/status 查看状态，/project 返回项目对话。" elif turn and turn["status"] == "queued": - admission = "正在处理前一条，这条已排队。" + # A first Turn can also await its native claim; + # queued alone does not prove a preceding request. + admission = "已收到，这条已排队。" elif (record.get("feedback", {}).get("received") or {}).get("ok") is not True: admission = "已收到，正在处理。" if admission: diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index d93942c260..d51a18baee 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -217,14 +217,28 @@ def observed(path): transport.reconcile() assert len(store.list_sessions()) == 1 assert [m["text"] for m in store.messages(row["session_id"]) if m["role"] == "user"] == [event["content"]] - final = ("notes-app", result["response"]["message"]) - assert provider.writes[-1] == final and provider.writes.count(final) == 1 + delivered = read(path) + terminal = delivered["deliveries"]["terminal"] + assert delivered["status"] == "delivered" and terminal["verified"] is True + # A running Turn can create a draft before recovery returns. Terminal + # delivery then edits it; writes records creations, not current bodies. + final_ref = terminal["attempt"]["message_ref"] + final_message = provider.messages[final_ref] + content = json.loads(final_message["body"]["content"]) + final_text = result["response"]["message"] + assert content["zh_cn"]["content"][0][0]["text"] == final_text + assert sum(message.get("body") == final_message["body"] + for message in provider.messages.values()) == 1 assert all(profile == "notes-app" for profile, _ in provider.writes) packets = [json.loads(line) for line in ordinary[4].read_text().splitlines()] assert sum(packet["method"] == "turn/start" for packet in packets) == 1 if group: assert provider.topic_writes[0][:2] == (event["chat_id"], event["message_id"]) + visible_before = json.dumps(provider.messages, sort_keys=True) + writes_before = list(provider.writes) assert transport.reconcile() == 0 + assert json.dumps(provider.messages, sort_keys=True) == visible_before + assert provider.writes == writes_before finally: release_core.set() runtime.close() diff --git a/tests/test_lark_private_feedback.py b/tests/test_lark_private_feedback.py index 4a97ae7eeb..ed5c8c61c7 100644 --- a/tests/test_lark_private_feedback.py +++ b/tests/test_lark_private_feedback.py @@ -18,6 +18,74 @@ def active(store, row): time.sleep(.01) +@pytest.mark.parametrize('group', [False, True]) +@pytest.mark.parametrize('feedback', ['enabled', 'disabled', 'unavailable']) +def test_first_durable_queued_turn_does_not_claim_a_previous_request(ordinary, monkeypatch, group, feedback): # noqa: F811 + store, runtime, provider, transport = (connect_group if group else connect)(ordinary) + resume = runtime.resume_session_queue + # Hold only the execution effect. Admission, its durable native queue and + # the provider reply/recovery path remain the production implementations. + monkeypatch.setattr(runtime, 'resume_session_queue', lambda **kwargs: None) + transport.reaction_feedback = feedback != 'disabled' + provider.fail_reaction_create = feedback == 'unavailable' + try: + event = (provider.topic('first_queued', 'ordinary work') if group else + provider.event('notes-app', 'first_queued', 'ordinary work')) + assert transport.admit('notes-app', event)['status'] == 'durably_accepted' + row = transport.core.pending()[0] + sid, tid = row['session_id'], row['turn_id'] + assert store.load_session(sid)['active_turn_id'] is None + assert [turn['turn_id'] for turn in store.queued_turns(sid)] == [tid] + assert len(store.list_sessions()) == 1 + transport.reconcile() + # The first Turn really is queued; this alone cannot establish a + # preceding request. The notice must not invent one, even without Get. + assert len(provider.writes) == 1 + notice = provider.writes[0][1] + assert '前一条' not in notice + assert '排队' in notice + path = transport.root / f"{row['request_ref']}.json" + assert json.loads(path.read_text())['deliveries']['admission']['verified'] is True + before = list(provider.writes) + transport.reconcile() + assert provider.writes == before + monkeypatch.setattr(runtime, 'resume_session_queue', resume) + transport.core.recover(request_ref=row['request_ref']) + assert runtime.wait_for_turn(session_id=sid, turn_id=tid, timeout_sec=10)['status'] == 'completed' + transport.reconcile() + assert provider.writes.count(('notes-app', 'Runtime response.')) == 1 + assert transport.core.read_request(row['request_ref'])['delivery_verified'] is True + assert store.load_session(sid)['active_turn_id'] is None + assert len(store.list_sessions()) == 1 + if group: + assert all(root == event['message_id'] for _, root, _ in provider.topic_writes) + finally: + runtime.close() + + +@pytest.mark.parametrize('group', [False, True]) +def test_queued_notice_recovery_keeps_its_existing_exact_intent(ordinary, monkeypatch, group): # noqa: F811 + _, runtime, provider, transport = (connect_group if group else connect)(ordinary) + monkeypatch.setattr(runtime, 'resume_session_queue', lambda **kwargs: None) + try: + event = (provider.topic('old_queued_notice', 'ordinary work') if group else + provider.event('notes-app', 'old_queued_notice', 'ordinary work')) + transport.admit('notes-app', event) + row = transport.core.pending()[0] + path = transport.root / f"{row['request_ref']}.json" + record = json.loads(path.read_text()) + legacy = '正在处理前一条，这条已排队。' + record['deliveries']['admission'] = {'text': legacy} + path.write_text(json.dumps(record)) + transport.reconcile() + assert provider.writes == [('notes-app', legacy)] + transport.reconcile() + assert provider.writes == [('notes-app', legacy)] + assert json.loads(path.read_text())['deliveries']['admission']['verified'] is True + finally: + runtime.close() + + @pytest.mark.parametrize('group', [False, True]) def test_durable_admission_feedback_precedes_blocked_execution_recovery(ordinary, monkeypatch, group): # noqa: F811 store, runtime, provider, transport = (connect_group if group else connect)(ordinary) @@ -263,7 +331,7 @@ def test_default_feedback_tracks_queue_execution_stop_and_app_isolation(ordinary assert ('notes-app', slow['message_id'], 'OnIt') in provider.reaction_creates assert ('notes-app', queued['message_id'], 'Get') in provider.reaction_creates assert ('notes-app', queued['message_id'], 'OnIt') not in provider.reaction_creates - assert provider.writes == [('notes-app', '正在处理前一条，这条已排队。'), + assert provider.writes == [('notes-app', '已收到，这条已排队。'), ('notes-app', '⏳ **已开始处理**')] assert any(row[1:] == (slow['message_id'], 'OnIt') for row in provider.reactions.values()) # A new provider instance reuses durable receipts, not in-memory emoji state.