diff --git a/loopx/extensions/lark/manager_reply_parts.py b/loopx/extensions/lark/manager_reply_parts.py index f8154faa3..6be3ead7d 100644 --- a/loopx/extensions/lark/manager_reply_parts.py +++ b/loopx/extensions/lark/manager_reply_parts.py @@ -11,6 +11,10 @@ has the whole answer even when the caller never got to write its own receipt (a failed settle write, or a process that stopped right after the last part). That case settles from the record instead of re-sending nothing forever. + +A send the provider accepted but did not read back leaves its provider locator +in the same record. The next attempt verifies that locator before sending the +part again, so an ambiguous send is reconciled instead of repeated. """ from __future__ import annotations @@ -19,7 +23,7 @@ from pathlib import Path from typing import Any, Mapping -from .inbox_reply import reply_lark_event_inbox +from .inbox_reply import reply_lark_event_inbox, verify_lark_inbox_reply from .outbound import DEFAULT_LARK_TEXT_LIMIT, split_lark_outbound_text # An oversized answer is delivered as a bounded sequence rather than a flood: @@ -32,6 +36,7 @@ ) PART_DELIVERY_COMPLETE_KEY = "delivery_parts_complete" PART_DELIVERY_VERIFIED_KEY = "delivery_parts_verified" +PART_ATTEMPT_KEY = "delivery_part_attempt" PART_DELIVERY_INCOMPLETE = "reply_part_delivery_incomplete" PART_DELIVERY_COMPLETION_UNVERIFIED = "reply_part_delivery_completion_unverified" @@ -52,11 +57,16 @@ def plan_manager_reply_parts(reply_text: str) -> tuple[list[str], bool]: def _part_verified(reply: Mapping[str, Any]) -> bool: - """Whether the provider readback confirmed this part on the channel.""" + """Whether the provider reported this part present on the channel. + + A part can be confirmed either by the readback that follows its own send or + by the reconciliation of an earlier send the provider accepted but did not + read back. Both mean the reader has that text, which is the fact the counter + records; a reconciled part has no new write of its own. + """ return bool( - reply.get("external_write_performed") is True - and reply.get("verification_performed") is True + reply.get("verification_performed") is True and reply.get("reply_verified") is True ) @@ -136,6 +146,52 @@ def part_delivery_incomplete_reason(delivery_state: Mapping[str, Any]) -> str: return PART_DELIVERY_INCOMPLETE +def recorded_part_attempt( + delivery_state: Mapping[str, Any], index: int +) -> Mapping[str, Any] | None: + """The provider locator of the part that was attempted but not confirmed.""" + + recorded = delivery_state.get(PART_ATTEMPT_KEY) + if not isinstance(recorded, Mapping) or recorded.get("index") != index: + return None + attempt = recorded.get("attempt") + return attempt if isinstance(attempt, Mapping) else None + + +def reconciled_part_reply( + *, + parts: list[str], + index: int, + delivery_state: Mapping[str, Any], + reply_runner: Any, + root: Path, + config_path: Path, + message_id: str, +) -> Mapping[str, Any] | None: + """Confirm a previously attempted part instead of sending it twice. + + Returns the verification result when the provider still reports the part on + the channel, and ``None`` when there is nothing to reconcile or the provider + could not confirm it (the caller then sends the part, as before). The + verification performs no write of its own. + """ + + attempt = recorded_part_attempt(delivery_state, index) + if attempt is None: + return None + verified = verify_lark_inbox_reply( + project=root, + config_path=config_path, + message_id=message_id, + text=parts[index], + attempt=attempt, + runner=reply_runner, + ) + if verified.get("reply_verified") is not True: + return None + return {**dict(verified), "part_reconciled": True} + + def deliver_manager_reply_parts( *, parts: list[str], @@ -184,15 +240,40 @@ def deliver_manager_reply_parts( write_delivery(delivery_path, delivery_state) last: Mapping[str, Any] | None = None for index in range(sent, len(parts)): - last = reply_lark_event_inbox( - project=root, + last = reconciled_part_reply( + parts=parts, + index=index, + delivery_state=delivery_state, + reply_runner=reply_runner, + root=root, config_path=config_path, message_id=message_id, - text=parts[index], - content_format=content_format, - execute=True, - runner=reply_runner, ) + if last is None: + # The locator of the part being sent now replaces any older one, so + # the record always points at the most recent unconfirmed attempt. + delivery_state.pop(PART_ATTEMPT_KEY, None) + + def record_attempt(attempt: Mapping[str, Any], *, index=index) -> None: + delivery_state[PART_ATTEMPT_KEY] = { + "index": index, + "attempt": dict(attempt), + } + delivery_state["updated_at"] = datetime.now( + timezone.utc + ).isoformat() + write_delivery(delivery_path, delivery_state) + + last = reply_lark_event_inbox( + project=root, + config_path=config_path, + message_id=message_id, + text=parts[index], + content_format=content_format, + execute=True, + runner=reply_runner, + delivery_attempt_recorder=record_attempt, + ) if not _part_accepted(last): delivery_state.update( delivery_parts_sent=index, @@ -201,6 +282,7 @@ def deliver_manager_reply_parts( ) write_delivery(delivery_path, delivery_state) return None + delivery_state.pop(PART_ATTEMPT_KEY, None) delivery_state.update( delivery_parts_sent=index + 1, **( diff --git a/tests/extensions/test_lark_goal_topic_runtime.py b/tests/extensions/test_lark_goal_topic_runtime.py index bcc3f3698..0de07b6df 100644 --- a/tests/extensions/test_lark_goal_topic_runtime.py +++ b/tests/extensions/test_lark_goal_topic_runtime.py @@ -2913,3 +2913,115 @@ def interrupted_receipt_write(path, payload): pending = inspect_lark_event_inbox(project=kwargs["runtime_root"], config_path=config_path) assert pending["items"] == [] + + +def test_a_part_whose_readback_failed_is_reconciled_instead_of_sent_twice( + tmp_path, monkeypatch, +): + """An ambiguous part send must be confirmed, not repeated. + + The provider accepted the first part and returned a message id, but its + readback could not confirm it. Re-sending that part would show the reader the + same text twice, so the retry verifies the recorded provider locator first. + """ + + from loopx.extensions.lark import goal_topic_runtime as runtime + from loopx.extensions.lark.manager_reply_delivery import delivery_path + from loopx.extensions.lark.manager_reply_parts import PART_ATTEMPT_KEY + + target_path, binding_path = tmp_path / "targets.json", tmp_path / "bindings.json" + _seed_legacy_topic(target_path, binding_path) + original_decide = runtime.decide_lark_topic_event + body = "测" * 60000 + + def manager_decision(**kwargs): + result = original_decide(**kwargs) + result["route"].update( + conversation_kind="manager", ingress_mode="session_queue", + authority_mode="turn_authorized", + event_id=kwargs["event"]["event_id"], + connector={"response_policy": "topic_reply"}, + ) + return result + + monkeypatch.setattr(runtime, "decide_lark_topic_event", manager_decision) + monkeypatch.setattr( + runtime, + "ensure_lark_event_inbox_received_reaction", + lambda **kw: {"ok": True, "status": "already_received"}, + ) + state: dict[str, Any] = {} + answered: list[str] = [] + + def answer(route, text): + answered.append(text) + return { + "response_text": body, + "effect_receipt": runtime._session_turn_effect(route), + } + + working_runner = _reply_runner(state) + missing_readbacks: list[str] = [] + + def ambiguous_runner(args: list[str]) -> dict[str, Any]: + if "+messages-mget" in args and not missing_readbacks: + missing_readbacks.append("om_reply_fixture") + return { + "returncode": 0, + "stdout": json.dumps({"data": {"items": []}}), + "stderr": "", + } + return working_runner(args) + + kwargs = { + "target_payload": read_goal_channel_targets(target_path), + "binding_payloads": {"goal-alpha": read_goal_channel_binding(binding_path)}, + "event": { + "event_id": "evt_incoming", + "message_id": "om_incoming", + "chat_id": "oc_public_fixture", + "root_id": "om_topic_alpha", + "create_time": "2026-08-14T21:00:00Z", + "content": "@linkmacbot report", + "mentioned": True, + "sender_type": "user", + }, + "runtime_root": tmp_path / "runtime", + "answer": answer, + "reply_runner": ambiguous_runner, + } + first = runtime.process_lark_goal_topic_event(**kwargs) + + assert first["status"] == "reply_delivery_pending" + config_path = Path(first["inbox_config_ref"]) + state_path = delivery_path( + project=kwargs["runtime_root"], config_path=config_path, + message_id="om_incoming", + ) + saved = json.loads(state_path.read_text()) + assert saved["delivery_parts_sent"] == 0 + assert saved[PART_ATTEMPT_KEY]["index"] == 0 + sent_once = [ + call[call.index("--text") + 1] + for call in state["calls"] + if "+messages-reply" in call and "--dry-run" not in call + ] + assert len(sent_once) == 1 + + kwargs["reply_runner"] = working_runner + second = runtime.process_lark_goal_topic_event(**kwargs) + + assert second["ok"] is True + assert second["status"] == "replied_and_acknowledged" + assert len(answered) == 1 + sent_total = [ + call[call.index("--text") + 1] + for call in state["calls"] + if "+messages-reply" in call and "--dry-run" not in call + ] + assert sent_total.count(sent_once[0]) == 1 + assert len(sent_total) == MANAGER_REPLY_MAX_PARTS + settled = json.loads(state_path.read_text()) + assert settled["status"] == "acknowledged" + assert settled["delivery_parts_sent"] == MANAGER_REPLY_MAX_PARTS + assert PART_ATTEMPT_KEY not in settled diff --git a/tests/extensions/test_lark_manager_reply_parts.py b/tests/extensions/test_lark_manager_reply_parts.py index fa101cf07..eabfc7b30 100644 --- a/tests/extensions/test_lark_manager_reply_parts.py +++ b/tests/extensions/test_lark_manager_reply_parts.py @@ -9,6 +9,7 @@ import loopx.extensions.lark.manager_reply_parts as parts_module from loopx.extensions.lark.manager_reply_parts import ( + PART_ATTEMPT_KEY, PART_DELIVERY_COMPLETE_KEY, PART_DELIVERY_COMPLETION_UNVERIFIED, PART_DELIVERY_INCOMPLETE, @@ -17,6 +18,7 @@ deliver_manager_reply_parts, part_delivery_incomplete_reason, plan_manager_reply_parts, + recorded_part_attempt, ) @@ -226,3 +228,137 @@ def cleanup_pending_then_ok(**kwargs): assert delivery["sends"].count(parts[0]) == 1 assert delivery["sends"][1] == parts[1] assert delivery["state"]["delivery_parts_sent"] == len(parts) + + +ATTEMPT = { + "schema_version": "manager_return_delivery_attempt_v0", + "provider": "lark", + "message_ref": "om_reply_fixture", + "intent_digest": "sha256:" + "a" * 64, + "provider_receipt": "sha256:" + "b" * 64, +} + + +def test_a_send_without_a_readback_records_its_provider_locator( + monkeypatch, delivery, +): + """An unverified send must leave the locator a later attempt can check.""" + + parts, _ = plan_manager_reply_parts(BODY) + + def unverified_send(**kwargs): + delivery["sends"].append(kwargs["text"]) + recorder = kwargs.get("delivery_attempt_recorder") + if recorder is not None: + recorder(dict(ATTEMPT)) + return { + "ok": False, + "status": "sent_unverified", + "idempotency_key": ATTEMPT["provider_receipt"], + "external_write_performed": True, + "verification_performed": False, + "reply_verified": False, + } + + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", unverified_send) + + assert delivery["deliver"](parts) is None + + assert delivery["state"]["delivery_parts_sent"] == 0 + assert delivery["state"][PART_ATTEMPT_KEY] == {"index": 0, "attempt": ATTEMPT} + assert recorded_part_attempt(delivery["state"], 0) == ATTEMPT + # A different part has no locator to reconcile. + assert recorded_part_attempt(delivery["state"], 1) is None + + +def test_a_recorded_locator_is_confirmed_instead_of_sending_the_part_again( + monkeypatch, delivery, +): + """The reader must not receive the same part twice after an ambiguous send.""" + + parts, _ = plan_manager_reply_parts(BODY) + delivery["state"].update( + delivery_part_count=len(parts), + delivery_parts_sent=0, + **{PART_ATTEMPT_KEY: {"index": 0, "attempt": ATTEMPT}}, + ) + verified_with: list[dict] = [] + + def readback(**kwargs): + verified_with.append(dict(kwargs)) + return { + "ok": True, + "verification_performed": True, + "reply_verified": True, + "part_reconciled": True, + } + + monkeypatch.setattr(parts_module, "verify_lark_inbox_reply", readback) + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", delivery["install"]()) + + assert delivery["deliver"](parts)["ok"] is True + + # No write happened for the reconciled part, and the sequence continued at + # the part after it. + assert verified_with[0]["attempt"] == ATTEMPT + assert verified_with[0]["text"] == parts[0] + assert delivery["sends"][0] == parts[1] + assert delivery["sends"].count(parts[0]) == 0 + assert delivery["state"]["delivery_parts_sent"] == len(parts) + assert PART_ATTEMPT_KEY not in delivery["state"] + + +def test_an_unconfirmed_locator_still_sends_the_part(monkeypatch, delivery): + """A locator the provider cannot confirm must not drop the reader's text.""" + + parts, _ = plan_manager_reply_parts(BODY) + delivery["state"].update( + delivery_part_count=len(parts), + delivery_parts_sent=0, + **{PART_ATTEMPT_KEY: {"index": 0, "attempt": ATTEMPT}}, + ) + + monkeypatch.setattr( + parts_module, + "verify_lark_inbox_reply", + lambda **kwargs: { + "ok": False, + "verification_performed": True, + "reply_verified": False, + "blocker": "provider_message_missing", + }, + ) + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", delivery["install"]()) + + assert delivery["deliver"](parts)["ok"] is True + + assert delivery["sends"][0] == parts[0] + + +def test_a_confirmed_locator_settles_a_sequence_with_no_new_write( + monkeypatch, delivery, +): + parts, _ = plan_manager_reply_parts("短答复") + assert len(parts) == 1 + delivery["state"].update( + delivery_part_count=1, + delivery_parts_sent=0, + **{PART_ATTEMPT_KEY: {"index": 0, "attempt": ATTEMPT}}, + ) + monkeypatch.setattr( + parts_module, + "verify_lark_inbox_reply", + lambda **kwargs: { + "ok": True, + "verification_performed": True, + "reply_verified": True, + "part_reconciled": True, + }, + ) + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", delivery["install"]()) + + assert delivery["deliver"](parts)["ok"] is True + + assert delivery["sends"] == [] + assert delivery["state"][PART_DELIVERY_COMPLETE_KEY] is True + assert delivery["state"][PART_DELIVERY_VERIFIED_KEY] is True