From 0f088664294b3464da19730d10a46dbec92a9d9b Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Mon, 21 Sep 2026 20:33:35 +0800 Subject: [PATCH 1/3] fix(lark): tell the reader when a manager answer stalls mid-sequence An over-limit manager answer is delivered as ordered parts, and the note that says where the full answer lives only travels with the last part. When the provider keeps rejecting the next part, the reader is left holding the leading fragments with nothing that says the answer was cut off. Count the stalls in the delivery record and, once the sequence has failed more than twice, post one bounded notice naming how many parts went and that the rest is still retried. The count is persisted on every failed attempt because a retry reloads the record from disk, and it is dropped when the record no longer describes this split. Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- loopx/extensions/lark/manager_reply_parts.py | 74 +++++++++++++++++++- 1 file changed, 72 insertions(+), 2 deletions(-) diff --git a/loopx/extensions/lark/manager_reply_parts.py b/loopx/extensions/lark/manager_reply_parts.py index 2d25a3a2f..ac6104326 100644 --- a/loopx/extensions/lark/manager_reply_parts.py +++ b/loopx/extensions/lark/manager_reply_parts.py @@ -37,8 +37,18 @@ PART_DELIVERY_COMPLETE_KEY = "delivery_parts_complete" PART_DELIVERY_VERIFIED_KEY = "delivery_parts_verified" PART_ATTEMPT_KEY = "delivery_part_attempt" +PART_STALL_NOTICE_KEY = "delivery_part_stall_notice" +PART_STALL_COUNT_KEY = "delivery_part_stall_count" PART_DELIVERY_INCOMPLETE = "reply_part_delivery_incomplete" PART_DELIVERY_COMPLETION_UNVERIFIED = "reply_part_delivery_completion_unverified" +# The notice is a last resort, not the outcome: the sequence counts how often it +# stopped mid-answer and only speaks after that happened more than once, so a +# transient provider hiccup never turns into a message the reader did not need. +PART_STALL_NOTICE_MIN_STALLS = 3 +MANAGER_REPLY_STALL_NOTICE = ( + "本条答复超过可发送长度，目前只发出了前面的 {sent}/{count} 段；" + "完整答复保存在 LoopX 管家会话中，剩余分段会继续重试。" +) def plan_manager_reply_parts(reply_text: str) -> tuple[list[str], bool]: @@ -192,6 +202,35 @@ def reconciled_part_reply( return {**dict(verified), "part_reconciled": True} +def plan_stalled_part_notice(delivery_state: Mapping[str, Any]) -> str | None: + """The bounded notice a stalled sequence posts once, after real retries. + + A reader who received the first parts of an over-limit answer currently + learns nothing more: the overflow note that says where the full answer lives + only travels with the last part, and the remaining parts are retried in the + background. This notice states what was delivered and where the rest is, and + it waits for more than one failed attempt so an ordinary hiccup stays quiet. + """ + + if delivery_state.get(PART_STALL_NOTICE_KEY) is True: + return None + stalls = delivery_state.get(PART_STALL_COUNT_KEY) + if ( + not isinstance(stalls, int) + or isinstance(stalls, bool) + or stalls < PART_STALL_NOTICE_MIN_STALLS + ): + return None + sent = delivery_state.get("delivery_parts_sent") + count = delivery_state.get("delivery_part_count") + for value in (sent, count): + if not isinstance(value, int) or isinstance(value, bool): + return None + if not 0 < sent < count: + return None + return MANAGER_REPLY_STALL_NOTICE.format(sent=sent, count=count) + + def deliver_manager_reply_parts( *, parts: list[str], @@ -223,6 +262,9 @@ def deliver_manager_reply_parts( ): # A different split than the one on record cannot be resumed safely. sent = 0 + # The stall count describes this sequence's retries, so it goes with the + # split that produced them instead of counting toward a different one. + delivery_state.pop(PART_STALL_COUNT_KEY, None) 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 @@ -334,9 +376,36 @@ def deliver_manager_reply_after_length_failure( message_id=message_id, content_format="text", ) - return reply, ( - None if reply is not None else part_delivery_incomplete_reason(delivery_state) + if reply is not None: + return reply, None + # The answer is only partly on the channel. Tell the reader once, after the + # sequence has already failed more than one attempt, instead of leaving the + # delivered parts looking like the whole answer. + delivery_state[PART_STALL_COUNT_KEY] = ( + int(delivery_state.get(PART_STALL_COUNT_KEY) or 0) + 1 ) + notice = plan_stalled_part_notice(delivery_state) + if notice is not None: + spoken = reply_lark_event_inbox( + project=root, + config_path=config_path, + message_id=message_id, + text=notice, + content_format="text", + execute=True, + runner=reply_runner, + ) + if spoken.get("ok") is True or spoken.get("reply_verified") is True: + delivery_state[PART_STALL_NOTICE_KEY] = True + delivery_state["last_delivery_notice_status"] = str( + spoken.get("status") or "reply_failed" + ) + # The stall count has to survive this attempt either way. A retry reloads the + # record from disk, so an unwritten increment would restart at zero and the + # notice would never be reached no matter how often the sequence stalled. + delivery_state["updated_at"] = datetime.now(timezone.utc).isoformat() + write_delivery(delivery_path, delivery_state) + return None, part_delivery_incomplete_reason(delivery_state) def manager_part_delivery_pending_result( @@ -354,6 +423,7 @@ def manager_part_delivery_pending_result( "reason": reason, "delivery_part_count": delivery_state.get("delivery_part_count"), "delivery_parts_sent": delivery_state.get("delivery_parts_sent"), + "delivery_notice_sent": delivery_state.get(PART_STALL_NOTICE_KEY) is True, "format_degraded": True, "goal_id": goal_id, "inbox_config_ref": inbox_config_ref, From 3105c56452f080246ac1bc249f09554e5c8ad640 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Mon, 21 Sep 2026 20:33:37 +0800 Subject: [PATCH 2/3] test(lark): cover the stalled part-sequence notice Unit coverage for the notice rule (quiet before three stalls, quiet when nothing or everything was delivered, spoken once, re-offered only when the attempt itself failed) plus a route-level test that runs the manager topic through three stalling retries and one clean fourth attempt. Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../test_lark_goal_topic_runtime.py | 114 +++++++++++++++ .../test_lark_manager_reply_parts.py | 132 ++++++++++++++++++ 2 files changed, 246 insertions(+) diff --git a/tests/extensions/test_lark_goal_topic_runtime.py b/tests/extensions/test_lark_goal_topic_runtime.py index aa3309779..588bbaed6 100644 --- a/tests/extensions/test_lark_goal_topic_runtime.py +++ b/tests/extensions/test_lark_goal_topic_runtime.py @@ -2,6 +2,7 @@ import importlib import json +import re import subprocess import threading from collections.abc import Mapping @@ -3111,3 +3112,116 @@ def manager_decision(**kwargs): ).read_text() ) assert "delivery_part_count" not in saved + + +def test_a_stalled_part_sequence_tells_the_reader_what_was_delivered( + tmp_path, monkeypatch, +): + """A partly delivered answer must not look like a complete one. + + Two parts reach the channel and the third cannot be delivered. The reader + currently learns nothing more, so after the sequence has already failed more + than one attempt it is told how many parts went and where the rest is. + """ + + from loopx.extensions.lark import goal_topic_runtime as runtime + from loopx.extensions.lark.manager_reply_parts import PART_STALL_NOTICE_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] = {} + working_runner = _reply_runner(state) + + def part_index(text: str) -> int | None: + match = re.match(r"\((\d+)/(\d+)\) ", text) + return int(match.group(1)) if match else None + + def stalling_runner(args: list[str]) -> dict[str, Any]: + if "+messages-reply" in args and "--dry-run" not in args: + text = args[args.index("--text") + 1] + index = part_index(text) + if index is not None and index >= 3: + return {"returncode": 1} + 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": lambda route, text: { + "response_text": body, + "effect_receipt": runtime._session_turn_effect(route), + }, + "reply_runner": stalling_runner, + } + + def delivered_texts() -> list[str]: + return [ + call[call.index("--text") + 1] + for call in state["calls"] + if "+messages-reply" in call and "--dry-run" not in call + ] + + first = runtime.process_lark_goal_topic_event(**kwargs) + second = runtime.process_lark_goal_topic_event(**kwargs) + + # Two real retries failed: still quiet, and the delivered parts are recorded. + assert first["status"] == "reply_delivery_pending" + assert first["delivery_parts_sent"] == 2 + assert first["delivery_notice_sent"] is False + assert second["delivery_parts_sent"] == 2 + assert second["delivery_notice_sent"] is False + assert not [text for text in delivered_texts() if text.startswith("本条答复")] + + third = runtime.process_lark_goal_topic_event(**kwargs) + + assert third["status"] == "reply_delivery_pending" + assert third["delivery_notice_sent"] is True + notices = [text for text in delivered_texts() if text.startswith("本条答复")] + assert len(notices) == 1 + assert "2/8" in notices[0] + + fourth = runtime.process_lark_goal_topic_event(**kwargs) + + assert fourth["delivery_notice_sent"] is True + assert len([text for text in delivered_texts() if text.startswith("本条答复")]) == 1 + config_path = Path(fourth["inbox_config_ref"]) + from loopx.extensions.lark.manager_reply_delivery import delivery_path + + saved = json.loads( + delivery_path( + project=kwargs["runtime_root"], config_path=config_path, + message_id="om_incoming", + ).read_text() + ) + assert saved[PART_STALL_NOTICE_KEY] is True + assert saved["status"] == "pending" diff --git a/tests/extensions/test_lark_manager_reply_parts.py b/tests/extensions/test_lark_manager_reply_parts.py index eabfc7b30..2c42dbadb 100644 --- a/tests/extensions/test_lark_manager_reply_parts.py +++ b/tests/extensions/test_lark_manager_reply_parts.py @@ -9,15 +9,20 @@ import loopx.extensions.lark.manager_reply_parts as parts_module from loopx.extensions.lark.manager_reply_parts import ( + MANAGER_REPLY_STALL_NOTICE, PART_ATTEMPT_KEY, PART_DELIVERY_COMPLETE_KEY, PART_DELIVERY_COMPLETION_UNVERIFIED, PART_DELIVERY_INCOMPLETE, PART_DELIVERY_VERIFIED_KEY, + PART_STALL_COUNT_KEY, + PART_STALL_NOTICE_KEY, completed_part_delivery_receipt, + deliver_manager_reply_after_length_failure, deliver_manager_reply_parts, part_delivery_incomplete_reason, plan_manager_reply_parts, + plan_stalled_part_notice, recorded_part_attempt, ) @@ -71,6 +76,7 @@ def deliver(parts: list[str]): "sends": sends, "install": install, "deliver": deliver, + "tmp": tmp_path, } @@ -153,6 +159,7 @@ def test_a_changed_split_restarts_instead_of_resuming_mid_answer(monkeypatch, de delivery_part_count=len(parts) + 1, delivery_parts_sent=len(parts) + 1, reply_idempotency_key="sha256:another-split", + **{PART_STALL_COUNT_KEY: 3}, **{ PART_DELIVERY_COMPLETE_KEY: True, PART_DELIVERY_VERIFIED_KEY: True, @@ -165,6 +172,8 @@ def test_a_changed_split_restarts_instead_of_resuming_mid_answer(monkeypatch, de assert delivery["sends"][0] == parts[0] assert delivery["state"]["delivery_part_count"] == len(parts) assert delivery["state"]["delivery_parts_sent"] == len(parts) + # Stalls counted for the abandoned split do not speak for this one. + assert PART_STALL_COUNT_KEY not in delivery["state"] def test_an_unkeyed_or_unfinished_record_never_claims_verification(): @@ -362,3 +371,126 @@ def test_a_confirmed_locator_settles_a_sequence_with_no_new_write( assert delivery["sends"] == [] assert delivery["state"][PART_DELIVERY_COMPLETE_KEY] is True assert delivery["state"][PART_DELIVERY_VERIFIED_KEY] is True + + +def _stalled_state(parts: list[str], *, stalls: int, sent: int) -> dict: + return { + "delivery_part_count": len(parts), + "delivery_parts_sent": sent, + PART_STALL_COUNT_KEY: stalls, + "delivery_text": BODY, + "content_format": "text", + } + + +def test_a_stalled_sequence_tells_the_reader_once_after_real_retries( + monkeypatch, delivery, +): + """Silence is the worst outcome: say what was delivered and where the rest is.""" + + parts, _ = plan_manager_reply_parts(BODY) + state = _stalled_state(parts, stalls=3, sent=2) + sent_texts: list[str] = [] + + def failing_parts(**kwargs): + sent_texts.append(kwargs["text"]) + if kwargs["text"].startswith("("): + return {"ok": False, "status": "reply_provider_failed", + "idempotency_key": None} + return {"ok": True, "status": "sent_verified", "idempotency_key": "sha256:n", + "verification_performed": True, "reply_verified": True} + + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", failing_parts) + + reply, reason = deliver_manager_reply_after_length_failure( + reply_text=BODY, + delivery_state=state, + delivery_path=delivery["tmp"] / "delivery.json", + write_delivery=lambda path, payload: None, + reply_runner=object(), + root=delivery["tmp"], + config_path=delivery["tmp"] / "config.json", + message_id="om_fixture", + ) + + assert reply is None + assert reason == PART_DELIVERY_INCOMPLETE + notices = [text for text in sent_texts if text.startswith("本条答复")] + assert notices == [MANAGER_REPLY_STALL_NOTICE.format(sent=2, count=len(parts))] + assert state[PART_STALL_NOTICE_KEY] is True + + # The reader is told once. A later attempt that still cannot finish the + # sequence does not repeat the notice. + sent_texts.clear() + state[PART_STALL_COUNT_KEY] = 4 + deliver_manager_reply_after_length_failure( + reply_text=BODY, + delivery_state=state, + delivery_path=delivery["tmp"] / "delivery.json", + write_delivery=lambda path, payload: None, + reply_runner=object(), + root=delivery["tmp"], + config_path=delivery["tmp"] / "config.json", + message_id="om_fixture", + ) + assert [text for text in sent_texts if text.startswith("本条答复")] == [] + + +def test_a_sequence_that_has_not_retried_yet_stays_quiet(delivery): + parts, _ = plan_manager_reply_parts(BODY) + + assert plan_stalled_part_notice(_stalled_state(parts, stalls=1, sent=2)) is None + assert plan_stalled_part_notice(_stalled_state(parts, stalls=2, sent=2)) is None + assert plan_stalled_part_notice(_stalled_state(parts, stalls=3, sent=2)) is not None + + +def test_a_sequence_that_delivered_nothing_stays_quiet(delivery): + """Nothing was delivered, so the last-attempt answer is still the whole story.""" + + parts, _ = plan_manager_reply_parts(BODY) + + assert plan_stalled_part_notice(_stalled_state(parts, stalls=9, sent=0)) is None + assert plan_stalled_part_notice(_stalled_state(parts, stalls=9, sent=len(parts))) is None + + +def test_a_rejected_notice_is_offered_again_on_the_next_attempt( + monkeypatch, delivery, +): + parts, _ = plan_manager_reply_parts(BODY) + state = _stalled_state(parts, stalls=3, sent=2) + attempts: list[str] = [] + + def rejecting_everything(**kwargs): + attempts.append(kwargs["text"]) + return {"ok": False, "status": "reply_provider_failed", "idempotency_key": None} + + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", rejecting_everything) + + deliver_manager_reply_after_length_failure( + reply_text=BODY, + delivery_state=state, + delivery_path=delivery["tmp"] / "delivery.json", + write_delivery=lambda path, payload: None, + reply_runner=object(), + root=delivery["tmp"], + config_path=delivery["tmp"] / "config.json", + message_id="om_fixture", + ) + + assert state.get(PART_STALL_NOTICE_KEY) is not True + assert state["last_delivery_notice_status"] == "reply_provider_failed" + assert any(text.startswith("本条答复") for text in attempts) + + state[PART_STALL_COUNT_KEY] = 4 + attempts.clear() + deliver_manager_reply_after_length_failure( + reply_text=BODY, + delivery_state=state, + delivery_path=delivery["tmp"] / "delivery.json", + write_delivery=lambda path, payload: None, + reply_runner=object(), + root=delivery["tmp"], + config_path=delivery["tmp"] / "config.json", + message_id="om_fixture", + ) + assert any(text.startswith("本条答复") for text in attempts) From 7fbfae49b0ce3248bd9bfd5c72e7d7e7ba70ae6a Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Tue, 22 Sep 2026 02:38:10 +0800 Subject: [PATCH 3/3] fix(lark): reconcile a stall notice the provider already took The transport records the provider locator of a sent message before it reads that message back, so a notice the provider accepted can come back `sent_unverified` with the text already on the channel. Counting that as a rejected notice let every later stalled attempt post the reader another copy, which is the one thing the notice is meant not to do. Keep the notice locator in the delivery record and verify it before sending again, exactly as a part is reconciled before it is re-sent, so the reader sees one notice per stalled sequence. The shared "the provider may already hold this text" predicate is the same one the part path uses, and the stall counter's comment now names what it counts. Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- loopx/extensions/lark/manager_reply_parts.py | 183 ++++++++++++++---- .../test_lark_manager_reply_parts.py | 91 +++++++++ 2 files changed, 240 insertions(+), 34 deletions(-) diff --git a/loopx/extensions/lark/manager_reply_parts.py b/loopx/extensions/lark/manager_reply_parts.py index ac6104326..1a4f56328 100644 --- a/loopx/extensions/lark/manager_reply_parts.py +++ b/loopx/extensions/lark/manager_reply_parts.py @@ -15,6 +15,11 @@ 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. + +A partly delivered answer also has to look partly delivered: once the sequence +has stopped mid-answer three times, one bounded notice tells the reader how much +went out and where the rest is. That notice records and verifies its own +locator the same way, so a reader is never told the same thing twice. """ from __future__ import annotations @@ -39,11 +44,13 @@ PART_ATTEMPT_KEY = "delivery_part_attempt" PART_STALL_NOTICE_KEY = "delivery_part_stall_notice" PART_STALL_COUNT_KEY = "delivery_part_stall_count" +PART_STALL_NOTICE_ATTEMPT_KEY = "delivery_part_stall_notice_attempt" PART_DELIVERY_INCOMPLETE = "reply_part_delivery_incomplete" PART_DELIVERY_COMPLETION_UNVERIFIED = "reply_part_delivery_completion_unverified" -# The notice is a last resort, not the outcome: the sequence counts how often it -# stopped mid-answer and only speaks after that happened more than once, so a -# transient provider hiccup never turns into a message the reader did not need. +# The notice is a last resort, not the outcome: the sequence counts the attempts +# that stopped the answer mid-sequence (progress does not reset that count, only +# a different split does) and only speaks after three of them, so a transient +# provider hiccup never turns into a message the reader did not need. PART_STALL_NOTICE_MIN_STALLS = 3 MANAGER_REPLY_STALL_NOTICE = ( "本条答复超过可发送长度，目前只发出了前面的 {sent}/{count} 段；" @@ -66,13 +73,13 @@ 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 reported this part present on the channel. +def _reply_verified(reply: Mapping[str, Any]) -> bool: + """Whether the provider reported this reply present on the channel. - A part can be confirmed either by the readback that follows its own send or + A reply 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. + read back. Both mean the reader has that text, which is the fact the record + keeps; a reconciled send has no new write of its own. """ return bool( @@ -81,17 +88,23 @@ def _part_verified(reply: Mapping[str, Any]) -> bool: ) -def _part_accepted(reply: Mapping[str, Any]) -> bool: - """Whether this part may be counted as delivered. +def _reply_on_channel(reply: Mapping[str, Any]) -> bool: + """Whether the reader may already have this reply's text. - ``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. + ``ok`` also requires the source reaction cleanup to have finished, so a + reply the provider already accepted can come back not-ok with a cleanup + still pending. Treating that as delivered is what keeps a retry from sending + the reader the same text twice, and the pending cleanup stays the + transport's own business. """ - return reply.get("ok") is True or _part_verified(reply) + return reply.get("ok") is True or _reply_verified(reply) + + +def _part_accepted(reply: Mapping[str, Any]) -> bool: + """Whether this part may be counted as delivered.""" + + return _reply_on_channel(reply) def _accepted_reply_facts(reply: Mapping[str, Any]) -> dict[str, Any]: @@ -99,7 +112,7 @@ def _accepted_reply_facts(reply: Mapping[str, Any]) -> dict[str, Any]: return { "reply_idempotency_key": reply.get("idempotency_key"), - PART_DELIVERY_VERIFIED_KEY: _part_verified(reply), + PART_DELIVERY_VERIFIED_KEY: _reply_verified(reply), } @@ -156,6 +169,15 @@ def part_delivery_incomplete_reason(delivery_state: Mapping[str, Any]) -> str: return PART_DELIVERY_INCOMPLETE +def _recorded_attempt(recorded: Any) -> Mapping[str, Any] | None: + """The provider locator inside one recorded attempt, when it is well formed.""" + + if not isinstance(recorded, Mapping): + return None + attempt = recorded.get("attempt") + return attempt if isinstance(attempt, Mapping) else None + + def recorded_part_attempt( delivery_state: Mapping[str, Any], index: int ) -> Mapping[str, Any] | None: @@ -164,8 +186,15 @@ def recorded_part_attempt( 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 + return _recorded_attempt(recorded) + + +def recorded_stall_notice_attempt( + delivery_state: Mapping[str, Any], +) -> Mapping[str, Any] | None: + """The provider locator of a stall notice that was sent but not confirmed.""" + + return _recorded_attempt(delivery_state.get(PART_STALL_NOTICE_ATTEMPT_KEY)) def reconciled_part_reply( @@ -202,6 +231,39 @@ def reconciled_part_reply( return {**dict(verified), "part_reconciled": True} +def reconciled_stall_notice( + *, + notice: str, + delivery_state: Mapping[str, Any], + reply_runner: Any, + root: Path, + config_path: Path, + message_id: str, +) -> Mapping[str, Any] | None: + """Confirm a previously sent stall notice instead of posting it twice. + + The provider can accept the notice and still fail the readback that proves + it, and this sequence keeps retrying until the remaining parts go through. + The recorded locator is what keeps such a notice from being sent again on + every later attempt, exactly as a part is reconciled before it is re-sent. + """ + + attempt = recorded_stall_notice_attempt(delivery_state) + if attempt is None: + return None + verified = verify_lark_inbox_reply( + project=root, + config_path=config_path, + message_id=message_id, + text=notice, + attempt=attempt, + runner=reply_runner, + ) + if verified.get("reply_verified") is not True: + return None + return {**dict(verified), "notice_reconciled": True} + + def plan_stalled_part_notice(delivery_state: Mapping[str, Any]) -> str | None: """The bounded notice a stalled sequence posts once, after real retries. @@ -209,7 +271,7 @@ def plan_stalled_part_notice(delivery_state: Mapping[str, Any]) -> str | None: learns nothing more: the overflow note that says where the full answer lives only travels with the last part, and the remaining parts are retried in the background. This notice states what was delivered and where the rest is, and - it waits for more than one failed attempt so an ordinary hiccup stays quiet. + it waits for three stalled attempts so an ordinary hiccup stays quiet. """ if delivery_state.get(PART_STALL_NOTICE_KEY) is True: @@ -231,6 +293,58 @@ def plan_stalled_part_notice(delivery_state: Mapping[str, Any]) -> str | None: return MANAGER_REPLY_STALL_NOTICE.format(sent=sent, count=count) +def deliver_stall_notice( + *, + delivery_state: dict[str, Any], + delivery_path: Path, + write_delivery, + reply_runner: Any, + root: Path, + config_path: Path, + message_id: str, +) -> Mapping[str, Any] | None: + """Post the once-per-sequence stall notice, confirming a prior send first. + + Returns the transport result of the send or reconciliation, and ``None`` + when this sequence has nothing to tell the reader. + """ + + notice = plan_stalled_part_notice(delivery_state) + if notice is None: + return None + reconciled = reconciled_stall_notice( + notice=notice, + delivery_state=delivery_state, + reply_runner=reply_runner, + root=root, + config_path=config_path, + message_id=message_id, + ) + # The locator of the send being attempted now replaces any older one, so the + # record always points at the most recent unconfirmed notice. + delivery_state.pop(PART_STALL_NOTICE_ATTEMPT_KEY, None) + if reconciled is not None: + return reconciled + + def record_attempt(attempt: Mapping[str, Any]) -> None: + # The locator has to survive the attempt that produced it: a retry + # reloads the record and verifies it instead of posting the notice again. + delivery_state[PART_STALL_NOTICE_ATTEMPT_KEY] = {"attempt": dict(attempt)} + delivery_state["updated_at"] = datetime.now(timezone.utc).isoformat() + write_delivery(delivery_path, delivery_state) + + return reply_lark_event_inbox( + project=root, + config_path=config_path, + message_id=message_id, + text=notice, + content_format="text", + execute=True, + runner=reply_runner, + delivery_attempt_recorder=record_attempt, + ) + + def deliver_manager_reply_parts( *, parts: list[str], @@ -355,8 +469,10 @@ def deliver_manager_reply_after_length_failure( """Deliver one over-limit manager answer as bounded parts. Returns the last accepted reply, or ``None`` plus the reason to report when - a part was rejected. Plain text is the only format a split can promise, so - the caller has already degraded presentation before calling this. + a part was rejected. A sequence that stops with parts still unsent also tells + the reader what went out, once, after enough failed attempts. Plain text is + the only format a split can promise, so the caller has already degraded + presentation before calling this. """ parts, truncated = plan_manager_reply_parts(reply_text) @@ -384,18 +500,17 @@ def deliver_manager_reply_after_length_failure( delivery_state[PART_STALL_COUNT_KEY] = ( int(delivery_state.get(PART_STALL_COUNT_KEY) or 0) + 1 ) - notice = plan_stalled_part_notice(delivery_state) - if notice is not None: - spoken = reply_lark_event_inbox( - project=root, - config_path=config_path, - message_id=message_id, - text=notice, - content_format="text", - execute=True, - runner=reply_runner, - ) - if spoken.get("ok") is True or spoken.get("reply_verified") is True: + spoken = deliver_stall_notice( + delivery_state=delivery_state, + delivery_path=delivery_path, + write_delivery=write_delivery, + reply_runner=reply_runner, + root=root, + config_path=config_path, + message_id=message_id, + ) + if spoken is not None: + if _reply_on_channel(spoken): delivery_state[PART_STALL_NOTICE_KEY] = True delivery_state["last_delivery_notice_status"] = str( spoken.get("status") or "reply_failed" diff --git a/tests/extensions/test_lark_manager_reply_parts.py b/tests/extensions/test_lark_manager_reply_parts.py index 2c42dbadb..4d66484c1 100644 --- a/tests/extensions/test_lark_manager_reply_parts.py +++ b/tests/extensions/test_lark_manager_reply_parts.py @@ -15,6 +15,7 @@ PART_DELIVERY_COMPLETION_UNVERIFIED, PART_DELIVERY_INCOMPLETE, PART_DELIVERY_VERIFIED_KEY, + PART_STALL_NOTICE_ATTEMPT_KEY, PART_STALL_COUNT_KEY, PART_STALL_NOTICE_KEY, completed_part_delivery_receipt, @@ -24,6 +25,7 @@ plan_manager_reply_parts, plan_stalled_part_notice, recorded_part_attempt, + recorded_stall_notice_attempt, ) @@ -494,3 +496,92 @@ def rejecting_everything(**kwargs): message_id="om_fixture", ) assert any(text.startswith("本条答复") for text in attempts) + + +def test_a_notice_the_provider_took_but_did_not_read_back_is_confirmed( + monkeypatch, delivery, +): + """A notice the provider accepted but could not read back is not repeated. + + The transport records the provider locator of the notice before it reads the + message back, so a send that comes back ``sent_unverified`` has to be + confirmed on the next attempt instead of being posted to the reader twice. + """ + + parts, _ = plan_manager_reply_parts(BODY) + state = _stalled_state(parts, stalls=2, sent=2) + notice = MANAGER_REPLY_STALL_NOTICE.format(sent=2, count=len(parts)) + sends: list[str] = [] + verifications: list[str] = [] + + def ambiguous_send(**kwargs): + sends.append(kwargs["text"]) + if kwargs["text"].startswith("("): + return { + "ok": False, + "status": "reply_provider_failed", + "idempotency_key": None, + } + kwargs["delivery_attempt_recorder"]( + { + "schema_version": "manager_return_delivery_attempt_v0", + "provider": "lark", + "message_ref": "om_notice", + "intent_digest": "sha256:notice-intent", + "provider_receipt": "sha256:notice-receipt", + } + ) + return { + "ok": False, + "status": "sent_unverified", + "idempotency_key": "sha256:notice-receipt", + "write_performed": True, + "verification_performed": True, + "reply_verified": False, + } + + def confirmed_notice(**kwargs): + verifications.append(kwargs["text"]) + return { + "ok": True, + "status": "sent_verified", + "idempotency_key": "sha256:notice-receipt", + "verification_performed": True, + "reply_verified": True, + } + + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", ambiguous_send) + monkeypatch.setattr(parts_module, "verify_lark_inbox_reply", confirmed_notice) + + def deliver_with_stalls(): + return deliver_manager_reply_after_length_failure( + reply_text=BODY, + delivery_state=state, + delivery_path=delivery["tmp"] / "delivery.json", + write_delivery=lambda path, payload: None, + reply_runner=object(), + root=delivery["tmp"], + config_path=delivery["tmp"] / "config.json", + message_id="om_fixture", + ) + + deliver_with_stalls() + + # The notice went out but its readback did not confirm it, so the reader may + # already have it and the record keeps that attempt's locator. + assert [text for text in sends if text.startswith("本条答复")] == [notice] + assert state.get(PART_STALL_NOTICE_KEY) is not True + assert recorded_stall_notice_attempt(state) == { + "schema_version": "manager_return_delivery_attempt_v0", + "provider": "lark", + "message_ref": "om_notice", + "intent_digest": "sha256:notice-intent", + "provider_receipt": "sha256:notice-receipt", + } + + deliver_with_stalls() + + assert verifications == [notice] + assert [text for text in sends if text.startswith("本条答复")] == [notice] + assert state[PART_STALL_NOTICE_KEY] is True + assert PART_STALL_NOTICE_ATTEMPT_KEY not in state