diff --git a/loopx/extensions/lark/manager_reply_parts.py b/loopx/extensions/lark/manager_reply_parts.py index 63ba81ffd..f8154faa3 100644 --- a/loopx/extensions/lark/manager_reply_parts.py +++ b/loopx/extensions/lark/manager_reply_parts.py @@ -5,6 +5,12 @@ already-validated body into ordered parts, send the parts the provider has not accepted yet, and record that progress in the same durable delivery state the single-message path uses, so a retry resumes instead of re-sending. + +The counter is not the whole record. A part is counted only after the transport +accepted it, so a state that already shows every part accepted means the reader +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. """ from __future__ import annotations @@ -24,6 +30,10 @@ "本条答复超过可投递长度，上面已按顺序发送前面的部分；" "完整答复保存在 LoopX 管家会话中。" ) +PART_DELIVERY_COMPLETE_KEY = "delivery_parts_complete" +PART_DELIVERY_VERIFIED_KEY = "delivery_parts_verified" +PART_DELIVERY_INCOMPLETE = "reply_part_delivery_incomplete" +PART_DELIVERY_COMPLETION_UNVERIFIED = "reply_part_delivery_completion_unverified" def plan_manager_reply_parts(reply_text: str) -> tuple[list[str], bool]: @@ -41,6 +51,91 @@ def plan_manager_reply_parts(reply_text: str) -> tuple[list[str], bool]: return parts, truncated +def _part_verified(reply: Mapping[str, Any]) -> bool: + """Whether the provider readback confirmed this part on the channel.""" + + return bool( + reply.get("external_write_performed") is True + and reply.get("verification_performed") is True + and reply.get("reply_verified") is True + ) + + +def _part_accepted(reply: Mapping[str, Any]) -> bool: + """Whether this part may be counted as delivered. + + ``ok`` also requires the source reaction cleanup to have finished, so a part + the provider already verified can come back not-ok with a cleanup still + pending. Its text is on the channel either way: counting it is what keeps a + retry from sending the reader the same part twice, and the pending cleanup + stays the transport's own business. + """ + + return reply.get("ok") is True or _part_verified(reply) + + +def _accepted_reply_facts(reply: Mapping[str, Any]) -> dict[str, Any]: + """The durable facts of one part the provider confirmed.""" + + return { + "reply_idempotency_key": reply.get("idempotency_key"), + PART_DELIVERY_VERIFIED_KEY: _part_verified(reply), + } + + +def completed_part_delivery_receipt( + delivery_state: Mapping[str, Any], +) -> dict[str, Any] | None: + """The verified receipt for a sequence whose every part was accepted. + + Returns ``None`` unless the durable record proves both that the sequence + finished and that the provider verified the last part, so a caller never + marks a delivery verified on the strength of an unfinished or unverified + record. + """ + + if delivery_state.get(PART_DELIVERY_COMPLETE_KEY) is not True: + return None + if delivery_state.get(PART_DELIVERY_VERIFIED_KEY) is not True: + return None + key = delivery_state.get("reply_idempotency_key") + if not isinstance(key, str) or not key.startswith("sha256:"): + return None + return { + "ok": True, + "status": "sent_verified", + "idempotency_key": key, + "content_format": "text", + "external_write_performed": True, + "verification_performed": True, + "reply_verified": True, + "part_delivery_reused": True, + } + + +def part_delivery_incomplete_reason(delivery_state: Mapping[str, Any]) -> str: + """Name why a sequence stopped, when the record already says all parts went. + + A recorded counter that reached the part count without a verified + completion cannot be re-sent (the parts are already on the channel) and + cannot be settled either, so it gets its own reason instead of the generic + incomplete one. + """ + + recorded = delivery_state.get("delivery_part_count") + sent = delivery_state.get("delivery_parts_sent") + if ( + isinstance(recorded, int) + and not isinstance(recorded, bool) + and isinstance(sent, int) + and not isinstance(sent, bool) + and recorded > 0 + and sent == recorded + ): + return PART_DELIVERY_COMPLETION_UNVERIFIED + return PART_DELIVERY_INCOMPLETE + + def deliver_manager_reply_parts( *, parts: list[str], @@ -72,12 +167,20 @@ def deliver_manager_reply_parts( ): # A different split than the one on record cannot be resumed safely. sent = 0 + elif sent == len(parts): + # Every part is already on the channel. Settle from the recorded + # acceptance instead of reporting an incomplete sequence that no retry + # could ever finish (re-sending would duplicate the whole answer). + return completed_part_delivery_receipt(delivery_state) delivery_state.update( delivery_part_count=len(parts), delivery_parts_sent=sent, format_degraded=True, updated_at=datetime.now(timezone.utc).isoformat(), ) + # A sequence that is still being sent is not a complete one, even when a + # previous attempt recorded a verified completion for a different split. + delivery_state[PART_DELIVERY_COMPLETE_KEY] = False write_delivery(delivery_path, delivery_state) last: Mapping[str, Any] | None = None for index in range(sent, len(parts)): @@ -90,7 +193,7 @@ def deliver_manager_reply_parts( execute=True, runner=reply_runner, ) - if not last.get("ok"): + if not _part_accepted(last): delivery_state.update( delivery_parts_sent=index, last_delivery_status=str(last.get("status") or "reply_failed"), @@ -100,7 +203,11 @@ def deliver_manager_reply_parts( return None delivery_state.update( delivery_parts_sent=index + 1, - reply_idempotency_key=last.get("idempotency_key"), + **( + {PART_DELIVERY_COMPLETE_KEY: True, **_accepted_reply_facts(last)} + if index + 1 == len(parts) + else _accepted_reply_facts(last) + ), updated_at=datetime.now(timezone.utc).isoformat(), ) write_delivery(delivery_path, delivery_state) @@ -142,7 +249,9 @@ def deliver_manager_reply_after_length_failure( message_id=message_id, content_format="text", ) - return reply, (None if reply is not None else "reply_part_delivery_incomplete") + return reply, ( + None if reply is not None else part_delivery_incomplete_reason(delivery_state) + ) def manager_part_delivery_pending_result( diff --git a/tests/extensions/test_lark_goal_topic_runtime.py b/tests/extensions/test_lark_goal_topic_runtime.py index 6c0daeb77..bcc3f3698 100644 --- a/tests/extensions/test_lark_goal_topic_runtime.py +++ b/tests/extensions/test_lark_goal_topic_runtime.py @@ -14,6 +14,7 @@ from loopx.extensions.lark.manager_reply_parts import ( MANAGER_REPLY_MAX_PARTS, MANAGER_REPLY_OVERFLOW_NOTE, + PART_DELIVERY_COMPLETE_KEY, ) from loopx.extensions.lark.event_collector import _jq_projection from loopx.extensions.lark.event_inbox import inspect_lark_event_inbox @@ -2797,3 +2798,118 @@ def no_duplicate_reply(args: list[str]) -> dict[str, Any]: assert second["status"] == "replied_and_acknowledged" assert second["saved_response_reused"] is True assert delivery_calls == [(proposal_id,), (proposal_id,)] + + +def test_a_fully_sent_part_sequence_settles_after_an_interrupted_receipt_write( + tmp_path, monkeypatch, +): + """The reader already has every part; the delivery must stop saying pending. + + The part loop advances the durable counter only after the provider accepted + a part, so a stop between the last part and the caller's own receipt used to + leave the answer reported as an incomplete sequence forever: every retry + re-sent nothing and the source was never acknowledged. + """ + + from loopx.extensions.lark import goal_topic_runtime as runtime + from loopx.extensions.lark.manager_reply_delivery import delivery_path + + 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 + # Large enough that the provider refuses the markdown body and the answer + # has to be delivered as a bounded part sequence. + 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), + } + + 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": _reply_runner(state), + } + real_write = runtime._write_manager_delivery + + def interrupted_receipt_write(path, payload): + if payload.get("status") == "sent_verified": + raise OSError("receipt write interrupted") + return real_write(path, payload) + + monkeypatch.setattr(runtime, "_write_manager_delivery", interrupted_receipt_write) + first = runtime.process_lark_goal_topic_event(**kwargs) + + assert first["status"] == "reply_delivery_receipt_unavailable" + assert first["source_acknowledged"] is False + 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["status"] == "pending" + assert saved["delivery_part_count"] == MANAGER_REPLY_MAX_PARTS + assert saved["delivery_parts_sent"] == MANAGER_REPLY_MAX_PARTS + assert saved[PART_DELIVERY_COMPLETE_KEY] is True + assert MANAGER_REPLY_OVERFLOW_NOTE in state["reply_text"] + delivered_once = [ + call for call in state["calls"] + if "+messages-reply" in call and "--dry-run" not in call + ] + assert len(delivered_once) == MANAGER_REPLY_MAX_PARTS + + monkeypatch.setattr(runtime, "_write_manager_delivery", real_write) + second = runtime.process_lark_goal_topic_event(**kwargs) + + # The whole answer was already on the channel, so the retry re-sends + # nothing, settles the delivery and acknowledges the source. + assert second["ok"] is True + assert second["status"] == "replied_and_acknowledged" + assert second["saved_response_reused"] is True + assert len(answered) == 1 + assert [ + call for call in state["calls"] + if "+messages-reply" in call and "--dry-run" not in call + ] == delivered_once + settled = json.loads(state_path.read_text()) + assert settled["status"] == "acknowledged" + assert settled["reply_verified"] is True + assert settled["delivery_parts_sent"] == MANAGER_REPLY_MAX_PARTS + pending = inspect_lark_event_inbox(project=kwargs["runtime_root"], + config_path=config_path) + assert pending["items"] == [] diff --git a/tests/extensions/test_lark_manager_reply_parts.py b/tests/extensions/test_lark_manager_reply_parts.py new file mode 100644 index 000000000..fa101cf07 --- /dev/null +++ b/tests/extensions/test_lark_manager_reply_parts.py @@ -0,0 +1,228 @@ +"""The part sequence must resume, settle, and never double-send an answer.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +import loopx.extensions.lark.manager_reply_parts as parts_module +from loopx.extensions.lark.manager_reply_parts import ( + PART_DELIVERY_COMPLETE_KEY, + PART_DELIVERY_COMPLETION_UNVERIFIED, + PART_DELIVERY_INCOMPLETE, + PART_DELIVERY_VERIFIED_KEY, + completed_part_delivery_receipt, + deliver_manager_reply_parts, + part_delivery_incomplete_reason, + plan_manager_reply_parts, +) + + +BODY = "\n".join(f"- 条目 {index} " + "x" * 60 for index in range(1, 200)) + + +@pytest.fixture +def delivery(tmp_path: Path): + """A fake provider boundary plus the durable state a retry would reload.""" + + state: dict = {} + writes: list[dict] = [] + sends: list[str] = [] + + def install(*, fail_at: int | None = None, verified: bool = True): + def fake_reply(**kwargs): + sends.append(kwargs["text"]) + if fail_at is not None and len(sends) == fail_at: + return {"ok": False, "status": "reply_failed", "idempotency_key": None} + return { + "ok": True, + "status": "sent_verified", + "idempotency_key": f"sha256:part-{len(sends)}", + "content_format": kwargs.get("content_format", "text"), + "external_write_performed": True, + "verification_performed": True, + "reply_verified": verified, + } + + return fake_reply + + def deliver(parts: list[str]): + return deliver_manager_reply_parts( + parts=parts, + delivery_state=state, + delivery_path=tmp_path / "delivery.json", + write_delivery=lambda path, payload: writes.append( + json.loads(json.dumps(payload)) + ), + reply_runner=object(), + root=tmp_path, + config_path=tmp_path / "config.json", + message_id="om_fixture", + content_format="text", + ) + + return { + "state": state, + "writes": writes, + "sends": sends, + "install": install, + "deliver": deliver, + } + + +def test_a_partially_accepted_sequence_resumes_at_the_first_unsent_part( + monkeypatch, delivery +): + parts, _ = plan_manager_reply_parts(BODY) + assert len(parts) > 3 + + monkeypatch.setattr( + parts_module, "reply_lark_event_inbox", delivery["install"](fail_at=3) + ) + assert delivery["deliver"](parts) is None + assert delivery["state"]["delivery_parts_sent"] == 2 + + accepted = list(delivery["sends"]) + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", delivery["install"]()) + assert delivery["deliver"](parts)["ok"] is True + + # The retry starts at the first part the provider never accepted and never + # repeats a part the reader already has. + assert delivery["sends"][len(accepted) :][0] == parts[2] + assert delivery["state"]["delivery_parts_sent"] == len(parts) + assert delivery["state"][PART_DELIVERY_COMPLETE_KEY] is True + assert delivery["state"][PART_DELIVERY_VERIFIED_KEY] is True + assert delivery["state"]["reply_idempotency_key"].startswith("sha256:") + + +def test_an_already_delivered_sequence_settles_from_the_record(monkeypatch, delivery): + parts, _ = plan_manager_reply_parts(BODY) + delivery["state"].update( + delivery_part_count=len(parts), + delivery_parts_sent=len(parts), + reply_idempotency_key="sha256:last-part", + **{ + PART_DELIVERY_COMPLETE_KEY: True, + PART_DELIVERY_VERIFIED_KEY: True, + }, + ) + # The provider must not be asked to send anything again. + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", delivery["install"]()) + + receipt = delivery["deliver"](parts) + + assert delivery["sends"] == [] + assert receipt == { + "ok": True, + "status": "sent_verified", + "idempotency_key": "sha256:last-part", + "content_format": "text", + "external_write_performed": True, + "verification_performed": True, + "reply_verified": True, + "part_delivery_reused": True, + } + + +def test_a_complete_but_unverified_record_reports_its_own_reason(monkeypatch, delivery): + parts, _ = plan_manager_reply_parts(BODY) + delivery["state"].update( + delivery_part_count=len(parts), + delivery_parts_sent=len(parts), + reply_idempotency_key="sha256:last-part", + **{PART_DELIVERY_COMPLETE_KEY: True}, + ) + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", delivery["install"]()) + + assert delivery["deliver"](parts) is None + # Nothing is re-sent, and the reason says why the record cannot settle. + assert delivery["sends"] == [] + assert ( + part_delivery_incomplete_reason(delivery["state"]) + == PART_DELIVERY_COMPLETION_UNVERIFIED + ) + + +def test_a_changed_split_restarts_instead_of_resuming_mid_answer(monkeypatch, delivery): + parts, _ = plan_manager_reply_parts(BODY) + delivery["state"].update( + delivery_part_count=len(parts) + 1, + delivery_parts_sent=len(parts) + 1, + reply_idempotency_key="sha256:another-split", + **{ + PART_DELIVERY_COMPLETE_KEY: True, + PART_DELIVERY_VERIFIED_KEY: True, + }, + ) + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", delivery["install"]()) + + assert delivery["deliver"](parts)["ok"] is True + + assert delivery["sends"][0] == parts[0] + assert delivery["state"]["delivery_part_count"] == len(parts) + assert delivery["state"]["delivery_parts_sent"] == len(parts) + + +def test_an_unkeyed_or_unfinished_record_never_claims_verification(): + assert ( + completed_part_delivery_receipt( + { + PART_DELIVERY_COMPLETE_KEY: True, + PART_DELIVERY_VERIFIED_KEY: True, + "reply_idempotency_key": "reply-fixture", + } + ) + is None + ) + assert ( + completed_part_delivery_receipt( + {PART_DELIVERY_COMPLETE_KEY: True, "reply_idempotency_key": "sha256:x"} + ) + is None + ) + assert part_delivery_incomplete_reason({}) == PART_DELIVERY_INCOMPLETE + + +def test_a_verified_part_with_pending_cleanup_is_never_sent_twice(monkeypatch, delivery): + """``ok`` also requires the reaction cleanup, which must not re-send a part. + + A part the provider already read back is on the channel; stopping the + sequence on a pending cleanup makes the retry post the same text again. + """ + + parts, _ = plan_manager_reply_parts(BODY) + + def cleanup_pending_then_ok(**kwargs): + delivery["sends"].append(kwargs["text"]) + index = len(delivery["sends"]) + facts = { + "external_write_performed": True, + "verification_performed": True, + "reply_verified": True, + } + if index == 1: + return { + "ok": False, + "status": "sent_verified_cleanup_pending", + "idempotency_key": "sha256:part-1", + **facts, + } + return { + "ok": True, + "status": "sent_verified", + "idempotency_key": f"sha256:part-{index}", + **facts, + } + + monkeypatch.setattr( + parts_module, "reply_lark_event_inbox", cleanup_pending_then_ok + ) + + receipt = delivery["deliver"](parts) + + assert receipt["ok"] is True + assert delivery["sends"].count(parts[0]) == 1 + assert delivery["sends"][1] == parts[1] + assert delivery["state"]["delivery_parts_sent"] == len(parts)