fix(lark): stop repeating a part the provider sent without a locator

loopx/capabilities/manager_context/roundtrip.py

@@ -78,6 +78,24 @@ def _delivery_attempt(value):
     )
 
 
+def _attempt_locator(value):
+    """The provider locator a recorded attempt can be verified against.
+
+    ``None`` covers both a record no normalization can read and the typed state
+    where the provider accepted the write without reporting a message id. The
+    attempt still proves a write happened; it just cannot name a readback
+    target, and that is what stops the pump from treating the return as unsent.
+    """
+
+    if value is None:
+        return None
+    try:
+        message_ref = _delivery_attempt(value).get("message_ref")
+    except (ValueError, EffectRuntimeRejected):
+        return None
+    return message_ref if isinstance(message_ref, str) and message_ref.strip() else None
+
+
 def _verification_decision(outcome):
     return dict(
         effect_runtime_result(
@@ -379,6 +397,22 @@ def drain(root, registry, store, external_sender, *, now=None, cancelled=lambda:
                                 },
                             )
                             continue
+                        if _attempt_locator(state.get("attempt")) is None:
+                            # The attempt records a provider write that carried no
+                            # locator, so no readback can prove it and the provider
+                            # must not be called again. Converging here keeps the
+                            # attempt as the evidence of that write while making
+                            # the return terminal, instead of looping the pump
+                            # through verification attempts forever.
+                            _write(
+                                state_path,
+                                {
+                                    **state,
+                                    "status": "explicit_unverified",
+                                    "error": "provider_locator_unavailable",
+                                },
+                            )
+                            continue
                         verifier = getattr(external_sender, "verify", None)
                         if not callable(verifier):
                             _write(
@@ -489,21 +523,39 @@ def record_attempt(value):
                     if sent.get("reply_verified") is not True:
                         if sent.get("external_write_performed") is True:
                             current = _read(state_path) if state_path.exists() else {}
-                            _write(
-                                state_path,
-                                (
+                            if (
+                                current.get("attempt") is not None
+                                and _attempt_locator(current.get("attempt")) is not None
+                            ):
+                                _write(
+                                    state_path,
                                     {
                                         **current,
                                         "status": "verification_required",
                                         "error": "provider_delivery_unverified",
-                                    }
-                                    if current.get("attempt") is not None
-                                    else {
-                                        "status": "explicit_unverified",
-                                        "error": "provider_locator_unavailable",
-                                    }
-                                ),
-                            )
+                                    },
+                                )
+                            else:
+                                # Either nothing was recorded or the record says
+                                # the provider took the write without a locator.
+                                # Both leave no readback target, so the return is
+                                # terminal rather than retryable: a retry would
+                                # post the same text again.
+                                _write(
+                                    state_path,
+                                    (
+                                        {
+                                            **current,
+                                            "status": "explicit_unverified",
+                                            "error": "provider_locator_unavailable",
+                                        }
+                                        if current.get("attempt") is not None
+                                        else {
+                                            "status": "explicit_unverified",
+                                            "error": "provider_locator_unavailable",
+                                        }
+                                    ),
+                                )
                             continue
                         raise ValueError("return_transport_unavailable")
                     transport = {
@@ -530,6 +582,19 @@ def record_attempt(value):
                             state_path,
                             {"status": "explicit_unverified", "error": error},
                         )
+                    elif _attempt_locator(current.get("attempt")) is None:
+                        # The record says the provider took the write and named
+                        # no locator, so there is nothing left to verify and a
+                        # retry would post the same text again. Converge now
+                        # instead of leaving the return in a retryable state.
+                        _write(
+                            state_path,
+                            {
+                                **current,
+                                "status": "explicit_unverified",
+                                "error": "provider_locator_unavailable",
+                            },
+                        )
                     processed += 1
                     continue
                 attempts = int(state.get("attempts", 0)) + 1

loopx/control_plane/collaboration/return_delivery.ts

@@ -28,6 +28,20 @@ function matchingString(value: unknown, label: string, pattern: RegExp): string
   return result;
 }
 
+/**
+ * The provider locator for one recorded attempt, when the provider gave one.
+ *
+ * `null` is the typed state for "the provider accepted the write and reported
+ * no message id". Such an attempt is still the durable record that a write
+ * happened; what it cannot do is name a readback target. Keeping the key
+ * required and the absence explicit is what stops a later retry from treating
+ * an unlocatable write as a write that never happened.
+ */
+function optionalOpaqueRef(value: unknown, label: string): string | null {
+  if (value === null) return null;
+  return matchingString(value, label, OPAQUE_REF);
+}
+
 export function normalizeManagerReturnDeliveryAttempt(value: unknown): JsonObject {
   const attempt = requireJsonObject(value, "attempt");
   const keys = Object.keys(attempt).sort();
@@ -45,11 +59,7 @@ export function normalizeManagerReturnDeliveryAttempt(value: unknown): JsonObjec
   return {
     schema_version: MANAGER_RETURN_DELIVERY_ATTEMPT_SCHEMA,
     provider: matchingString(attempt.provider, "attempt.provider", PROVIDER),
-    message_ref: matchingString(
-      attempt.message_ref,
-      "attempt.message_ref",
-      OPAQUE_REF,
-    ),
+    message_ref: optionalOpaqueRef(attempt.message_ref, "attempt.message_ref"),
     intent_digest: matchingString(
       attempt.intent_digest,
       "attempt.intent_digest",

loopx/extensions/lark/inbox_reply.py

@@ -571,8 +571,40 @@ def _deliver_lark_inbox_outbound(
             provider_preview_verified=True,
         )
 
+    intent_digest = _intent_digest(profile, chat_id, receipt)
     reply_message_id = _message_id(_json_object(send.get("stdout")))
     if not reply_message_id:
+        # The provider accepted the write and reported no message id, so there is
+        # nothing a later readback could key on. Recording the attempt with an
+        # empty locator is what keeps a retry from posting the same text again:
+        # the durable record proves a write happened even though it cannot be
+        # located, and a locator nothing can verify must not be re-sent blindly.
+        if delivery_attempt_recorder is not None:
+            try:
+                delivery_attempt_recorder(
+                    {
+                        "schema_version": "manager_return_delivery_attempt_v0",
+                        "provider": "lark",
+                        "message_ref": None,
+                        "intent_digest": intent_digest,
+                        "provider_receipt": receipt,
+                    }
+                )
+            except (OSError, TypeError, ValueError):
+                return _result(
+                    status="sent_unverified",
+                    ok=False,
+                    execute=True,
+                    receipt=receipt,
+                    identity_verified=True,
+                    membership_verified=True,
+                    write_performed=True,
+                    placement=placement,
+                    blocker="lark_inbox_reply_delivery_attempt_not_persisted",
+                    format_preflight_passed=True,
+                    provider_preview_performed=True,
+                    provider_preview_verified=True,
+                )
         return _result(
             status="sent_unverified",
             ok=False,
@@ -587,7 +619,6 @@ def _deliver_lark_inbox_outbound(
             provider_preview_performed=True,
             provider_preview_verified=True,
         )
-    intent_digest = _intent_digest(profile, chat_id, receipt)
     if delivery_attempt_recorder is not None:
         try:
             delivery_attempt_recorder(

loopx/extensions/lark/manager_reply_parts.py

@@ -28,7 +28,11 @@
 from pathlib import Path
 from typing import Any, Mapping
 
-from .inbox_reply import reply_lark_event_inbox, verify_lark_inbox_reply
+from .inbox_reply import (
+    MESSAGE_ID_PATTERN,
+    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:
@@ -189,6 +193,35 @@ def recorded_part_attempt(
     return _recorded_attempt(recorded)
 
 
+def attempt_provider_locator(attempt: Mapping[str, Any]) -> str | None:
+    """The message id a recorded attempt can be verified against, when it has one.
+
+    A send the provider accepted without reporting a message id records its
+    intent instead of a locator. That record still proves a write happened, and
+    ``None`` here is what tells the sequence to stop instead of posting the same
+    text a second time.
+    """
+
+    message_ref = str(attempt.get("message_ref") or "").strip()
+    return message_ref if MESSAGE_ID_PATTERN.fullmatch(message_ref) else None
+
+
+def _locator_unavailable_result(*, reconciled_key: str) -> dict[str, Any]:
+    """The typed outcome for an attempt no readback can key on."""
+
+    return {
+        "ok": False,
+        "status": "sent_unverified",
+        "idempotency_key": None,
+        "external_write_performed": True,
+        "verification_performed": False,
+        "reply_verified": False,
+        "blocker": "lark_inbox_reply_not_verified",
+        reconciled_key: False,
+        "part_locator_unavailable": True,
+    }
+
+
 def recorded_stall_notice_attempt(
     delivery_state: Mapping[str, Any],
 ) -> Mapping[str, Any] | None:
@@ -218,6 +251,12 @@ def reconciled_part_reply(
     attempt = recorded_part_attempt(delivery_state, index)
     if attempt is None:
         return None
+    if attempt_provider_locator(attempt) is None:
+        # The provider took the write and gave no message id, so this part cannot
+        # be read back. Sending it again risks delivering the same text twice,
+        # which is worse than reporting the sequence as unverified: the record
+        # keeps the part where it stopped until the caller decides.
+        return _locator_unavailable_result(reconciled_key="part_reconciled")
     verified = verify_lark_inbox_reply(
         project=root,
         config_path=config_path,
@@ -251,6 +290,8 @@ def reconciled_stall_notice(
     attempt = recorded_stall_notice_attempt(delivery_state)
     if attempt is None:
         return None
+    if attempt_provider_locator(attempt) is None:
+        return _locator_unavailable_result(reconciled_key="notice_reconciled")
     verified = verify_lark_inbox_reply(
         project=root,
         config_path=config_path,
@@ -320,11 +361,22 @@ def deliver_stall_notice(
         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:
+        if reconciled.get("notice_reconciled") is not True:
+            # The provider accepted this notice and nothing can read it back, so
+            # this record is the only evidence the reader may already have it.
+            # Dropping it here would let the next retry post the same notice
+            # again; a confirmed notice, below, is the case that settles it.
+            return reconciled
+        # A confirmed notice is settled: the stall flag carries that fact from
+        # here on, so the attempt that proved it is no longer needed.
+        delivery_state.pop(PART_STALL_NOTICE_ATTEMPT_KEY, None)
         return reconciled
+    # The locator of the send being attempted now replaces any older one, so the
+    # record always points at the most recent unconfirmed notice. This runs only
+    # on the path that is about to call the provider, where the old record is
+    # genuinely superseded.
+    delivery_state.pop(PART_STALL_NOTICE_ATTEMPT_KEY, None)
 
     def record_attempt(attempt: Mapping[str, Any]) -> None:
         # The locator has to survive the attempt that produced it: a retry

tests/control_plane_ts/manager_return_delivery.test.ts

@@ -16,6 +16,13 @@ const attempt = {
 
 test("normalizes the exact provider-neutral delivery attempt", () => {
   assert.deepEqual(normalizeManagerReturnDeliveryAttempt(attempt), attempt);
+  // The provider accepted the write and reported no message id. The attempt is
+  // still the record of that write, so the locator is typed as absent instead
+  // of being rejected or faked.
+  assert.deepEqual(
+    normalizeManagerReturnDeliveryAttempt({ ...attempt, message_ref: null }),
+    { ...attempt, message_ref: null },
+  );
   assert.throws(
     () => normalizeManagerReturnDeliveryAttempt({ ...attempt, private_payload: "no" }),
     /unsupported or missing fields/,
@@ -24,6 +31,10 @@ test("normalizes the exact provider-neutral delivery attempt", () => {
     () => normalizeManagerReturnDeliveryAttempt({ ...attempt, message_ref: "bad ref" }),
     /message_ref is invalid/,
   );
+  assert.throws(
+    () => normalizeManagerReturnDeliveryAttempt({ ...attempt, message_ref: "" }),
+    /message_ref must be a non-empty string/,
+  );
 });
 
 test("classifies verification without exposing provider prose", () => {