Avoid decoding idle App feedback history in the Lark worker

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

@@ -295,6 +295,9 @@ receipts without blocking message delivery or starting work. Unknown provider
 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.
+The idle scan reuses only App message and Turn references while the local journal
+is unchanged. It reads current Turn status each pass and uses the original
+canonical input, source and provider verification before any reaction.
 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

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

@@ -209,6 +209,8 @@ Turn 已进入 starting/running 后添加 `OnIt`；完成、失败或中断后
 provider 消息；反馈失败保留可恢复回执，不阻断正文送达或启动工作，未知 provider
 写入不重发。App 可选反馈最多占用既有回复池的一个槽位，在正文 outbox 锁外执行；
 回写时仅合并反馈字段，保留并发完成的当前和后续答案投递。
+空闲扫描仅在本地日志未变化时复用 App 消息和 Turn 的引用；每轮仍读当前 Turn 状态，
+有表情待更新时沿原路径核验完整输入、来源和 provider 消息。
 不伪造 incoming event，也不结算原 Lark 请求。已有已结束镜像不补写表情，
 当前活动镜像可接续此反馈。`loopx chat --no-private-reactions` 仍可关闭反馈，正文路径
 不变。隔离真实 HTTP/Core 配合合成 provider 的检查只核验源码行为；安装后的实际

loopx/extensions/lark/private_conversations.py

@@ -1441,10 +1441,7 @@ def _web_projection_feedback(self, row: dict[str, Any], message: dict[str, Any])
         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:
+        if not self._app_feedback_due(phase, status):
             return
         try:
             if message not in self._web_projection_messages(row, inputs_only=True):
@@ -1493,6 +1490,27 @@ def authorized() -> bool:
             phase["feedback"] = {"ok": False, "turn_status": status,
                 "blocker": "original_web_projection_unavailable"}
 
+    @staticmethod
+    def _app_feedback_due(phase: dict[str, Any], status: str | None) -> bool:
+        if "feedback" not in phase:
+            return status not in TERMINAL_TURN_STATES  # Do not backfill finished mirrors.
+        prior = phase["feedback"]
+        return not (prior.get("ok") and prior.get("turn_status") == status)
+
+    def _has_app_feedback_work(self, row: dict[str, Any]) -> bool:
+        # Discovery uses only reference metadata and current Turn state. A
+        # due item still follows the complete input/source/grant checks below.
+        store = self.core.controller.store
+        sid = row["session_id"]
+        for message_id, turn_id in self._delivery_discovery.web_input_turn_refs(store, sid):
+            phase = row["deliveries"].get("web." + message_id, {})
+            if not phase.get("verified"):
+                continue
+            turn = store.load_turn(sid, turn_id)
+            if turn is not None and self._app_feedback_due(phase, turn.get("status")):
+                return True
+        return False
+
     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.
@@ -1548,6 +1566,13 @@ def reconcile_app_feedback(self) -> int:
             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:
+                    due = self._has_app_feedback_work(row)
+                except (OSError, ValueError, KeyError):
+                    # An unavailable hint cannot suppress the original reader.
+                    due = True
+                if not due:
+                    continue
                 try:
                     messages = self._web_projection_messages(row, inputs_only=True)
                 except (OSError, ValueError, KeyError):

loopx/extensions/lark/private_delivery_discovery.py

@@ -3,17 +3,35 @@
 
 from collections.abc import Iterator
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
-from ...chat_store import _read_json
-from ...capabilities.native_chat.file_discovery import file_change_fingerprint
+from ...chat_store import ChatSessionStore, _read_json
+from ...capabilities.native_chat.file_discovery import FileChatDiscovery, file_change_fingerprint
 from .private_progress import legacy_recovery_repairs
 
 
 class PrivateDeliveryDiscovery:
     def __init__(self, root: Path) -> None:
         self.root = root
         self._completed: dict[Path, tuple[tuple[int, ...], dict[str, Any], bool]] = {}
+        self._web_inputs = FileChatDiscovery()
+
+    def web_input_turn_refs(self, store: ChatSessionStore, session_id: str) -> list[tuple[str, str]]:
+        """Observe App user message/Turn references, never bodies or effect authority."""
+        path = store._session_dir(session_id) / "messages.jsonl"
+
+        def read(_: Path) -> dict[str, Any]:
+            refs = []
+            for message in store.messages(session_id, strict=True):
+                if message.get("role") == "user" and message.get("origin") == "web":
+                    message_id, turn_id = message["message_id"], message["turn_id"]
+                    if not isinstance(message_id, str) or not isinstance(turn_id, str):
+                        raise ValueError("invalid App input reference")
+                    refs.append((message_id, turn_id))
+            return {"refs": refs}
+
+        [(_, view)] = self._web_inputs.rows(path.parent, (path,), read=read, project=lambda _, value: value)
+        return cast(list[tuple[str, str]], view["refs"])
 
     def rows(self) -> Iterator[tuple[Path, dict[str, Any], bool]]:
         """Reuse only unchanged delivered-file hints, never bodies or authority.

tests/test_chat_file_discovery.py

@@ -10,6 +10,7 @@
 from loopx.chat_store import ChatSessionStore, _atomic_write_json
 from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations
 from loopx.capabilities.native_chat.file_discovery import FileChatDiscovery
+from loopx.extensions.lark.private_delivery_discovery import PrivateDeliveryDiscovery
 
 
 @pytest.fixture
@@ -76,6 +77,22 @@ def test_new_queued_input_and_active_exclusion_stay_fresh(original_store):
     assert store.queued_turns(sid) == []
 
 
+def test_web_input_references_follow_append_and_same_mtime_replacement(original_store):
+    store, sid = original_store
+    discovery = PrivateDeliveryDiscovery(store.root / "lark-private-deliveries")
+    message = store.append_message(sid, role="user", origin="web", turn_id="original", text="App body")
+    assert discovery.web_input_turn_refs(store, sid) == [(message["message_id"], "original")]
+    newer = store.append_message(sid, role="user", origin="web", turn_id="new", text="Second App input")
+    assert discovery.web_input_turn_refs(store, sid) == [(message["message_id"], "original"), (newer["message_id"], "new")]
+    journal = store.sessions_root / sid / "messages.jsonl"
+    stamp = journal.stat()
+    replacement = journal.with_suffix(".replacement")
+    replacement.write_text(json.dumps({**newer, "origin": "lark"}) + "\n")
+    os.utime(replacement, ns=(stamp.st_atime_ns, stamp.st_mtime_ns))
+    replacement.replace(journal)
+    assert discovery.web_input_turn_refs(store, sid) == []
+
+
 def test_concurrent_first_queue_observations_do_not_change_selection(original_store):
     store, sid = original_store
     _turn(store, sid, "history", "completed")

tests/test_lark_private_conversations.py

@@ -2545,7 +2545,7 @@ def test_app_long_input_feedback_uses_its_last_verified_post(ordinary):  # noqa:
         runtime.close()
 
 
-def test_enabling_app_feedback_does_not_backfill_old_finished_mirrors(ordinary):  # noqa: F811
+def test_enabling_app_feedback_does_not_backfill_old_finished_mirrors(ordinary, monkeypatch):  # noqa: F811
     store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary)
     transport.reaction_feedback = False
     baseline = list(provider.reaction_creates)
@@ -2558,6 +2558,21 @@ def test_enabling_app_feedback_does_not_backfill_old_finished_mirrors(ordinary):
         for _ in range(3):
             restarted.reconcile()
         assert provider.writes == sent and provider.reaction_creates == baseline
+        # An idle worker must not decode or authorize the full App history on
+        # every feedback tick just to find the same finished inputs again.
+        transcript = store.sessions_root / original["session_id"] / "messages.jsonl"
+        path_type = type(transcript)
+        original_read = path_type.read_bytes
+
+        def read(path):
+            assert path != transcript, "idle feedback reread the complete Chat transcript"
+            return original_read(path)
+
+        with monkeypatch.context() as idle:
+            idle.setattr(path_type, "read_bytes", read)
+            for _ in range(3):
+                restarted.reconcile_app_feedback()
+        assert provider.writes == sent and provider.reaction_creates == baseline
     finally:
         runtime.close()