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 8d99b74dbb..a65cc9192b 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/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 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 a8dd009309..7fef138d7f 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 @@ -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 的检查只核验源码行为；安装后的实际 diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index fcc82b4b14..494f7980a5 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/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): diff --git a/loopx/extensions/lark/private_delivery_discovery.py b/loopx/extensions/lark/private_delivery_discovery.py index b58baf6766..24794af71e 100644 --- a/loopx/extensions/lark/private_delivery_discovery.py +++ b/loopx/extensions/lark/private_delivery_discovery.py @@ -3,10 +3,10 @@ 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 @@ -14,6 +14,24 @@ 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. diff --git a/tests/test_chat_file_discovery.py b/tests/test_chat_file_discovery.py index baca091762..082f25d5fa 100644 --- a/tests/test_chat_file_discovery.py +++ b/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") diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 61989bf331..3c331d3f4a 100644 --- a/tests/test_lark_private_conversations.py +++ b/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()