perf(lark): avoid answer Turn reads during App feedback

loopx/extensions/lark/private_conversations.py

@@ -1376,7 +1376,7 @@ 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_messages(self, row: dict[str, Any]) -> list[dict[str, Any]]:
+    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")
         native = self.core.read_request(row["request_ref"])
@@ -1400,6 +1400,11 @@ def _web_projection_messages(self, row: dict[str, Any]) -> list[dict[str, Any]]:
             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")
@@ -1479,7 +1484,7 @@ def _web_projection_feedback(self, row: dict[str, Any], message: dict[str, Any])
         if prior.get("ok") and prior.get("turn_status") == status:
             return
         try:
-            if message not in self._web_projection_messages(row):
+            if message not in self._web_projection_messages(row, inputs_only=True):
                 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)):
@@ -1495,7 +1500,7 @@ def _web_projection_feedback(self, row: dict[str, Any], message: dict[str, Any])
 
             def authorized() -> bool:
                 try:
-                    if message not in self._web_projection_messages(row):
+                    if message not in self._web_projection_messages(row, inputs_only=True):
                         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"],
@@ -1506,7 +1511,7 @@ def authorized() -> bool:
                     # 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)):
+                            or message not in self._web_projection_messages(row, inputs_only=True)):
                         return False
                     current_source = _read_json(self.root / f"{row['request_ref']}.json")
                     current = store.load_turn(row["session_id"], message["turn_id"]) or {}
@@ -1581,7 +1586,7 @@ def reconcile_app_feedback(self) -> int:
                 with exclusive_file_lock(path, operation="read_web_input_feedback"):
                     row = _read_json(path)
                 try:
-                    messages = self._web_projection_messages(row)
+                    messages = self._web_projection_messages(row, inputs_only=True)
                 except (OSError, ValueError, KeyError):
                     continue
                 for message in messages:

tests/test_lark_private_feedback.py

@@ -325,6 +325,14 @@ def test_default_feedback_tracks_queue_execution_stop_and_app_isolation(ordinary
         transport.admit('notes-app', slow)
         first = transport.core.pending()[0]
         active(store, first)
+        # A queue claim precedes the host's confirmed start. Presentation must
+        # not be asserted until its canonical source event actually exists.
+        deadline = time.monotonic() + 10
+        while not any(event.get('kind') == 'agent.phase'
+                      and event.get('payload', {}).get('method') == 'turn/started'
+                      for event in store.events_after(first['session_id'], first['turn_id'], None)):
+            assert time.monotonic() < deadline
+            time.sleep(.01)
         queued = provider.event('notes-app', 'feedback_queued', 'follow-up')
         transport.admit('notes-app', queued)
         transport.reconcile()
@@ -358,6 +366,46 @@ def test_default_feedback_tracks_queue_execution_stop_and_app_isolation(ordinary
         runtime.close()
 
 
+def test_app_feedback_scan_work_tracks_inputs_not_retained_answer_chunks(ordinary, monkeypatch):  # noqa: F811
+    store, runtime, provider, transport = connect(ordinary)
+    try:
+        event = provider.event('notes-app', 'feedback-history', 'Original source')
+        transport.admit('notes-app', event)
+        source = transport.core.pending()[0]
+        sid = source['session_id']
+        runtime.wait_for_turn(session_id=sid, turn_id=source['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 input with retained answer history', 'client_turn_id': 'feedback-history'})
+        assert status == 202 and turn['status'] == 'queued'
+        for index in range(48):
+            store.append_message(sid, role='agent', turn_id=turn['turn_id'],
+                text=f'Retained answer part {index}. ' + 'Evidence stays in the body outbox. ' * 200)
+        transport._web_projection_scan_at = 0
+        transport._web_projection_paths()
+        journal = store.sessions_root / sid / 'messages.jsonl'
+        before = journal.read_bytes(), ordinary[4].read_bytes(), list(provider.writes)
+        load_turn = store.load_turn
+        reads = []
+
+        def read_turn(session_id, turn_id):
+            reads.append((session_id, turn_id))
+            return load_turn(session_id, turn_id)
+
+        monkeypatch.setattr(store, 'load_turn', read_turn)
+        transport.reconcile_app_feedback()
+        # No body has been mirrored yet, so this is discovery, not a reaction
+        # effect. One canonical input needs one Turn validation regardless of
+        # how many answer chunks the independent body outbox still owes.
+        assert reads == [(sid, turn['turn_id'])]
+        assert (journal.read_bytes(), ordinary[4].read_bytes(), provider.writes) == before
+        assert store.load_turn(sid, turn['turn_id'])['status'] == 'queued'
+    finally:
+        runtime.close()
+
+
 def test_terminal_cleanup_recovers_verified_answer_without_resend(ordinary):  # noqa: F811
     store, runtime, provider, transport = connect(ordinary)
     try: