fix(lark): reconcile an ambiguous part send instead of repeating it

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,
             **(

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