From 5b089800b421d1ac9a0aac5d5ba807ef026c7f8c Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 2 Nov 2026 05:15:14 +0530
Subject: [PATCH 1/3] fix(lark): track feedback on verified App input mirrors

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../app-conversation-and-async-inbox-v0.md    |  17 ++
 ...p-conversation-and-async-inbox-v0.zh-CN.md |  11 +
 loopx/extensions/lark/inbox_reactions.py      |  82 ++++++-
 .../extensions/lark/private_conversations.py  | 117 ++++++++-
 tests/extensions/test_lark_inbox_reactions.py | 112 +++++++++
 tests/test_lark_private_conversations.py      | 231 ++++++++++++++++++
 6 files changed, 549 insertions(+), 21 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 3adf3b1f45..75014c0890 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -282,6 +282,23 @@ a started write without a provider locator stays unconfirmed after restart.
 Revocation or changed identity prevents further delivery. Disable the existing
 connection to revoke it; rollback must preserve its private delivery journals.
 
+Newly verified App user mirrors also reuse the existing reaction journals:
+`Get` acknowledges receipt; only an observed starting/running Core Turn adds
+`OnIt`. Terminal completion, failure or interruption removes processing feedback
+while retaining receipt feedback. A queued input never claims to be processing,
+and assistant mirrors receive neither reaction. Long inputs use their last
+verified concrete post, not the composite parts locator. Turn-only changes are
+discovered within the existing five-second scan even without a transcript append.
+Before each effect, the worker rechecks the canonical input/Turn, current source
+and grants, and frozen provider message. Feedback failures retain recoverable
+receipts without blocking message delivery or starting work. Unknown provider
+writes are not retried. No synthetic incoming event or original-request
+settlement is used. Existing finished mirrors are not backfilled; current active
+mirrors can adopt the feedback path. `loopx chat --no-private-reactions` keeps
+feedback disabled without changing the message path. The isolated HTTP/Core and
+synthetic-provider checks qualify source behavior; installed Lark reactions and
+actual provider presentation remain separate acceptance work.
+
 Use **Continue original conversation** or the explicitly selected App steward
 entry above to reach this path. Old ordinary history is retained separately.
 This checkpoint covers text and bounded App image projection, not reply/thread
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 3c24da32cc..a54b0f1347 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
@@ -201,6 +201,17 @@ Core 重查 Session、当前绑定与受众，provider 核验原来源。消息
 没有 provider locator 的写入，重启后仍保留未确认。撤权或身份变化阻止后续投递。
 关闭既有连接即可撤销授权；回滚须保留私有投递 journal。
 
+新核验送达的 App 用户镜像也复用已有表情回执：`Get` 表示已收到，仅在观测到 Core
+Turn 已进入 starting/running 后添加 `OnIt`；完成、失败或中断后清理处理中表情，
+保留已收到。排队不冒充处理中，App 答案不加这两种表情。长输入作用于最后一条已核验的
+实际分段消息，不对组合 parts locator 操作。Turn 独立变化即使没有追加正文，也会由
+既有五秒扫描发现。每次副作用前重查 canonical 输入/Turn、当前来源和授权，以及冻结
+provider 消息；反馈失败保留可恢复回执，不阻断正文送达或启动工作，未知 provider
+写入不重发。不伪造 incoming event，也不结算原 Lark 请求。已有已结束镜像不补写表情，
+当前活动镜像可接续此反馈。`loopx chat --no-private-reactions` 仍可关闭反馈，正文路径
+不变。隔离真实 HTTP/Core 配合合成 provider 的检查只核验源码行为；安装后的实际
+Lark 表情和 provider 呈现仍独立验收。
+
 通过上述 **继续原会话** 或明确选定的 App 管家入口使用该路径；旧普通历史继续
 独立保留。本检查点覆盖文本及有界 App 图片投影，不覆盖回复/话题关系镜像、向
 Lark 历史回插旧时间或全渠道广播。Core 中的 App 图片复用既有资源上传与 post
diff --git a/loopx/extensions/lark/inbox_reactions.py b/loopx/extensions/lark/inbox_reactions.py
index a515cd682d..b38551aee5 100644
--- a/loopx/extensions/lark/inbox_reactions.py
+++ b/loopx/extensions/lark/inbox_reactions.py
@@ -626,6 +626,7 @@ def ensure_lark_event_inbox_received_reaction_locked(
 def _ensure_reaction_creation_locked(
     *, inbox: Path, message_id: str, emoji_type: str, reaction_phase: str,
     create_reaction: ReactionCreator, delete_reaction: ReactionDeleter,
+    captured_pending: bool = True,
 ) -> dict[str, Any]:
     """One provider creation lifecycle for received and processing feedback.
 
@@ -642,7 +643,7 @@ def _ensure_reaction_creation_locked(
             status=f"already_{reaction_phase}",
             ok=True,
             configured=True,
-            captured_pending=True,
+            captured_pending=captured_pending,
         )
     operation = _load_reaction_creation_operation(inbox=inbox, message_id=message_id, reaction_phase=reaction_phase)
     if operation and operation["emoji_type"] != emoji_type:
@@ -650,7 +651,7 @@ def _ensure_reaction_creation_locked(
             status="operation_config_changed",
             ok=False,
             configured=True,
-            captured_pending=True,
+            captured_pending=captured_pending,
             blocker=f"lark_inbox_{reaction_phase}_reaction_operation_config_changed",
         )
     if operation and operation["phase"] == "prepared":
@@ -658,7 +659,7 @@ def _ensure_reaction_creation_locked(
             status="provider_outcome_uncertain",
             ok=False,
             configured=True,
-            captured_pending=True,
+            captured_pending=captured_pending,
             blocker=f"lark_inbox_{reaction_phase}_reaction_provider_outcome_uncertain",
         )
     if operation and operation["phase"] == "created":
@@ -676,7 +677,7 @@ def _ensure_reaction_creation_locked(
                 status="receipt_failed",
                 ok=False,
                 configured=True,
-                captured_pending=True,
+                captured_pending=captured_pending,
                 external_writes_performed=False,
                 blocker=f"lark_inbox_{reaction_phase}_reaction_receipt_failed",
             )
@@ -685,7 +686,7 @@ def _ensure_reaction_creation_locked(
             status="receipt_recovered",
             ok=True,
             configured=True,
-            captured_pending=True,
+            captured_pending=captured_pending,
         )
     _write_reaction_creation_operation(
         inbox=inbox,
@@ -700,7 +701,7 @@ def _ensure_reaction_creation_locked(
             status="failed",
             ok=False,
             configured=True,
-            captured_pending=True,
+            captured_pending=captured_pending,
             blocker=f"lark_inbox_{reaction_phase}_reaction_create_failed",
         )
     try:
@@ -720,7 +721,7 @@ def _ensure_reaction_creation_locked(
             status="operation_receipt_failed",
             ok=False,
             configured=True,
-            captured_pending=True,
+            captured_pending=captured_pending,
             created_count=int(not cleaned_up),
             external_writes_performed=True,
             blocker=(
@@ -742,7 +743,7 @@ def _ensure_reaction_creation_locked(
             status="receipt_failed",
             ok=False,
             configured=True,
-            captured_pending=True,
+            captured_pending=captured_pending,
             created_count=1,
             external_writes_performed=True,
             blocker=f"lark_inbox_{reaction_phase}_reaction_receipt_failed",
@@ -752,11 +753,74 @@ def _ensure_reaction_creation_locked(
         status=reaction_phase,
         ok=True,
         configured=True,
-        captured_pending=True,
+        captured_pending=captured_pending,
         created_count=1,
     )
 
 
+def _reconcile_outbound_input_reactions(
+    *, project: str | Path, config_path: str | Path, message_id: str,
+    processing: bool, complete: bool, before_write: Callable[[], bool],
+    runner: CommandRunner,
+) -> dict[str, Any]:
+    """Reuse reaction receipts for an independently verified outbound input.
+
+    This is not an Inbox capture or settlement. The private-conversation
+    caller supplies fresh canonical/source and provider-message verification.
+    """
+    config = load_lark_event_inbox_config(project=project, config_path=config_path)
+    if not config["enabled"] or not MESSAGE_ID_PATTERN.fullmatch(message_id):
+        raise ValueError("outbound feedback requires an enabled inbox and exact message")
+    if config["reply"].get("received_reaction_policy") != "retain":
+        raise ValueError("outbound feedback requires retained receipt feedback")
+    inbox = config["inbox_path"]
+    with lark_inbox_reaction_lock(inbox=inbox, message_id=message_id):
+        if not before_write():
+            raise ValueError("outbound feedback authority unavailable")
+        created_count = 0
+        writes_performed = False
+        for phase in ("received", "processing") if processing and not complete else ("received",):
+            emoji = str(config["reply"].get(f"{phase}_reaction_emoji") or "")
+            if not emoji:
+                continue
+
+            def create(target: str, value: str) -> str | None:
+                if not before_write():
+                    # This callback is reached only for a newly prepared
+                    # creation. No provider call occurred; old unknown writes
+                    # never reach it and keep their original journal.
+                    _clear_reaction_creation_operation(inbox=inbox, message_id=target, reaction_phase=phase)
+                    return None
+                return _create_reaction(runner=runner, profile=config["reply"]["sender_profile"],
+                    message_id=target, emoji_type=value)
+
+            result = _ensure_reaction_creation_locked(inbox=inbox, message_id=message_id,
+                emoji_type=emoji, reaction_phase=phase, captured_pending=False,
+                create_reaction=create,
+                delete_reaction=lambda target, reaction: before_write() and _delete_reaction(
+                    runner=runner, profile=config["reply"]["sender_profile"], message_id=target, reaction_id=reaction))
+            created_count += result["created_count"]
+            writes_performed |= result["external_writes_performed"]
+            if not result["ok"]:
+                result["created_count"] = created_count
+                result["external_writes_performed"] = writes_performed
+                return result
+        if complete:
+            def authorized_runner(args: Sequence[str]) -> Mapping[str, Any]:
+                return runner(args) if before_write() else {"returncode": 1}
+
+            result = _complete_lark_event_inbox_reactions_locked(config=config,
+                message_id=message_id, execute=True, runner=authorized_runner)
+        else:
+            result = _operation_result(operation="processing" if processing else "received",
+                status="processing" if processing else "received", ok=True, execute=True,
+                configured=bool(config["reply"].get("received_reaction_emoji")
+                    or config["reply"].get("processing_reaction_emoji")))
+        result["created_count"] += created_count
+        result["external_writes_performed"] |= writes_performed
+        return result
+
+
 def ensure_lark_event_inbox_received_reaction(
     *,
     project: str | Path,
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index 70f25f8e30..065b919211 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -29,7 +29,8 @@
 from .identity_shapes import LARK_APP_ID_PATTERN as APP_ID_PATTERN
 from .inbox_reply import _message, reply_lark_event_inbox, update_lark_inbox_reply, verify_lark_inbox_reply
 from .manager_context import manager_failure_reply
-from .inbox_reactions import mark_lark_event_inbox_processing, mark_lark_event_inbox_received
+from .inbox_reactions import (_reconcile_outbound_input_reactions,
+                             mark_lark_event_inbox_processing, mark_lark_event_inbox_received)
 from .outbound import present_lark_conversation_text as _presentation_text
 from .private_images import private_message_caption, private_message_images
 from .return_files import uploaded_chat_images, verify_chat_images
@@ -171,7 +172,7 @@ def __init__(self, *, controller: Any, runtime_root: Path, runner: Any, cli_bin:
         self.root = controller.store.root / "lark-private-deliveries"
         self._context_recovery_probe_at: dict[Path, float] = {}
         self._web_projection_scan_at = 0.0
-        self._web_projection_snapshots: dict[str, tuple[int, int]] = {}
+        self._web_projection_snapshots: dict[str, tuple[int, int, int]] = {}
         from .owner_attention import LarkOwnerAttention
         self.owner_attention = LarkOwnerAttention(self)
         from .private_ingress import PrivateIngressObserver
@@ -1328,8 +1329,17 @@ def _web_projection_paths(self) -> list[Path]:
             for sid, source in sources.items():
                 try:
                     stat = (store.sessions_root / sid / "messages.jsonl").stat()
-                    snapshot = (stat.st_mtime_ns, stat.st_size)
-                    if self._web_projection_snapshots.get(sid) == snapshot:
+                    # A queued/running/terminal transition need not append a
+                    # message. These are discovery hints, never grant proofs.
+                    snapshot = (stat.st_mtime_ns, stat.st_size,
+                        (store.sessions_root / sid / "turns").stat().st_mtime_ns if self.reaction_feedback else 0)
+                    key = identity_ref(source["binding_id"], sid)
+                    path = root / f"{key}.json"
+                    previous = _read_json(path) if self.reaction_feedback and path.exists() else {}
+                    retry_feedback = self.reaction_feedback and any(
+                        isinstance(phase.get("feedback"), dict) and not phase["feedback"].get("ok")
+                        for phase in previous.get("deliveries", {}).values())
+                    if self._web_projection_snapshots.get(sid) == snapshot and not retry_feedback:
                         continue
                     messages = store.messages(sid)
                     web_turns = {message["turn_id"] for message in messages
@@ -1337,8 +1347,6 @@ def _web_projection_paths(self) -> list[Path]:
                     selected = [message for message in messages if message.get("turn_id") in web_turns
                                 and message.get("role") in {"user", "agent"}]
                     if selected:
-                        key = identity_ref(source["binding_id"], sid)
-                        path = root / f"{key}.json"
                         with exclusive_file_lock(path, operation="discover_web_conversation_projection"):
                             row = _read_json(path) if path.exists() else {
                                 "schema_version": "lark_private_web_projection_v0",
@@ -1346,8 +1354,21 @@ def _web_projection_paths(self) -> list[Path]:
                                 "binding_id": source["binding_id"], "profile": source["profile"],
                                 "source": source["source"], "event": source["event"],
                                 "session_id": sid, "deliveries": {}, "status": "pending"}
-                            if any(not row["deliveries"].get("web." + message["message_id"], {}).get("verified")
-                                   for message in selected):
+                            needs_feedback = False
+                            if self.reaction_feedback:
+                                for message in selected:
+                                    if message["role"] != "user":
+                                        continue
+                                    state = row["deliveries"].get("web." + message["message_id"], {})
+                                    turn = store.load_turn(sid, message["turn_id"]) or {}
+                                    feedback = state.get("feedback")
+                                    # Do not backfill reactions onto old,
+                                    # already-finished verified history.
+                                    needs_feedback |= (not feedback.get("ok") or
+                                        feedback.get("turn_status") != turn.get("status")) if isinstance(feedback, dict) else (
+                                            turn.get("status") not in TERMINAL_TURN_STATES)
+                            if needs_feedback or any(not row["deliveries"].get("web." + message["message_id"], {}).get("verified")
+                                                     for message in selected):
                                 row["status"] = "pending"
                                 _atomic_write_json(path, row)
                     self._web_projection_snapshots[sid] = snapshot
@@ -1387,6 +1408,73 @@ def _web_projection_messages(self, row: dict[str, Any]) -> list[dict[str, Any]]:
                 selected.append(message)
             return selected
 
+    @staticmethod
+    def _web_projection_delivery_id(row: dict[str, Any], message: dict[str, Any]) -> str:
+        """The existing provider identity binds this exact canonical message."""
+        return hashlib.sha256(json.dumps([row["binding_id"], row["session_id"], message["message_id"]],
+            separators=(",", ":")).encode()).hexdigest()
+
+    def _web_projection_feedback(self, row: dict[str, Any], message: dict[str, Any]) -> None:
+        if not self.reaction_feedback or message["role"] != "user":
+            return
+        phase = row["deliveries"].get("web." + message["message_id"], {})
+        if not phase.get("verified"):
+            return
+        store = self.core.controller.store
+        turn = store.load_turn(row["session_id"], message["turn_id"]) or {}
+        status = turn.get("status")
+        if "feedback" not in phase and status in TERMINAL_TURN_STATES:
+            return  # Existing finished mirrors keep their frozen history.
+        prior = phase.setdefault("feedback", {})
+        if prior.get("ok") and prior.get("turn_status") == status:
+            return
+        try:
+            if message not in self._web_projection_messages(row):
+                raise ValueError("canonical App input changed")
+            expected_text = f"**你（App）**\n\n{message['text']}"
+            if phase.get("text") not in (expected_text, _presentation_text(expected_text)):
+                raise ValueError("App input no longer matches its frozen mirror")
+            source = _read_json(self.root / f"{row['request_ref']}.json")
+            if (source.get("context_recovery") or turn.get("context_recovery")
+                    or status not in {"queued", "starting", "running", *TERMINAL_TURN_STATES}):
+                raise ValueError("App feedback source/Turn unavailable")
+            config = self._inbox(row)
+            from .post_delivery import attachment_reply
+            _, target = attachment_reply(project=self.runtime_root, config_path=config,
+                attempt=phase["attempt"], text=phase["text"])
+
+            def authorized() -> bool:
+                try:
+                    if message not in self._web_projection_messages(row):
+                        return False
+                    verified = verify_lark_inbox_reply(project=self.runtime_root, config_path=config,
+                        message_id=row["event"]["message_id"], text=phase["text"], attempt=phase["attempt"],
+                        reply_placement="chat_root", delivery_id=self._web_projection_delivery_id(row, message),
+                        image_keys=tuple(phase.get("image_keys", [])), runner=self._reply_runner,
+                        finalize_reactions=False, source_authority_verifier=lambda profile, chat, ref:
+                            self._source_authority(row, profile, chat, ref))
+                    # Provider IO can outlast a grant or a Turn. Re-read both
+                    # before this exact mirror's reaction, including cleanup.
+                    if (verified.get("ok") is not True or verified.get("reply_verified") is not True
+                            or message not in self._web_projection_messages(row)):
+                        return False
+                    current_source = _read_json(self.root / f"{row['request_ref']}.json")
+                    current = store.load_turn(row["session_id"], message["turn_id"]) or {}
+                    return (current.get("status") == status
+                        and current.get("origin") == "web" and not current.get("context_recovery")
+                        and not current_source.get("context_recovery"))
+                except (OSError, ValueError, KeyError):
+                    return False
+
+            result = _reconcile_outbound_input_reactions(project=self.runtime_root, config_path=config,
+                message_id=target["message_ref"], processing=status in {"starting", "running"},
+                complete=status in TERMINAL_TURN_STATES, before_write=authorized, runner=self._reply_runner)
+            phase["feedback"] = {"ok": result.get("ok") is True, "turn_status": status,
+                "message_ref": target["message_ref"], "result": result}
+        except (OSError, ValueError, KeyError):
+            phase["feedback"] = {"ok": False, "turn_status": status,
+                "blocker": "original_web_projection_unavailable"}
+
     def _reconcile_web_projection(self, path: Path) -> int:
         # The existing worker pool serializes this outbox with a durable fence.
         # Recovery reads the same provider attempt; it never starts a Core Turn.
@@ -1397,6 +1485,8 @@ def _reconcile_web_projection(self, path: Path) -> int:
                 pending = next((message for message in messages
                     if not row["deliveries"].get("web." + message["message_id"], {}).get("verified")), None)
                 if pending is None:
+                    for message in messages:
+                        self._web_projection_feedback(row, message)
                     # Older verified placeholders retain their exact attempts.
                     # They cannot be relabelled as media delivery or resent.
                     row["status"] = "partial" if any(message.get("attachments") and not
@@ -1411,15 +1501,18 @@ def authorized() -> bool:
                         return False
                 who = "你" if pending["role"] == "user" else "LoopX"
                 text = f"**{who}（App）**\n\n{pending['text']}"
-                delivery_id = hashlib.sha256(json.dumps(
-                    [row["binding_id"], row["session_id"], pending["message_id"]],
-                    separators=(",", ":")).encode()).hexdigest()
+                if self.reaction_feedback and pending["role"] == "user":
+                    # Persist feedback intent before a mirror write/recovery;
+                    # old completed mirrors without it are not backfilled.
+                    row["deliveries"].setdefault("web." + pending["message_id"], {}).setdefault("feedback", {})
+                    _atomic_write_json(path, row)
                 delivered = self._deliver(path, row, "web." + pending["message_id"], text,
                     inbox=lambda: self._inbox(row), authorization_check=authorized,
-                    reply_placement="chat_root", delivery_id=delivery_id, finalize_source=False,
+                    reply_placement="chat_root", delivery_id=self._web_projection_delivery_id(row, pending), finalize_source=False,
                     image_attachments=pending.get("attachments"))
                 if delivered:
                     row.pop("blocker", None)
+                    self._web_projection_feedback(row, pending)
                 _atomic_write_json(path, row)
                 return int(delivered)
             except (OSError, ValueError, KeyError):
diff --git a/tests/extensions/test_lark_inbox_reactions.py b/tests/extensions/test_lark_inbox_reactions.py
index f47406cf71..1f0d9ac9aa 100644
--- a/tests/extensions/test_lark_inbox_reactions.py
+++ b/tests/extensions/test_lark_inbox_reactions.py
@@ -13,6 +13,7 @@
 from loopx.extensions.lark import inbox_reactions as inbox_reactions_module
 from loopx.extensions.lark.event_inbox import load_lark_event_inbox_config
 from loopx.extensions.lark.inbox_reactions import (
+    _reconcile_outbound_input_reactions,
     complete_lark_event_inbox_reactions,
     ensure_lark_event_inbox_received_reaction,
     lark_inbox_reaction_receipts,
@@ -128,6 +129,117 @@ def __call__(self, args: Sequence[str]) -> dict[str, Any]:
         raise AssertionError(call)
 
 
+def test_outbound_feedback_reuses_journals_without_capturing_or_settling_input(tmp_path):
+    config, inbox, project = _fixture(tmp_path)
+    settings = json.loads(config.read_text())
+    settings["reply"]["received_reaction_policy"] = "retain"
+    config.write_text(json.dumps(settings))
+    original = (inbox / "om_reaction_fixture.json").read_bytes()
+    runner = ReactionRunner()
+    target = "om_verified_app_input"
+    from concurrent.futures import ThreadPoolExecutor
+
+    def receive(_):
+        return _reconcile_outbound_input_reactions(project=project, config_path=config,
+            message_id=target, processing=False, complete=False, before_write=lambda: True, runner=runner)
+
+    with ThreadPoolExecutor(max_workers=4) as pool:
+        results = list(pool.map(receive, range(4)))
+    assert all(result["ok"] for result in results)
+    assert sum(result["created_count"] for result in results) == 1
+    assert sum(result["external_writes_performed"] for result in results) == 1
+    for processing, complete, created, deleted in ((True, False, 1, 0), (False, True, 0, 1), (False, True, 0, 0)):
+        result = _reconcile_outbound_input_reactions(project=project, config_path=config,
+            message_id=target, processing=processing, complete=complete,
+            before_write=lambda: True, runner=runner)
+        assert result["ok"]
+        assert (result["created_count"], result["deleted_count"]) == (created, deleted)
+        assert result["external_writes_performed"] == bool(created + deleted)
+    assert [json.loads(call[call.index("--data") + 1])["reaction_type"]["emoji_type"]
+            for call in runner.calls if "create" in call] == ["Get", "OnIt"]
+    assert sum("delete" in call for call in runner.calls) == 1
+    assert set(lark_inbox_reaction_receipts(inbox=inbox, message_id=target)) == {"received"}
+    assert not (inbox / f"{target}.json").exists()
+    assert not (inbox / "processed.json").exists()
+    assert (inbox / "om_reaction_fixture.json").read_bytes() == original
+    assert not lark_inbox_reaction_receipts(inbox=inbox, message_id="om_reaction_fixture")
+
+
+def test_outbound_feedback_retains_known_write_counts_when_processing_fails(tmp_path):
+    config, inbox, project = _fixture(tmp_path)
+    settings = json.loads(config.read_text())
+    settings["reply"]["received_reaction_policy"] = "retain"
+    config.write_text(json.dumps(settings))
+    runner = ReactionRunner()
+
+    def fail_processing(args):
+        if "create" in args and json.loads(args[args.index("--data") + 1])["reaction_type"]["emoji_type"] == "OnIt":
+            return {"returncode": 1}
+        return runner(args)
+
+    result = _reconcile_outbound_input_reactions(project=project, config_path=config,
+        message_id="om_app_partial_feedback", processing=True, complete=False,
+        before_write=lambda: True, runner=fail_processing)
+    assert not result["ok"]
+    assert result["created_count"] == 1 and result["external_writes_performed"]
+    assert set(lark_inbox_reaction_receipts(inbox=inbox, message_id="om_app_partial_feedback")) == {"received"}
+
+
+def test_outbound_feedback_rechecks_authority_before_creation_and_cleanup(tmp_path):
+    config, inbox, project = _fixture(tmp_path)
+    settings = json.loads(config.read_text())
+    settings["reply"]["received_reaction_policy"] = "retain"
+    config.write_text(json.dumps(settings))
+    runner = ReactionRunner()
+    checks = iter((True, False))
+    result = _reconcile_outbound_input_reactions(project=project, config_path=config,
+        message_id="om_app_authority", processing=False, complete=False,
+        before_write=lambda: next(checks), runner=runner)
+    assert not result["ok"] and runner.calls == []
+    assert list((inbox / "reactions" / "received-operations").glob("*.json")) == []
+    for processing in (False, True):
+        assert _reconcile_outbound_input_reactions(project=project, config_path=config,
+            message_id="om_app_authority", processing=processing, complete=False,
+            before_write=lambda: True, runner=runner)["ok"]
+    before = list(runner.calls)
+    # Revocation before a terminal effect cannot erase the processing receipt.
+    with pytest.raises(ValueError, match="authority"):
+        _reconcile_outbound_input_reactions(project=project, config_path=config,
+            message_id="om_app_authority", processing=False, complete=True,
+            before_write=lambda: False, runner=runner)
+    assert runner.calls == before
+    assert "processing" in lark_inbox_reaction_receipts(inbox=inbox, message_id="om_app_authority")
+
+
+@pytest.mark.parametrize("processing", [False, True])
+def test_outbound_feedback_retains_an_unknown_provider_write_without_retry(tmp_path, processing):
+    config, inbox, project = _fixture(tmp_path)
+    settings = json.loads(config.read_text())
+    settings["reply"]["received_reaction_policy"] = "retain"
+    config.write_text(json.dumps(settings))
+    runner = ReactionRunner()
+    target = "om_app_lost_receipt"
+    if processing:
+        assert _reconcile_outbound_input_reactions(project=project, config_path=config,
+            message_id=target, processing=False, complete=False,
+            before_write=lambda: True, runner=runner)["ok"]
+    calls = len(runner.calls)
+
+    def lost_reply(args):
+        result = runner(args)
+        if "create" in args:
+            raise OSError("synthetic response lost after the provider write")
+        return result
+
+    for _ in range(2):
+        result = _reconcile_outbound_input_reactions(project=project, config_path=config,
+            message_id=target, processing=processing, complete=False,
+            before_write=lambda: True, runner=lost_reply)
+        assert not result["ok"]
+    assert len(runner.calls) == calls + 1
+    assert not (inbox / f"{target}.json").exists()
+
+
 def test_received_reaction_boundary_acknowledges_pending_message_once(
     tmp_path: Path,
 ) -> None:
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index d008709a1e..bd1bf3f0d6 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -805,10 +805,105 @@ def test_app_messages_and_answers_reach_the_original_chat_once_without_new_execu
         assert len(store.list_sessions()) == 1
         # Provider echoes are Bot-authored, never a second human input.
         assert len([m for m in store.messages(sid) if m.get("origin") == "web"]) == 2
+        user_mirrors = {ref for ref, message in provider.messages.items()
+            if "你（App）" in json.dumps(message.get("body", {}), ensure_ascii=False)}
+        assert len(user_mirrors) == 2
+        assert {(profile, ref, emoji) for profile, ref, emoji in provider.reaction_creates
+                if ref in user_mirrors} == {("notes-app", ref, "Get") for ref in user_mirrors}
+        # Completed App inputs retain receipt feedback, never a processing
+        # reaction on the assistant mirror or a different original request.
+        assert {ref for _profile, ref, _emoji in provider.reaction_creates if ref.startswith("om_out_")} == user_mirrors
     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"
+    fake = ordinary[-2]
+    marker = '        if "typed terminal failure" in json.dumps(request.get("params") or {}):'
+    source = fake.read_text()
+    assert marker in source
+    fake.write_text(source.replace(marker,
+        '        from pathlib import Path\n'
+        '        if "App lifecycle" in json.dumps(request.get("params") or {}):\n'
+        f'            while {terminal != "interrupted"!r} and not Path({str(gate)!r}).exists():\n'
+        '                time.sleep(.01)\n' + marker))
+    store, runtime, provider, transport = connect(ordinary)
+    originals, turns = {}, {}
+    try:
+        for profile in ("notes-app", "steward-app"):
+            event = provider.event(profile, f"lifecycle-{profile}", "Original source")
+            transport.admit(profile, event)
+            original = next(row for row in transport.core.pending()
+                            if row["binding_id"] == transport._binding(profile)["binding_id"])
+            runtime.wait_for_turn(session_id=original["session_id"], turn_id=original["turn_id"], timeout_sec=10)
+            transport.reconcile()
+            originals[profile] = original
+        original_bytes = {row["request_ref"]: (transport.root / f"{row['request_ref']}.json").read_bytes()
+                          for row in originals.values()}
+        captured = ordinary[4].read_bytes()
+        with monkeypatch.context() as queued:
+            queued.setattr(runtime, "resume_session_queue", lambda **_: None)
+            for profile, original in originals.items():
+                text = f"App lifecycle {profile}" + (
+                    " typed terminal failure" if terminal == "failed" else
+                    " wait for interrupt" if terminal == "interrupted" else "")
+                status, turn = ordinary[3](f"/api/chat/sessions/{original['session_id']}/turns",
+                    {"message": text, "client_turn_id": f"app-lifecycle-{profile}"})
+                assert status == 202 and turn["status"] == "queued"
+                turns[profile] = turn
+        transport._web_projection_scan_at = 0
+        _drain_app_projection(transport, lambda: sum("你（App）" in text for _, text in provider.writes) == 2)
+        transport.reconcile()  # Finish body delivery while both inputs remain queued.
+        targets = {profile: next(ref for ref, row in provider.messages.items()
+            if f"App lifecycle {profile}" in json.dumps(row.get("body", {}), ensure_ascii=False)
+            and "你（App）" in json.dumps(row.get("body", {}), ensure_ascii=False)) for profile in originals}
+        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()}
+        assert ordinary[4].read_bytes() == captured  # Feedback did not start either queued input.
+        transcripts = {row["session_id"]: (store.sessions_root / row["session_id"] / "messages.jsonl").read_bytes()
+                       for row in originals.values()}
+        for profile, original in originals.items():
+            runtime.resume_session_queue(session_id=original["session_id"], work_dir=ordinary[-1], objective="")
+        deadline = time.monotonic() + 10
+        while any(store.load_turn(row["session_id"], turns[profile]["turn_id"])["status"] != "running"
+                  for profile, row in originals.items()):
+            assert time.monotonic() < deadline
+            time.sleep(.01)
+        assert all((store.sessions_root / sid / "messages.jsonl").read_bytes() == data
+                   for sid, data in transcripts.items())
+        transport._web_projection_scan_at = 0
+        _drain_app_projection(transport, lambda: all((profile, ref, "OnIt") in provider.reactions.values()
+                                                   for profile, ref in targets.items()))
+        transport.reconcile()
+        if terminal == "interrupted":
+            for profile, row in originals.items():
+                runtime.interrupt_turn(session_id=row["session_id"], turn_id=turns[profile]["turn_id"])
+        else:
+            gate.touch()
+        for profile, row in originals.items():
+            ended = runtime.wait_for_turn(session_id=row["session_id"], turn_id=turns[profile]["turn_id"], timeout_sec=10)
+            assert ended["status"] == terminal
+        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()))
+        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()
+        sent, reactions = list(provider.writes), list(provider.reaction_creates)
+        restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
+            runner=provider, cli_bin="lark-cli")
+        for _ in range(3):
+            restarted.reconcile()
+        assert provider.writes == sent and provider.reaction_creates == reactions
+        assert ordinary[4].read_bytes() == starts
+        assert all((transport.root / f"{ref}.json").read_bytes() == data for ref, data in original_bytes.items())
+    finally:
+        gate.touch()
+        runtime.close()
+
+
 def test_app_projection_recovers_original_receipt_and_revocation_blocks_remaining_messages(ordinary):  # noqa: F811
     store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary)
     provider.verify_replies = False
@@ -885,6 +980,142 @@ def test_app_projection_refuses_a_replaced_provider_identity(ordinary):  # noqa:
         runtime.close()
 
 
+@pytest.mark.parametrize("fault", ["revoked", "app", "source", "session", "turn", "context_recovery"])
+def test_app_feedback_rechecks_exact_source_and_turn_after_mirror_delivery(ordinary, monkeypatch, fault):  # noqa: F811
+    store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary)
+    deliver = transport._deliver
+    baseline = list(provider.reaction_creates)
+    starts = ordinary[4].read_bytes()
+
+    def changed(path, row, phase, text, **kwargs):
+        result = deliver(path, row, phase, text, **kwargs)
+        if result and phase.startswith("web.") and "你（App）" in text:
+            if fault == "revoked":
+                transport.bindings.disconnect(row["binding_id"], expected_revision=transport.bindings.read()["revision"])
+            elif fault == "app":
+                provider.profile_apps["notes-app"] = "cli_replaced_app"
+            elif fault == "source":
+                row["source"] = {**row["source"], "source_ref": "0" * 24}
+            elif fault == "session":
+                row["session_id"] = "missing-session"
+            elif fault == "context_recovery":
+                source_path = transport.root / f"{row['request_ref']}.json"
+                source = json.loads(source_path.read_text())
+                source["context_recovery"] = {"client_turn_id": "unsupported-recovery"}
+                source_path.write_text(json.dumps(source))
+        return result
+
+    if fault == "turn":
+        feedback = transport._web_projection_feedback
+
+        def wrong_turn(row, message):
+            # Canonical storage keeps its correct Turn; a stale delivery
+            # argument cannot redirect feedback to the original Lark input.
+            return feedback(row, {**message, "turn_id": original["turn_id"]})
+
+        monkeypatch.setattr(transport, "_web_projection_feedback", wrong_turn)
+    monkeypatch.setattr(transport, "_deliver", changed)
+    try:
+        transport._web_projection_scan_at = 0
+        transport.reconcile()
+        assert len(provider.writes) == before + 1
+        assert provider.reaction_creates == baseline
+        assert ordinary[4].read_bytes() == starts
+    finally:
+        runtime.close()
+
+
+@pytest.mark.parametrize("disabled", [False, True])
+def test_app_feedback_failure_or_opt_out_does_not_block_frozen_messages(ordinary, disabled):  # noqa: F811
+    store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary)
+    baseline = list(provider.reaction_creates)
+    transport.reaction_feedback = not disabled
+    provider.fail_reaction_create = not disabled
+    starts = ordinary[4].read_bytes()
+    try:
+        _drain_app_projection(transport, lambda: len(provider.writes) == before + 4)
+        transport.reconcile()
+        sent = list(provider.writes)
+        restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
+            runner=provider, cli_bin="lark-cli", reaction_feedback=not disabled)
+        for _ in range(3):
+            restarted.reconcile()
+        assert provider.writes == sent and provider.reaction_creates == baseline
+        assert ordinary[4].read_bytes() == starts
+        assert len(store.messages(original["session_id"])) == 6
+    finally:
+        runtime.close()
+
+
+def test_app_long_input_feedback_uses_its_last_verified_post(ordinary):  # noqa: F811
+    store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary)
+    text = "Long App input with complete source.\n" * 900 + "END_APP_INPUT"
+    sid = original["session_id"]
+    try:
+        status, turn = ordinary[3](f"/api/chat/sessions/{sid}/turns",
+            {"message": text, "client_turn_id": "long-app-feedback"})
+        assert status == 202
+        assert runtime.wait_for_turn(session_id=sid, turn_id=turn["turn_id"], timeout_sec=10)["status"] == "completed"
+        _drain_app_projection(transport, lambda: any("END_APP_INPUT" in body for _, body in provider.writes[before:]))
+        parts = [body for _, body in provider.writes[before:] if "Long App input" in body or "END_APP_INPUT" in body]
+        assert len(parts) > 1
+        assert "".join(re.sub(r"^（\d+/\d+）\n\n", "", part) for part in parts) == "**你（App）**\n\n" + text
+        refs = {ref for ref, message in provider.messages.items()
+                if "Long App input" in json.dumps(message.get("body", {})) or "END_APP_INPUT" in json.dumps(message.get("body", {}))}
+        last = next(ref for ref in refs if "END_APP_INPUT" in json.dumps(provider.messages[ref]))
+        assert [entry for entry in provider.reaction_creates if entry[1] in refs] == [("notes-app", last, "Get")]
+    finally:
+        runtime.close()
+
+
+def test_enabling_app_feedback_does_not_backfill_old_finished_mirrors(ordinary):  # noqa: F811
+    store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary)
+    transport.reaction_feedback = False
+    baseline = list(provider.reaction_creates)
+    try:
+        _drain_app_projection(transport, lambda: len(provider.writes) == before + 4)
+        transport.reconcile()
+        sent = list(provider.writes)
+        restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
+            runner=provider, cli_bin="lark-cli", reaction_feedback=True)
+        for _ in range(3):
+            restarted.reconcile()
+        assert provider.writes == sent and provider.reaction_creates == baseline
+    finally:
+        runtime.close()
+
+
+def test_app_feedback_refuses_a_changed_input_with_an_old_verified_mirror(ordinary, monkeypatch):  # noqa: F811
+    store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary)
+    deliver = transport._deliver
+    changed_target = []
+    captured = ordinary[4].read_bytes()
+
+    def amend_after_send(path, row, phase, text, **kwargs):
+        result = deliver(path, row, phase, text, **kwargs)
+        if result and "你（App）" in text and not changed_target:
+            changed_target.append(row["deliveries"][phase]["attempt"]["message_ref"])
+            transcript = store.sessions_root / original["session_id"] / "messages.jsonl"
+            messages = [json.loads(line) for line in transcript.read_text().splitlines()]
+            for message in messages:
+                if phase == "web." + message["message_id"]:
+                    message["text"] = "Changed canonical input; frozen provider body is now stale."
+            # Deliberate corruption of disposable canonical source data,
+            # not an edit API or a mutation of an active user conversation.
+            transcript.write_text("".join(json.dumps(message) + "\n" for message in messages))
+        return result
+
+    monkeypatch.setattr(transport, "_deliver", amend_after_send)
+    try:
+        _drain_app_projection(transport, lambda: len(provider.writes) == before + 4)
+        transport.reconcile()
+        assert changed_target
+        assert not any(ref == changed_target[0] for _, ref, _ in provider.reaction_creates)
+        assert ordinary[4].read_bytes() == captured
+    finally:
+        runtime.close()
+
+
 def finish_group_turn(runtime, transport, *, message):
     row = next(row for row in transport.core.pending() if row["message"] == message
                or row["message"].endswith(f"\n\nCurrent user message:\n{message}"))

From 194cd355ee934a348e486d9f522956f4791a8cb1 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 00:27:25 +0800
Subject: [PATCH 2/3] fix(lark): isolate optional App feedback from answer
 delivery

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../app-conversation-and-async-inbox-v0.md    |  5 +-
 ...p-conversation-and-async-inbox-v0.zh-CN.md |  4 +-
 .../lark/goal_topic_runtime_service.py        | 26 +++++--
 .../extensions/lark/private_conversations.py  | 74 +++++++++++--------
 tests/test_lark_private_conversations.py      | 51 +++++++++++++
 5 files changed, 124 insertions(+), 36 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 a7ed005a2f..10832fa80f 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -292,7 +292,10 @@ discovered within the existing five-second scan even without a transcript append
 Before each effect, the worker rechecks the canonical input/Turn, current source
 and grants, and frozen provider message. Feedback failures retain recoverable
 receipts without blocking message delivery or starting work. Unknown provider
-writes are not retried. No synthetic incoming event or original-request
+writes are not retried. Optional App feedback runs on at most one slot of the
+existing reply pool, outside the body outbox lock. Its final write merges only
+feedback fields, preserving concurrently delivered current and later answers.
+No synthetic incoming event or original-request
 settlement is used. Existing finished mirrors are not backfilled; current active
 mirrors can adopt the feedback path. `loopx chat --no-private-reactions` keeps
 feedback disabled without changing the message path. The isolated HTTP/Core 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 3b26f1cb74..67be26c632 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
@@ -207,7 +207,9 @@ Turn 已进入 starting/running 后添加 `OnIt`；完成、失败或中断后
 实际分段消息，不对组合 parts locator 操作。Turn 独立变化即使没有追加正文，也会由
 既有五秒扫描发现。每次副作用前重查 canonical 输入/Turn、当前来源和授权，以及冻结
 provider 消息；反馈失败保留可恢复回执，不阻断正文送达或启动工作，未知 provider
-写入不重发。不伪造 incoming event，也不结算原 Lark 请求。已有已结束镜像不补写表情，
+写入不重发。App 可选反馈最多占用既有回复池的一个槽位，在正文 outbox 锁外执行；
+回写时仅合并反馈字段，保留并发完成的当前和后续答案投递。
+不伪造 incoming event，也不结算原 Lark 请求。已有已结束镜像不补写表情，
 当前活动镜像可接续此反馈。`loopx chat --no-private-reactions` 仍可关闭反馈，正文路径
 不变。隔离真实 HTTP/Core 配合合成 provider 的检查只核验源码行为；安装后的实际
 Lark 表情和 provider 呈现仍独立验收。
diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index b29702089f..463f470e5c 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -93,6 +93,8 @@ def _reconcile_private(self) -> None:
         attention_due = 0.0
         reports: Future[int] | None = None
         reports_due = 0.0
+        feedback: Future[int] | None = None
+        feedback_due = 0.0
         ingress: Future[list[str]] | None = None
         ingress_due = time.monotonic() + 60
         wakeup_scope = getattr(self.runtime_controller, "turn_release_wakeup", None)
@@ -101,10 +103,17 @@ def _reconcile_private(self) -> None:
         with ThreadPoolExecutor(max_workers=4, thread_name_prefix="loopx-lark-private-reply") as pool, \
                 (wakeup_scope(self._delivery_wakeup) if callable(wakeup_scope) else nullcontext()):
             def step() -> None:
-                nonlocal attempted, attention, attention_due, reports, reports_due, ingress, ingress_due
+                nonlocal attempted, attention, attention_due, reports, reports_due, ingress, ingress_due, feedback, feedback_due
                 # Clear BEFORE scanning: an arrival during the scan must also
                 # wake the following wait, rather than disappear in that gap.
                 self._delivery_wakeup.clear()
+                if feedback is not None and feedback.done():
+                    try:
+                        feedback.result()
+                    except Exception:
+                        logging.getLogger(__name__).warning("App input reaction reconciliation is pending")
+                    feedback = None
+                    feedback_due = time.monotonic() + 5
                 if reports is not None and reports.done():
                     try:
                         reports.result()
@@ -160,19 +169,26 @@ def step() -> None:
                     if report_provider is not None and reports is None and time.monotonic() >= reports_due:
                         reports_due = time.monotonic() + 30
                     for path in sorted(paths, key=lambda path: attempted.get(path, 0)):
-                        if (len(pending) + int(reports is not None or reports_ready) + int(attention is not None or attention_ready)
+                        if (len(pending) + int(feedback is not None) + int(reports is not None or reports_ready) + int(attention is not None or attention_ready)
                                 + int(ingress is not None or ingress_ready) >= 4 or self._closed.is_set()):
                             break
                         if path in pending:
                             continue
                         pending[path] = pool.submit(transport.reconcile_request, path)
                         attempted[path] = attempted.get(path, 0) + 1
-                    if reports_ready and len(pending) + int(attention is not None or attention_ready) + int(ingress is not None or ingress_ready) < 4 and not self._closed.is_set():
+                    if reports_ready and len(pending) + int(feedback is not None) + int(attention is not None or attention_ready) + int(ingress is not None or ingress_ready) < 4 and not self._closed.is_set():
                         reports = pool.submit(report_provider.reconcile)
-                    if attention_ready and len(pending) + int(reports is not None) + int(ingress is not None or ingress_ready) < 4 and not self._closed.is_set():
+                    if attention_ready and len(pending) + int(feedback is not None) + int(reports is not None) + int(ingress is not None or ingress_ready) < 4 and not self._closed.is_set():
                         attention = pool.submit(owner_attention.reconcile)
-                    if ingress_ready and len(pending) + int(reports is not None) + int(attention is not None) < 4 and not self._closed.is_set():
+                    if ingress_ready and len(pending) + int(feedback is not None) + int(reports is not None) + int(attention is not None) < 4 and not self._closed.is_set():
                         ingress = pool.submit(observer.observe, cancelled=self._closed.is_set)
+                    app_feedback = getattr(transport, "reconcile_app_feedback", None)
+                    # Optional feedback uses one slot in this same bounded
+                    # pool and leaves capacity for canonical body deliveries.
+                    if (callable(app_feedback) and feedback is None and time.monotonic() >= feedback_due
+                            and len(pending) + int(reports is not None) + int(attention is not None)
+                            + int(ingress is not None) < 3 and not self._closed.is_set()):
+                        feedback = pool.submit(app_feedback)
                 except Exception:
                     logging.getLogger(__name__).warning("Private Chat delivery discovery is pending")
 
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index a03b7e3b7c..28a86a93b0 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -172,7 +172,7 @@ def __init__(self, *, controller: Any, runtime_root: Path, runner: Any, cli_bin:
         self.root = controller.store.root / "lark-private-deliveries"
         self._context_recovery_probe_at: dict[Path, float] = {}
         self._web_projection_scan_at = 0.0
-        self._web_projection_snapshots: dict[str, tuple[int, int, int]] = {}
+        self._web_projection_snapshots: dict[str, tuple[int, int]] = {}
         from .owner_attention import LarkOwnerAttention
         self.owner_attention = LarkOwnerAttention(self)
         from .private_ingress import PrivateIngressObserver
@@ -1329,17 +1329,10 @@ def _web_projection_paths(self) -> list[Path]:
             for sid, source in sources.items():
                 try:
                     stat = (store.sessions_root / sid / "messages.jsonl").stat()
-                    # A queued/running/terminal transition need not append a
-                    # message. These are discovery hints, never grant proofs.
-                    snapshot = (stat.st_mtime_ns, stat.st_size,
-                        (store.sessions_root / sid / "turns").stat().st_mtime_ns if self.reaction_feedback else 0)
+                    snapshot = (stat.st_mtime_ns, stat.st_size)
                     key = identity_ref(source["binding_id"], sid)
                     path = root / f"{key}.json"
-                    previous = _read_json(path) if self.reaction_feedback and path.exists() else {}
-                    retry_feedback = self.reaction_feedback and any(
-                        isinstance(phase.get("feedback"), dict) and not phase["feedback"].get("ok")
-                        for phase in previous.get("deliveries", {}).values())
-                    if self._web_projection_snapshots.get(sid) == snapshot and not retry_feedback:
+                    if self._web_projection_snapshots.get(sid) == snapshot:
                         continue
                     messages = store.messages(sid)
                     web_turns = {message["turn_id"] for message in messages
@@ -1354,21 +1347,8 @@ def _web_projection_paths(self) -> list[Path]:
                                 "binding_id": source["binding_id"], "profile": source["profile"],
                                 "source": source["source"], "event": source["event"],
                                 "session_id": sid, "deliveries": {}, "status": "pending"}
-                            needs_feedback = False
-                            if self.reaction_feedback:
-                                for message in selected:
-                                    if message["role"] != "user":
-                                        continue
-                                    state = row["deliveries"].get("web." + message["message_id"], {})
-                                    turn = store.load_turn(sid, message["turn_id"]) or {}
-                                    feedback = state.get("feedback")
-                                    # Do not backfill reactions onto old,
-                                    # already-finished verified history.
-                                    needs_feedback |= (not feedback.get("ok") or
-                                        feedback.get("turn_status") != turn.get("status")) if isinstance(feedback, dict) else (
-                                            turn.get("status") not in TERMINAL_TURN_STATES)
-                            if needs_feedback or any(not row["deliveries"].get("web." + message["message_id"], {}).get("verified")
-                                                     for message in selected):
+                            if any(not row["deliveries"].get("web." + message["message_id"], {}).get("verified")
+                                   for message in selected):
                                 row["status"] = "pending"
                                 _atomic_write_json(path, row)
                     self._web_projection_snapshots[sid] = snapshot
@@ -1485,8 +1465,6 @@ def _reconcile_web_projection(self, path: Path) -> int:
                 pending = next((message for message in messages
                     if not row["deliveries"].get("web." + message["message_id"], {}).get("verified")), None)
                 if pending is None:
-                    for message in messages:
-                        self._web_projection_feedback(row, message)
                     # Older verified placeholders retain their exact attempts.
                     # They cannot be relabelled as media delivery or resent.
                     row["status"] = "partial" if any(message.get("attachments") and not
@@ -1512,7 +1490,6 @@ def authorized() -> bool:
                     image_attachments=pending.get("attachments"))
                 if delivered:
                     row.pop("blocker", None)
-                    self._web_projection_feedback(row, pending)
                 _atomic_write_json(path, row)
                 return int(delivered)
             except (OSError, ValueError, KeyError):
@@ -1520,8 +1497,47 @@ def authorized() -> bool:
                 _atomic_write_json(path, row)
                 return 0
 
+    def reconcile_app_feedback(self) -> int:
+        """Reconcile optional reactions independently of the body outbox.
+
+        The live service reserves at most one existing reply worker for this
+        pass. A slow reaction must never own the body-delivery fence or prevent
+        a later canonical answer from reaching the same audience.
+        """
+        if not self.reaction_feedback:
+            return 0
+        for path in sorted((self.root / "web-projections").glob("*.json")):
+            with exclusive_file_lock(path.with_suffix(".feedback"), operation="reconcile_web_input_feedback"):
+                with exclusive_file_lock(path, operation="read_web_input_feedback"):
+                    row = _read_json(path)
+                try:
+                    messages = self._web_projection_messages(row)
+                except (OSError, ValueError, KeyError):
+                    continue
+                for message in messages:
+                    phase_key = "web." + message["message_id"]
+                    phase = row["deliveries"].get(phase_key, {})
+                    # Keep a frozen comparison without the independent
+                    # feedback field; concurrent body deliveries stay intact.
+                    frozen = {key: value for key, value in phase.items() if key != "feedback"}
+                    prior_feedback = phase.get("feedback")
+                    self._web_projection_feedback(row, message)
+                    if "feedback" not in phase or phase["feedback"] == prior_feedback:
+                        continue
+                    with exclusive_file_lock(path, operation="record_web_input_feedback"):
+                        current = _read_json(path)
+                        saved = current["deliveries"].get(phase_key, {})
+                        if (all(current.get(key) == row.get(key) for key in
+                                ("request_ref", "binding_id", "profile", "source", "event", "session_id"))
+                                and {key: value for key, value in saved.items() if key != "feedback"} == frozen):
+                            saved["feedback"] = phase["feedback"]
+                            _atomic_write_json(path, current)
+        return 0
+
     def reconcile(self) -> int:
-        return sum(self.reconcile_request(path) for path in self.pending_delivery_paths())
+        delivered = sum(self.reconcile_request(path) for path in self.pending_delivery_paths())
+        self.reconcile_app_feedback()
+        return delivered
 
     def _recovery_source_observation(self, record: dict[str, Any], selected: dict[str, Any]) -> dict[str, Any] | None:
         source = self._source_message(record, selected=selected)
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index c9243c8a11..0816fce0ae 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -3720,6 +3720,57 @@ def delayed(self, args, cwd=None, timeout=None):
 
 
 
+def test_slow_app_reaction_does_not_block_current_or_later_answers(ordinary, monkeypatch):  # noqa: F811
+    from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService
+
+    store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary)
+    blocked, release = threading.Event(), threading.Event()
+    provider_call = Provider.__call__
+
+    def held_reaction(self, args, cwd=None, timeout=None):
+        if "reactions" in args and "create" in args:
+            ref = args[args.index("--message-id") + 1]
+            if "你（App）" in json.dumps(self.messages[ref], ensure_ascii=False):
+                blocked.set()
+                assert release.wait(20)
+        return provider_call(self, args, cwd, timeout)
+
+    monkeypatch.setattr(Provider, "__call__", held_reaction)
+    starts = ordinary[4].read_bytes()
+    service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=store.root.parent,
+        runtime_controller=runtime, private_conversations=transport)
+    service.start()
+    try:
+        assert blocked.wait(8)
+        deadline = time.monotonic() + 7
+        while len(provider.writes) < before + 4:
+            assert time.monotonic() < deadline, "A held input reaction blocked a completed answer"
+            time.sleep(.01)
+        assert ordinary[4].read_bytes() == starts  # No model replay to recover delivery.
+        status, later = ordinary[3](f"/api/chat/sessions/{original['session_id']}/turns", {
+            "message": "Later App input", "client_turn_id": "later-app-input"})
+        assert status == 202
+        assert runtime.wait_for_turn(session_id=original["session_id"], turn_id=later["turn_id"],
+            timeout_sec=10)["status"] == "completed"
+        transport._web_projection_scan_at = 0
+        service._delivery_wakeup.set()
+        deadline = time.monotonic() + 6
+        while len(provider.writes) < before + 6:
+            assert time.monotonic() < deadline, "A held reaction blocked later canonical messages"
+            time.sleep(.01)
+        assert not release.is_set()
+        assert any("Later App input" in text for _, text in provider.writes[before:])
+        assert len(store.list_sessions()) == 1
+    finally:
+        release.set()
+        service.close()
+        service._delivery_thread.join(10)
+        runtime.close()
+    projection = json.loads(next((transport.root / "web-projections").glob("*.json")).read_text())
+    assert len(projection["deliveries"]) == 6
+    assert all(phase.get("verified") for phase in projection["deliveries"].values())
+
+
 def test_private_reply_workers_have_no_executor_backlog_and_stop_scheduling(tmp_path):
     import threading
     from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService

From c6d05be24d28cd346b521fdf60fd752fcf9fdeff Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 00:46:54 +0800
Subject: [PATCH 3/3] test(lark): measure held feedback after scanner startup

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 tests/test_lark_private_conversations.py | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index 0816fce0ae..9e2fe69e6a 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -3732,7 +3732,7 @@ def held_reaction(self, args, cwd=None, timeout=None):
             ref = args[args.index("--message-id") + 1]
             if "你（App）" in json.dumps(self.messages[ref], ensure_ascii=False):
                 blocked.set()
-                assert release.wait(20)
+                assert release.wait(60)
         return provider_call(self, args, cwd, timeout)
 
     monkeypatch.setattr(Provider, "__call__", held_reaction)
@@ -3741,8 +3741,11 @@ def held_reaction(self, args, cwd=None, timeout=None):
         runtime_controller=runtime, private_conversations=transport)
     service.start()
     try:
-        assert blocked.wait(8)
-        deadline = time.monotonic() + 7
+        # Scanner startup and source verification are setup, not the held-IO
+        # invariant. Start the original ten-second delivery bound only once
+        # this exact provider operation is demonstrably pending.
+        assert blocked.wait(20)
+        deadline = time.monotonic() + 10
         while len(provider.writes) < before + 4:
             assert time.monotonic() < deadline, "A held input reaction blocked a completed answer"
             time.sleep(.01)
@@ -3754,7 +3757,7 @@ def held_reaction(self, args, cwd=None, timeout=None):
             timeout_sec=10)["status"] == "completed"
         transport._web_projection_scan_at = 0
         service._delivery_wakeup.set()
-        deadline = time.monotonic() + 6
+        deadline = time.monotonic() + 10
         while len(provider.writes) < before + 6:
             assert time.monotonic() < deadline, "A held reaction blocked later canonical messages"
             time.sleep(.01)