fix(lark): keep queued notices faithful to native state

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

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，以及重启后原会话恢复和已确认回复不重复发送。

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:

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()

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.