diff --git a/loopx/extensions/lark/manager_reply_parts.py b/loopx/extensions/lark/manager_reply_parts.py index 2d25a3a2f..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 @@ -37,8 +42,20 @@ 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_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 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} 段；" + "完整答复保存在 LoopX 管家会话中，剩余分段会继续重试。" +) def plan_manager_reply_parts(reply_text: str) -> tuple[list[str], bool]: @@ -56,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( @@ -71,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]: @@ -89,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), } @@ -146,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: @@ -154,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( @@ -192,6 +231,120 @@ 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. + + 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 three stalled attempts 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_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], @@ -223,6 +376,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 @@ -313,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) @@ -334,9 +492,35 @@ 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 ) + 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" + ) + # 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 +538,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, 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..4d66484c1 100644 --- a/tests/extensions/test_lark_manager_reply_parts.py +++ b/tests/extensions/test_lark_manager_reply_parts.py @@ -9,16 +9,23 @@ 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_NOTICE_ATTEMPT_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, + recorded_stall_notice_attempt, ) @@ -71,6 +78,7 @@ def deliver(parts: list[str]): "sends": sends, "install": install, "deliver": deliver, + "tmp": tmp_path, } @@ -153,6 +161,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 +174,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 +373,215 @@ 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) + + +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