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 8c8718be7c..10832fa80f 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,26 @@ 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. 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 +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 141cb240ed..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 @@ -201,6 +201,19 @@ Core 重查 Session、当前绑定与受众，provider 核验原来源。消息 没有 provider locator 的写入，重启后仍保留未确认。撤权或身份变化阻止后续投递。 关闭既有连接即可撤销授权；回滚须保留私有投递 journal。 +新核验送达的 App 用户镜像也复用已有表情回执：`Get` 表示已收到，仅在观测到 Core +Turn 已进入 starting/running 后添加 `OnIt`；完成、失败或中断后清理处理中表情， +保留已收到。排队不冒充处理中，App 答案不加这两种表情。长输入作用于最后一条已核验的 +实际分段消息，不对组合 parts locator 操作。Turn 独立变化即使没有追加正文，也会由 +既有五秒扫描发现。每次副作用前重查 canonical 输入/Turn、当前来源和授权，以及冻结 +provider 消息；反馈失败保留可恢复回执，不阻断正文送达或启动工作，未知 provider +写入不重发。App 可选反馈最多占用既有回复池的一个槽位，在正文 outbox 锁外执行； +回写时仅合并反馈字段，保留并发完成的当前和后续答案投递。 +不伪造 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/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/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 4cdaec5de9..28a86a93b0 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 @@ -1329,6 +1330,8 @@ def _web_projection_paths(self) -> list[Path]: try: stat = (store.sessions_root / sid / "messages.jsonl").stat() snapshot = (stat.st_mtime_ns, stat.st_size) + key = identity_ref(source["binding_id"], sid) + path = root / f"{key}.json" if self._web_projection_snapshots.get(sid) == snapshot: continue messages = store.messages(sid) @@ -1337,8 +1340,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", @@ -1387,6 +1388,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. @@ -1411,12 +1479,14 @@ 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) @@ -1427,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/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 8c73459308..9e2fe69e6a 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -971,10 +971,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 @@ -1051,6 +1146,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}")) @@ -3489,6 +3720,60 @@ 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(60) + 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: + # 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) + 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() + 10 + 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