diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 051377abe4..679fd4e7f5 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/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: diff --git a/tests/test_lark_private_feedback.py b/tests/test_lark_private_feedback.py index ed5c8c61c7..36eb1e03be 100644 --- a/tests/test_lark_private_feedback.py +++ b/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: