fix(lark): settle a part sequence from what the provider already accepted

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(

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"] == []