fix(lark): mirror terminal App failures to the original chat

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

@@ -275,6 +275,12 @@ 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.
+An exact App-origin terminal error also projects a public-safe failure notice
+with the original completion time. The notice is derived from the stored Turn
+and its matching error row, without exporting raw diagnostics. It uses the same
+outbox and message identity, so an already-mirrored input can receive its
+historical failure notice once after recovery. Earlier stop errors and completed
+Turns do not qualify as failures.
 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;

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

@@ -196,6 +196,10 @@ Web 来源，通过已有 Core 队列及新鲜来源准入。连接列表独立
 输入只提供地址来源证据；投影不会重新受理它或启动模型工作。每次发送和恢复回执前，
 Core 重查 Session、当前绑定与受众，provider 核验原来源。消息标明 App 发言者，在
 该 outbox 内按 canonical 追加顺序发送，长正文沿用既有富文本分段，保留完整内容。
+App 来源 Turn 的终态错误也沿原私聊补齐。根据原 Turn 与精确匹配的错误行生成
+失败提示，并标注原结束时间。原始诊断留在 App。恢复时使用原 outbox 与消息身份，
+已送达的输入可以补回一条历史失败提示。先前未确认的停止提示与已完成的 Turn
+不作为终态失败投递。
 即使正文相同，不同 canonical 消息及分段也使用不同 provider 幂等键；未提供该身份
 的既有调用者保持原回执 digest。恢复先读回已知发送，再继续未发送部分；已经开始但
 没有 provider locator 的写入，重启后仍保留未确认。撤权或身份变化阻止后续投递。

loopx/extensions/lark/private_conversations.py

@@ -1315,11 +1315,7 @@ def _web_projection_paths(self) -> list[Path]:
                         continue
                     key = identity_ref(source["binding_id"], sid)
                     path = root / f"{key}.json"
-                    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"}]
+                    selected = self._web_projection_transcript(sid)
                     if selected:
                         with exclusive_file_lock(path, operation="discover_web_conversation_projection"):
                             row = _read_json(path) if path.exists() else {
@@ -1339,6 +1335,40 @@ def _web_projection_paths(self) -> list[Path]:
                     continue
         return [path for path in sorted(root.glob("*.json")) if _read_json(path).get("status") == "pending"]
 
+    def _web_projection_transcript(self, session_id: str, *, inputs_only: bool = False,
+                                   verify_turns: bool = False) -> list[dict[str, Any]]:
+        store = self.core.controller.store
+        messages = store.messages(session_id)
+        web_turns = {message["turn_id"] for message in messages
+                     if message.get("role") == "user" and message.get("origin") == "web"}
+        roles = {"user"} if inputs_only else {"user", "agent", "error"}
+        selected = []
+        failed_turns = set()
+        for message in messages:
+            turn_id, role = message.get("turn_id"), message.get("role")
+            if turn_id not in web_turns or role not in roles:
+                continue
+            turn = store.load_turn(session_id, turn_id) if verify_turns or role == "error" else None
+            same_turn = bool(turn and turn.get("origin") == "web" and turn.get("session_id") == session_id
+                             and turn.get("turn_id") == turn_id)
+            if role == "error":
+                # An unconfirmed stop may leave an earlier error row in the
+                # transcript. Use only the exact terminal failure, once per Turn.
+                if (not same_turn or turn_id in failed_turns or turn["status"] not in {"failed", "timed_out"}
+                        or not turn.get("completed_at") or turn.get("error") != message.get("text")
+                        or message.get("attachments")):
+                    continue
+                failed_turns.add(turn_id)
+                if turn.get("error_code") == "server_restarted":
+                    failure = "服务在处理此请求时重启，本轮未完成。详情可在 App 的原会话查看。"
+                else:
+                    _, failure = manager_failure_reply(turn)
+                message = {**message, "text": f"App 执行记录（{turn['completed_at']}）\n\n{failure}"}
+            elif verify_turns and not same_turn:
+                raise ValueError("canonical App input identity changed")
+            selected.append(message)
+        return selected
+
     def _web_projection_messages(self, row: dict[str, Any], *, inputs_only: bool = False) -> list[dict[str, Any]]:
         store = self.core.controller.store
         source = _read_json(self.root / f"{row['request_ref']}.json")
@@ -1356,23 +1386,7 @@ def _web_projection_messages(self, row: dict[str, Any], *, inputs_only: bool = F
                 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
-                # Reactions belong to canonical user inputs. The independent
-                # body outbox still reads and validates every answer; feedback
-                # need not reload those Turns just to discard their messages.
-                if inputs_only and message["role"] != "user":
-                    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
+            return self._web_projection_transcript(session["session_id"], inputs_only=inputs_only, verify_turns=True)
 
     @staticmethod
     def _web_projection_delivery_id(row: dict[str, Any], message: dict[str, Any]) -> str:

tests/test_lark_private_conversations.py

@@ -2336,6 +2336,83 @@ def test_app_messages_and_answers_reach_the_original_chat_once_without_new_execu
         runtime.close()
 
 
+@pytest.mark.parametrize("recovery", ["readback", "revoked"])
+def test_app_failed_turn_mirrors_only_its_terminal_notice_to_the_original_chat(ordinary, monkeypatch, recovery):  # noqa: F811
+    store, runtime, provider, transport = connect(ordinary)
+    profile = "steward-app"
+    event = provider.event(profile, "original-app-failure-source", "Original steward chat")
+    transport.admit(profile, event)
+    original = transport.core.pending()[0]
+    sid = original["session_id"]
+    try:
+        runtime.wait_for_turn(session_id=sid, turn_id=original["turn_id"], timeout_sec=10)
+        transport.reconcile()
+        with monkeypatch.context() as queued:
+            queued.setattr(runtime, "resume_session_queue", lambda **_: None)
+            status, turn = ordinary[3](f"/api/chat/sessions/{sid}/turns", {
+                "message": "App request with typed terminal failure", "client_turn_id": "app-failed-once"})
+        assert status == 202 and turn["status"] == "queued"
+        turn_id = turn["turn_id"]
+        stale = store.append_message(sid, role="error", text="Unconfirmed synthetic stop detail", turn_id=turn_id)
+        _drain_app_projection(transport, lambda: any("你（App）" in text for _, text in provider.writes))
+        transport.reconcile()
+        path, = (transport.root / "web-projections").glob("*.json")
+        original_outbox = json.loads(path.read_text())
+        assert original_outbox["status"] == "delivered"
+        runtime.resume_session_queue(session_id=sid, work_dir=ordinary[-1], objective="")
+        failed = runtime.wait_for_turn(session_id=sid, turn_id=turn_id, timeout_sec=10)
+        assert failed["status"] == "failed" and failed["error_code"] == "cyber_policy", failed
+        errors = [item for item in store.messages(sid) if item["turn_id"] == turn_id
+                  and item["role"] == "error" and item["text"] == failed["error"]]
+        canonical, = errors
+        duplicate = store.append_message(sid, role="error", text=failed["error"], turn_id=turn_id)
+        phase = "web." + canonical["message_id"]
+        turns_before = ordinary[4].read_bytes()
+        transcript = (store.sessions_root / sid / "messages.jsonl").read_bytes()
+        before = len(provider.writes)
+
+        def worker():
+            return LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
+                runner=provider, cli_bin="lark-cli")
+
+        if recovery == "revoked":
+            binding = transport._binding(profile)
+            transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"])
+            restarted = worker()
+            for _ in range(3):
+                restarted.reconcile()
+            assert len(provider.writes) == before
+            assert phase not in json.loads(path.read_text())["deliveries"]
+        else:
+            provider.verify_replies = False
+            _drain_app_projection(worker(), lambda: len(provider.writes) == before + 1)
+            attempted = json.loads(path.read_text())["deliveries"][phase]
+            assert attempted["attempt"] and not attempted.get("verified")
+            provider.verify_replies = True
+            restarted = worker()
+            _drain_app_projection(restarted, lambda: json.loads(path.read_text())["deliveries"][phase].get("verified"))
+            restarted.reconcile()
+            outbox = json.loads(path.read_text())
+            assert outbox["status"] == "delivered"
+            assert outbox["deliveries"][phase]["attempt"] == attempted["attempt"]
+            for key, old in original_outbox["deliveries"].items():
+                assert {k: v for k, v in outbox["deliveries"][key].items() if k != "feedback"} == {
+                    k: v for k, v in old.items() if k != "feedback"}
+            assert "web." + stale["message_id"] not in outbox["deliveries"]
+            assert "web." + duplicate["message_id"] not in outbox["deliveries"]
+            assert len(provider.writes) == before + 1
+            sender, body = provider.writes[-1]
+            assert sender == profile and "上游安全策略拦截" in body and failed["completed_at"] in body
+            assert failed["error"] not in body and stale["text"] not in body
+            mirror = provider.messages[attempted["attempt"]["message_ref"]]
+            assert mirror["chat_id"] == event["chat_id"]
+        assert ordinary[4].read_bytes() == turns_before
+        assert (store.sessions_root / sid / "messages.jsonl").read_bytes() == transcript
+        assert len(store.list_sessions()) == 1
+    finally:
+        runtime.close()
+
+
 @pytest.mark.parametrize("terminal", ["completed", "failed", "interrupted"])
 def test_app_mirror_feedback_tracks_turn_only_changes_and_separate_sessions(ordinary, monkeypatch, tmp_path, terminal):  # noqa: F811
     gate = tmp_path / "finish-app-input"
@@ -2407,6 +2484,8 @@ def test_app_mirror_feedback_tracks_turn_only_changes_and_separate_sessions(ordi
         transport._web_projection_scan_at = 0
         _drain_app_projection(transport, lambda: not any(ref in targets.values() and emoji == "OnIt"
                                                        for _, ref, emoji in provider.reactions.values()))
+        if terminal == "failed":
+            _drain_app_projection(transport, lambda: sum("App 执行记录" in body for _, body in provider.writes) == 2)
         assert {(profile, ref, emoji) for profile, ref, emoji in provider.reactions.values() if ref in targets.values()} == {
             (profile, ref, "Get") for profile, ref in targets.items()}
         starts = ordinary[4].read_bytes()