diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index e8f31a444..fdbb1be71 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/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 diff --git a/loopx/control_plane/collaboration/return_delivery.ts b/loopx/control_plane/collaboration/return_delivery.ts index 68855f81d..1efb99a6a 100644 --- a/loopx/control_plane/collaboration/return_delivery.ts +++ b/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", diff --git a/loopx/extensions/lark/inbox_reply.py b/loopx/extensions/lark/inbox_reply.py index 2d635c4b6..ab9a8d7de 100644 --- a/loopx/extensions/lark/inbox_reply.py +++ b/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( diff --git a/loopx/extensions/lark/manager_reply_parts.py b/loopx/extensions/lark/manager_reply_parts.py index 1a4f56328..6dcebc62d 100644 --- a/loopx/extensions/lark/manager_reply_parts.py +++ b/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 diff --git a/tests/control_plane_ts/manager_return_delivery.test.ts b/tests/control_plane_ts/manager_return_delivery.test.ts index 1f6e7d623..4b2cefd2b 100644 --- a/tests/control_plane_ts/manager_return_delivery.test.ts +++ b/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", () => { diff --git a/tests/extensions/test_lark_inbox_reactions.py b/tests/extensions/test_lark_inbox_reactions.py index 62e7b63fe..e15dbb644 100644 --- a/tests/extensions/test_lark_inbox_reactions.py +++ b/tests/extensions/test_lark_inbox_reactions.py @@ -329,6 +329,7 @@ def __init__( member_bucket: str = "users", member_read_denied: bool = False, auth_failures: int = 0, + send_message_id: str | None = "om_reply_fixture", ) -> None: self.calls: list[list[str]] = [] self.matching_readback = matching_readback @@ -339,6 +340,7 @@ def __init__( self.member_bucket = member_bucket self.member_read_denied = member_read_denied self.auth_failures = auth_failures + self.send_message_id = send_message_id def __call__(self, args: Sequence[str]) -> dict[str, Any]: call = list(args) @@ -428,7 +430,11 @@ def __call__(self, args: Sequence[str]) -> dict[str, Any]: } return { "returncode": 0, - "stdout": json.dumps({"message_id": "om_reply_fixture"}), + "stdout": json.dumps( + {"message_id": self.send_message_id} + if self.send_message_id is not None + else {} + ), "stderr": "", } if "+messages-mget" in call: @@ -976,6 +982,51 @@ def test_unverified_reply_records_private_locator_and_read_only_recovery( ) +def test_a_send_that_reports_no_message_id_records_its_intent( + tmp_path: Path, +) -> None: + """A provider write without a message id must still leave a record. + + Nothing can read back a send that reported no message id, so without the + record a retry has no evidence a write happened and posts the same text + again. The recorded intent is what stops that, and no readback is attempted + because there is no message id to key it on. + """ + + config, _, project = _fixture(tmp_path, lifecycle=False) + attempts: list[dict[str, str]] = [] + runner = ReplyRunner(send_message_id=None) + + sent = reply_lark_event_inbox( + project=project, + config_path=config, + message_id="om_reaction_fixture", + text="处理完成", + execute=True, + runner=runner, + delivery_attempt_recorder=attempts.append, + ) + + assert sent["status"] == "sent_unverified" + assert sent["external_write_performed"] is True + assert sent["reply_verified"] is False + assert sent["blocker"] == "lark_inbox_reply_not_verified" + assert attempts == [ + { + "schema_version": "manager_return_delivery_attempt_v0", + "provider": "lark", + # The provider supplied no message id, so the locator is absent + # rather than an invalid empty string the canonical contract would + # reject: the attempt records the write, not a readback target. + "message_ref": None, + "intent_digest": attempts[0]["intent_digest"], + "provider_receipt": sent["idempotency_key"], + } + ] + assert attempts[0]["intent_digest"].startswith("sha256:") + assert not any("+messages-mget" in call for call in runner.calls) + + def test_read_only_recovery_rejects_changed_intent_without_provider_call( tmp_path: Path, ) -> None: diff --git a/tests/extensions/test_lark_manager_reply_parts.py b/tests/extensions/test_lark_manager_reply_parts.py index 4d66484c1..854b29154 100644 --- a/tests/extensions/test_lark_manager_reply_parts.py +++ b/tests/extensions/test_lark_manager_reply_parts.py @@ -249,6 +249,37 @@ def cleanup_pending_then_ok(**kwargs): "provider_receipt": "sha256:" + "b" * 64, } +# The provider accepted the write and reported no message id, so the attempt +# carries its intent instead of a locator nothing could read back. +LOCATOR_LESS_ATTEMPT = {**ATTEMPT, "message_ref": None} + + +def test_a_recorded_send_without_a_locator_is_never_sent_again( + monkeypatch, delivery, +): + """A write that reported no message id must not be repeated blindly.""" + + parts, _ = plan_manager_reply_parts(BODY) + delivery["state"].update( + delivery_part_count=len(parts), + delivery_parts_sent=0, + **{PART_ATTEMPT_KEY: {"index": 0, "attempt": LOCATOR_LESS_ATTEMPT}}, + ) + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", delivery["install"]()) + + assert delivery["deliver"](parts) is None + + # The part stays where it stopped: no second write, and the record is kept + # so the reader is told the answer is incomplete rather than duplicated. + assert delivery["sends"] == [] + assert delivery["state"]["delivery_parts_sent"] == 0 + assert delivery["state"][PART_ATTEMPT_KEY] == { + "index": 0, + "attempt": LOCATOR_LESS_ATTEMPT, + } + assert delivery["state"]["last_delivery_status"] == "sent_unverified" + assert delivery["state"].get(PART_DELIVERY_COMPLETE_KEY) is not True + def test_a_send_without_a_readback_records_its_provider_locator( monkeypatch, delivery, @@ -585,3 +616,84 @@ def deliver_with_stalls(): 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 + + +def test_a_stall_notice_without_a_locator_survives_its_own_writeback( + monkeypatch, delivery, +): + """A notice the provider accepted without a locator is never sent twice. + + The notice attempt is the only durable evidence that the reader may already + have the notice. Dropping it while the send is still being reconciled makes + the next retry post the same text again, so the record has to survive the + writeback that follows a reconciliation result. + """ + + 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] = [] + written: list[dict] = [] + + def locator_less_send(**kwargs): + sends.append(kwargs["text"]) + if kwargs["text"].startswith("("): + return { + "ok": False, + "status": "reply_provider_failed", + "idempotency_key": None, + } + kwargs["delivery_attempt_recorder"]( + dict(LOCATOR_LESS_ATTEMPT) + ) + return { + "ok": False, + "status": "sent_unverified", + "idempotency_key": "sha256:notice-receipt", + "external_write_performed": True, + "verification_performed": False, + "reply_verified": False, + } + + def readback(**kwargs): + verifications.append(kwargs["text"]) + raise AssertionError("a notice without a locator has nothing to verify") + + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", locator_less_send) + monkeypatch.setattr(parts_module, "verify_lark_inbox_reply", readback) + + def deliver_with_stalls(current: dict): + return deliver_manager_reply_after_length_failure( + reply_text=BODY, + delivery_state=current, + delivery_path=delivery["tmp"] / "delivery.json", + write_delivery=lambda path, payload: written.append( + json.loads(json.dumps(payload)) + ), + reply_runner=object(), + root=delivery["tmp"], + config_path=delivery["tmp"] / "config.json", + message_id="om_fixture", + ) + + deliver_with_stalls(state) + first = written[-1] + + # The notice left the provider once, and the attempt that proves it is on + # disk even though nothing can read the message back. + assert [text for text in sends if text.startswith("本条答复")] == [notice] + assert first[PART_STALL_NOTICE_ATTEMPT_KEY] == {"attempt": LOCATOR_LESS_ATTEMPT} + assert first.get(PART_STALL_NOTICE_KEY) is not True + assert first["last_delivery_notice_status"] == "sent_unverified" + + # A retry reloads the record that was just written: it reconciles instead of + # posting the notice again, and the evidence of the first send stays. + deliver_with_stalls(json.loads(json.dumps(first))) + + assert [text for text in sends if text.startswith("本条答复")] == [notice] + assert verifications == [] + assert written[-1][PART_STALL_NOTICE_ATTEMPT_KEY] == { + "attempt": LOCATOR_LESS_ATTEMPT + } + assert written[-1]["last_delivery_notice_status"] == "sent_unverified" diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index 9e2d257ca..4f1f5de1a 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -795,3 +795,73 @@ def forbidden(*_): assert not [m for m in store.messages(session["session_id"]) if m.get("origin") == "manager_followup"] assert reply_status(root, receipt)[0]["status"] == "retry_pending" + + +def test_provider_write_without_a_locator_is_terminal_and_not_repeated(flow): + """A write the provider took without a message id is never sent again. + + The canonical attempt contract holds "accepted, no locator" as a typed + state. The pump has to converge there: a return whose only record names no + readback target cannot stay retryable, because the retry would post text the + reader may already have. + """ + + root, registry, store, create = flow + session, _, receipt = create(True) + rid = receipt["request_id"] + acknowledge(root, "research", "worker", rid, "adopt", "Checked") + report( + root, + "research", + "worker", + rid, + "conclusion", + "Processed with a recorded validation result.", + ) + + class Transport: + def __init__(self): + self.send_calls = 0 + self.verify_calls = 0 + + def send_with_attempt(self, route, session, turn, text, record_attempt): + self.send_calls += 1 + record_attempt( + { + "schema_version": "manager_return_delivery_attempt_v0", + "provider": "lark", + "message_ref": None, + "intent_digest": "sha256:" + "a" * 64, + "provider_receipt": "sha256:" + "b" * 64, + } + ) + raise RuntimeError("synthetic process interruption after provider send") + + def verify(self, route, session, turn, text, attempt): + self.verify_calls += 1 + raise AssertionError("a write with no locator has nothing to verify") + + transport = Transport() + drain(root, registry, store, transport) + first = reply_status(root, receipt)[0] + + # The write is recorded and the return settles as unverifiable in the same + # pass: nothing can read it back, so it must not stay retryable. + assert first["status"] == "explicit_unverified" + assert first["error"] == "provider_locator_unavailable" + assert transport.send_calls == 1 + + state_path = next( + (root / ".local" / "manager-context" / "replies").glob( + f"{receipt['request_id']}/conclusion.delivery.json" + ) + ) + state = json.loads(state_path.read_text()) + assert state["attempt"]["message_ref"] is None + assert state["attempt"]["intent_digest"] == "sha256:" + "a" * 64 + + drain(root, registry, ChatSessionStore(root), transport) + again = reply_status(root, receipt)[0] + assert again["status"] == "explicit_unverified" + assert transport.send_calls == 1 + assert transport.verify_calls == 0