diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index ec5d6e2478..86cd526afe 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -562,6 +562,23 @@ unsupported transport remains an explicit delivery gap. A path mentioned only in text is not an attachment. This slice does not qualify local-Web file presentation, group disclosure, cross-host transfer or inline video playback. Installed live provider qualification remains separate from the synthetic transport regressions. + +Oversized Markdown results stay rich posts. The Lark presentation splitter counts +UTF-8 bytes of the serialized provider request, including escaped JSON, numbering +and attachments, against the documented 30 KB post bound (30,000 bytes as a +conservative local interpretation). It keeps the complete +body across ordered parts without a fixed part-count cap or a plain-text fallback; +code fences and table column headers continue across oversized blocks. The +existing private Lark part-delivery journal retains the frozen plan and each +provider locator. Recovery verifies known writes before continuing never-sent +parts; unknown writes, changed intent or lost locators cannot be blindly resent. +Attachments remain on the last concrete post. The adapter marks a composite +delivery complete only after every part and required attachment readback succeeds. +An atomic link or table row that cannot fit one post remains an explicit delivery +failure; it is never silently truncated or sent as plain text. +Historical single-text receipts remain verify-only compatibility. Intentional +text and identity-verified mention delivery keep their existing contracts. + Before downgrading to an older runtime that does not understand result files, pause the reply pump while any file-bearing result remains pending; an old text-only reader cannot certify attachment delivery. Retain the result snapshots diff --git a/loopx/extensions/lark/goal_topic_runtime.py b/loopx/extensions/lark/goal_topic_runtime.py index 4b5fb1fd1c..ea4106426f 100644 --- a/loopx/extensions/lark/goal_topic_runtime.py +++ b/loopx/extensions/lark/goal_topic_runtime.py @@ -66,13 +66,11 @@ sync_manager_context, ) from .manager_reply_parts import ( - deliver_manager_reply_after_length_failure, - manager_part_delivery_pending_result, manager_part_delivery_readback, ) from .manager_reply_format import repair_manager_reply_text from .private_json import write_private_json_atomic -from .outbound import LarkOutboundTextError, safe_lark_plain_text_fallback +from .outbound import LarkOutboundTextError, present_lark_conversation_text from .inbox_reactions import ( _create_reaction, _delete_reaction, @@ -1401,6 +1399,7 @@ def _process_lark_goal_topic_event( # owning unsafe markup, and a repaired markdown answer stays markdown # instead of degrading to plain text. reply_text, rich_text_repairs = repair_manager_reply_text(reply_text) + reply_text = present_lark_conversation_text(reply_text) if manager and delivery_state is None: assert delivery_path is not None delivery_state = _pending_manager_delivery( @@ -1468,83 +1467,27 @@ def _process_lark_goal_topic_event( content_format=content_format, execute=True, runner=reply_runner, - # A manager answer is bounded by the provider's request limit, - # not by the compact notification length: one answer is one - # message, and the bounded part sequence is the fallback for a - # body the provider itself cannot take. + # Ordinary answers use the provider capacity. The shared Inbox + # sender splits oversized rich replies without a text downgrade. short_message_limit=None, ) except LarkOutboundTextError: if not manager: raise - # The rich body did not fit. Degrade presentation first, then split - # the degraded body: a persisted manager answer must be delivered in - # bounded parts instead of becoming `format_unrepresentable` with - # nothing on the channel. - try: - reply_text = safe_lark_plain_text_fallback(reply_text) - content_format = "text" - assert delivery_state is not None and delivery_path is not None - delivery_state.update( - delivery_text=reply_text, - delivery_digest=_manager_delivery_text_digest(reply_text), - content_format=content_format, - format_degraded=True, - attempt_count=int(delivery_state.get("attempt_count") or 0) + 1, - updated_at=datetime.now(timezone.utc).isoformat(), - ) - _write_manager_delivery(delivery_path, delivery_state) - try: - reply = reply_lark_event_inbox( - project=root, - config_path=config_path, - message_id=message_id, - text=reply_text, - content_format=content_format, - execute=True, - runner=reply_runner, - ) - except LarkOutboundTextError: - if not reply_text.strip(): - raise - part_reply, part_failure = ( - deliver_manager_reply_after_length_failure( - reply_text=reply_text, - delivery_state=delivery_state, - delivery_path=delivery_path, - write_delivery=_write_manager_delivery, - reply_runner=reply_runner, - root=root, - config_path=config_path, - message_id=message_id, - ) - ) - if part_failure: - return manager_part_delivery_pending_result( - reason=part_failure, - delivery_state=delivery_state, - goal_id=route["goal_id"], - inbox_config_ref=config_ref, - ) - reply = part_reply - except (LarkOutboundTextError, ValueError): - assert delivery_state is not None and delivery_path is not None - delivery_state.update( - last_delivery_status="format_unrepresentable", - updated_at=datetime.now(timezone.utc).isoformat(), - ) - _write_manager_delivery(delivery_path, delivery_state) - return { - "ok": False, - "status": "reply_delivery_pending", - "reason": "reply_format_invalid", - "format_degraded": bool(delivery_state.get("format_degraded")), - "goal_id": route["goal_id"], - "inbox_config_ref": config_ref, - "source_acknowledged": False, - } + # Length is handled by the shared rich-post splitter. Invalid + # presentation remains an explicit failure, never a text downgrade. + assert delivery_state is not None and delivery_path is not None + delivery_state.update(last_delivery_status="format_unrepresentable", + updated_at=datetime.now(timezone.utc).isoformat()) + _write_manager_delivery(delivery_path, delivery_state) + return {"ok": False, "status": "reply_delivery_pending", + "reason": "reply_format_invalid", "format_degraded": False, + "goal_id": route["goal_id"], "inbox_config_ref": config_ref, + "source_acknowledged": False} if manager and reply.get("content_format") in {"markdown", "text"}: content_format = str(reply["content_format"]) + if manager and delivery_state is not None: + delivery_state.update(manager_part_delivery_readback(reply)) if not reply.get("ok"): if manager: assert delivery_state is not None and delivery_path is not None @@ -1570,6 +1513,8 @@ def _process_lark_goal_topic_event( "rich_text_repairs": list( (delivery_state or {}).get("rich_text_repairs") or [] ), + **manager_part_delivery_readback(reply), + "delivery_notice_sent": reply.get("delivery_notice_sent") is True, } if manager else {} diff --git a/loopx/extensions/lark/inbox_reply.py b/loopx/extensions/lark/inbox_reply.py index 713b3521d6..8d1c4d6154 100644 --- a/loopx/extensions/lark/inbox_reply.py +++ b/loopx/extensions/lark/inbox_reply.py @@ -23,6 +23,7 @@ ) from .outbound import ( + LarkOutboundTextError, DEFAULT_LARK_TEXT_LIMIT, LARK_POST_REQUEST_MAX_BYTES, expected_lark_mention_identities, @@ -371,7 +372,8 @@ def _deliver_lark_inbox_outbound( ) # Reject an oversized content lower bound before building a CLI argument. # The full rendered request body is checked again after provider preview. - validate_lark_text_request_size({"content": json.dumps({"text": reply_text}, ensure_ascii=False, separators=(",", ":"))}) + if not markdown: + validate_lark_text_request_size({"content": json.dumps({"text": reply_text}, ensure_ascii=False, separators=(",", ":"))}) if not reply_text: raise ValueError("lark inbox reply requires non-empty text") @@ -550,22 +552,31 @@ def verify_source() -> tuple[bool, bool]: api_calls = preview_data.get("api") for call in api_calls if isinstance(api_calls, list) else []: if isinstance(call, Mapping) and isinstance(call.get("body"), Mapping): - validate_lark_text_request_size(call["body"]) + if not markdown: + validate_lark_text_request_size(call["body"]) body_bytes = len(json.dumps(call["body"], ensure_ascii=False, separators=(",", ":")).encode("utf-8")) if markdown and body_bytes > LARK_POST_REQUEST_MAX_BYTES: - if attachment_keys: - raise ValueError("result file post exceeds the provider message bound") - # No write has occurred. Preserve the full answer via the existing - # larger text transport rather than truncate or retry after send. - result = _deliver_lark_inbox_outbound( - project=project, config_path=config_path, message_id=message_id, - text=text, content_format="text", execute=execute, - provider_preflight=provider_preflight, runner=runner, before_send=before_send, - source_membership_verifier=source_membership_verifier, - delivery_attempt_recorder=delivery_attempt_recorder, - ) - result.update(content_format="text", format_fallback="post_size_limit") - return result + from .post_delivery import deliver_post_parts + from .presentation.markdown_parts import split_lark_markdown_posts + + parts = split_lark_markdown_posts(reply_text, call["body"]) + if len(parts) < 2: + raise LarkOutboundTextError("provider post serialization cannot fit the rich part plan") + if not execute: + return {**_result(status="preview_ready", ok=True, execute=False, + receipt=receipt, placement=placement, format_preflight_passed=True, + provider_preview_performed=True, provider_preview_verified=True), + "content_format": "markdown", "delivery_part_count": len(parts)} + return deliver_post_parts(receipt=receipt, + intent_digest=_intent_digest(profile, chat_id, receipt), + template=call["body"], reply_text=reply_text, arguments={ + "project": project, "config_path": config_path, "message_id": message_id, + "runner": runner, "before_send": before_send, + "delivery_attempt_recorder": delivery_attempt_recorder, + "source_membership_verifier": source_membership_verifier, + "source_authority_verifier": source_authority_verifier, + "finalize_reactions": finalize_reactions, + "attachment_keys": attachment_keys, "attachment_names": attachment_names}) guidance = None if before_send is not None: # Bind review to destination/profile as well as content and placement. @@ -608,7 +619,7 @@ def verify_source() -> tuple[bool, bool]: provider_args, ) if send.get("returncode") != 0: - return _result( + return {**_result( status="gate_required", ok=False, execute=True, @@ -620,7 +631,7 @@ def verify_source() -> tuple[bool, bool]: format_preflight_passed=True, provider_preview_performed=True, provider_preview_verified=True, - ) + ), "provider_write_attempted": True} intent_digest = _intent_digest(profile, chat_id, receipt) reply_message_id = _message_id(_json_object(send.get("stdout"))) @@ -855,9 +866,17 @@ def verify_lark_inbox_reply( finalize_reactions: bool = True, attachment_keys: tuple[str, ...] = (), attachment_names: tuple[str, ...] = (), + resume_parts: bool = False, + before_send: Callable[[str], Mapping[str, Any]] | None = None, ) -> dict[str, Any]: - """Read back one prior Lark reply without sending another message.""" + """Verify prior writes; optionally resume never-sent rich parts. + Readback is the default. Only the original delivery owner opts into tail + continuation, retaining its source checks and outbound review hook. + """ + + from .post_delivery import POST_PARTS_LOCATOR, verify_post_parts + batch = POST_PARTS_LOCATOR.fullmatch(str(attempt.get("message_ref") or "")) if ( set(attempt) != { "schema_version", @@ -868,7 +887,7 @@ def verify_lark_inbox_reply( } or attempt.get("schema_version") != "manager_return_delivery_attempt_v0" or attempt.get("provider") != "lark" - or not MESSAGE_ID_PATTERN.fullmatch(str(attempt.get("message_ref") or "")) + or (not batch and not MESSAGE_ID_PATTERN.fullmatch(str(attempt.get("message_ref") or ""))) ): return { "ok": False, @@ -880,13 +899,30 @@ def verify_lark_inbox_reply( reply_config = config["reply"] profile = str(reply_config.get("sender_profile") or "") chat_id = str(reply_config.get("chat_id") or "") + if batch: + original = reply_lark_event_inbox(project=project, config_path=config_path, + message_id=message_id, text=text, content_format="markdown", attachment_keys=attachment_keys) + receipt = original["idempotency_key"] + return verify_post_parts(receipt=receipt, intent_digest=_intent_digest(profile, chat_id, receipt), + attempt=attempt, resume=resume_parts, arguments={"project": project, + "config_path": config_path, "message_id": message_id, "runner": runner, + "source_membership_verifier": source_membership_verifier, + "source_authority_verifier": source_authority_verifier, + "before_send": before_send, "finalize_reactions": finalize_reactions, + "text": normalize_lark_outbound_text(text, limit=None, preserve_format=True), + "attachment_keys": attachment_keys, "attachment_names": attachment_names}) preview = None - for content_format in (("markdown",) if attachment_keys else ("markdown", "text")): + candidates = [(text, "markdown")] + if not attachment_keys: + # Historical text receipts are read-only compatibility, never a new + # downgrade from an oversized Markdown reply. + candidates.append((text, "text")) + for candidate_text, content_format in candidates: candidate = reply_lark_event_inbox( project=project, config_path=config_path, message_id=message_id, - text=text, + text=candidate_text, content_format=content_format, execute=False, runner=runner, @@ -899,6 +935,7 @@ def verify_lark_inbox_reply( == _intent_digest(profile, chat_id, receipt) ): preview = candidate + expected_text = candidate_text break if preview is None: return { @@ -922,7 +959,7 @@ def verify_lark_inbox_reply( "blocker": "provider_verification_unavailable", } markdown = preview.get("content_format") == "markdown" - reply_text = normalize_lark_outbound_text(text, limit=None, preserve_format=markdown) + reply_text = normalize_lark_outbound_text(expected_text, limit=None, preserve_format=markdown) bot_aliases, mention_error = _verified_mention_aliases( runner=runner, base=base, chat_id=chat_id, reply_text=reply_text ) @@ -1025,6 +1062,8 @@ def update_lark_inbox_reply( exact source audience, not only group membership. Reuse that one fresh verification throughout this bounded provider operation. """ + from .post_delivery import POST_PARTS_LOCATOR + config = load_lark_event_inbox_config(project=project, config_path=config_path) reply = config["reply"] profile, chat_id = str(reply["sender_profile"]), str(reply["chat_id"]) @@ -1035,10 +1074,15 @@ def update_lark_inbox_reply( if (set(attempt) != {"schema_version", "provider", "message_ref", "intent_digest", "provider_receipt"} or attempt.get("schema_version") != "manager_return_delivery_attempt_v0" or attempt.get("provider") != "lark" - or not MESSAGE_ID_PATTERN.fullmatch(str(attempt.get("message_ref") or "")) + or not (MESSAGE_ID_PATTERN.fullmatch(str(attempt.get("message_ref") or "")) + or POST_PARTS_LOCATOR.fullmatch(str(attempt.get("message_ref") or ""))) or attempt.get("provider_receipt") != prior_receipt or attempt.get("intent_digest") != _intent_digest(profile, chat_id, prior_receipt)): return {"ok": False, "reply_verified": False, "blocker": "provider_delivery_intent_conflict"} + if POST_PARTS_LOCATOR.fullmatch(str(attempt["message_ref"])): + # A sequence has no single remote message to replace. The caller may + # verify and retire that sequence before creating a separate final. + return {"ok": False, "reply_verified": False, "blocker": "provider_update_format_unsupported"} desired = reply_lark_event_inbox(project=project, config_path=config_path, message_id=message_id, text=text, content_format="markdown", execute=False) # Rich mentions use the existing identity-verified text sender instead. diff --git a/loopx/extensions/lark/manager_reply_format.py b/loopx/extensions/lark/manager_reply_format.py index aefaf48ee7..244ce48c32 100644 --- a/loopx/extensions/lark/manager_reply_format.py +++ b/loopx/extensions/lark/manager_reply_format.py @@ -1,11 +1,9 @@ """Repair the mechanically fixable rich-text defects in a manager answer. -A steward answer reaches Lark as provider text. When the model or the -transport hands over the presentation as one line of literal ``\\n`` tokens, the -strict outbound validator rejects the whole message and the recovery path -degrades the answer to plain text: the reader loses the markdown bullet list -and the delivery is recorded as degraded. An unresolved template placeholder -such as ``{new_description}`` passes that validator and reaches the reader as +A steward answer reaches Lark as a rich post. When presentation contains +literal ``\\n`` tokens outside code, repair the mechanical defect before +validation so Markdown lists and tables retain their rich presentation. +An unresolved template placeholder such as ``{new_description}`` passes that validator and reaches the reader as raw braces. This module owns the repair that runs *before* the first send: escaped newlines diff --git a/loopx/extensions/lark/manager_reply_parts.py b/loopx/extensions/lark/manager_reply_parts.py index 6dcebc62d3..c54ef6f1b7 100644 --- a/loopx/extensions/lark/manager_reply_parts.py +++ b/loopx/extensions/lark/manager_reply_parts.py @@ -13,8 +13,8 @@ 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. +in the same record. The next attempt verifies that locator before sending +any further part, 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 @@ -33,16 +33,8 @@ 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: -# past this many parts the answer keeps its leading parts and ends with a note -# naming where the full text is already saved. -MANAGER_REPLY_MAX_PARTS = 8 -MANAGER_REPLY_OVERFLOW_NOTE = ( - "本条答复超过可投递长度，上面已按顺序发送前面的部分；" - "完整答复保存在 LoopX 管家会话中。" -) +from .presentation.markdown_parts import split_lark_markdown_posts + PART_DELIVERY_COMPLETE_KEY = "delivery_parts_complete" PART_DELIVERY_VERIFIED_KEY = "delivery_parts_verified" PART_ATTEMPT_KEY = "delivery_part_attempt" @@ -57,24 +49,15 @@ # provider hiccup never turns into a message the reader did not need. PART_STALL_NOTICE_MIN_STALLS = 3 MANAGER_REPLY_STALL_NOTICE = ( - "本条答复超过可发送长度，目前只发出了前面的 {sent}/{count} 段；" + "本条答复已拆分，目前已核验前面的 {sent}/{count} 段；" "完整答复保存在 LoopX 管家会话中，剩余分段会继续重试。" ) def plan_manager_reply_parts(reply_text: str) -> tuple[list[str], bool]: - """Return the parts to deliver and whether the remainder was replaced.""" + """Return complete rich parts; the legacy truncation flag is always false.""" - parts = split_lark_outbound_text( - reply_text, - limit=DEFAULT_LARK_TEXT_LIMIT, - max_parts=MANAGER_REPLY_MAX_PARTS, - overflow_note=MANAGER_REPLY_OVERFLOW_NOTE, - ) - truncated = len(parts) == MANAGER_REPLY_MAX_PARTS and ( - MANAGER_REPLY_OVERFLOW_NOTE in parts[-1] - ) - return parts, truncated + return split_lark_markdown_posts(reply_text, {"msg_type": "post", "uuid": "loopx-" + "0" * 32}), False def _reply_verified(reply: Mapping[str, Any]) -> bool: @@ -142,7 +125,7 @@ def completed_part_delivery_receipt( "ok": True, "status": "sent_verified", "idempotency_key": key, - "content_format": "text", + "content_format": delivery_state.get("content_format", "text"), "external_write_performed": True, "verification_performed": True, "reply_verified": True, @@ -239,13 +222,12 @@ def reconciled_part_reply( root: Path, config_path: Path, message_id: str, + reply_options: Mapping[str, Any] | None = None, ) -> 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. + Return ``None`` only when no write was recorded. A known write stays pending + until readback confirms it; verification performs no write of its own. """ attempt = recorded_part_attempt(delivery_state, index) @@ -264,9 +246,11 @@ def reconciled_part_reply( text=parts[index], attempt=attempt, runner=reply_runner, + **dict(reply_options or {}), ) if verified.get("reply_verified") is not True: - return None + # A known write with unavailable/mismatched readback is never unsent. + return verified return {**dict(verified), "part_reconciled": True} @@ -278,6 +262,7 @@ def reconciled_stall_notice( root: Path, config_path: Path, message_id: str, + reply_options: Mapping[str, Any] | None = None, ) -> Mapping[str, Any] | None: """Confirm a previously sent stall notice instead of posting it twice. @@ -299,19 +284,18 @@ def reconciled_stall_notice( text=notice, attempt=attempt, runner=reply_runner, + **{key: value for key, value in (reply_options or {}).items() if key != "before_send"}, ) if verified.get("reply_verified") is not True: - return None + return verified 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 + A reader who received only the first parts needs to know the answer is + incomplete. This notice states what was delivered and where the rest is, and it waits for three stalled attempts so an ordinary hiccup stays quiet. """ @@ -343,6 +327,7 @@ def deliver_stall_notice( root: Path, config_path: Path, message_id: str, + reply_options: Mapping[str, Any] | None = None, ) -> Mapping[str, Any] | None: """Post the once-per-sequence stall notice, confirming a prior send first. @@ -360,6 +345,7 @@ def deliver_stall_notice( root=root, config_path=config_path, message_id=message_id, + reply_options=reply_options, ) if reconciled is not None: if reconciled.get("notice_reconciled") is not True: @@ -372,6 +358,8 @@ def deliver_stall_notice( # here on, so the attempt that proved it is no longer needed. delivery_state.pop(PART_STALL_NOTICE_ATTEMPT_KEY, None) return reconciled + if delivery_state.get("delivery_part_stall_notice_started") is True: + return _locator_unavailable_result(reconciled_key="notice_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 @@ -385,16 +373,24 @@ def record_attempt(attempt: Mapping[str, Any]) -> None: delivery_state["updated_at"] = datetime.now(timezone.utc).isoformat() write_delivery(delivery_path, delivery_state) - return reply_lark_event_inbox( + delivery_state["delivery_part_stall_notice_started"] = True + write_delivery(delivery_path, delivery_state) + result = reply_lark_event_inbox( project=root, config_path=config_path, message_id=message_id, text=notice, - content_format="text", + content_format="markdown", execute=True, runner=reply_runner, delivery_attempt_recorder=record_attempt, + **dict(reply_options or {}), ) + if (result.get("external_write_performed") is not True + and result.get("provider_write_attempted") is not True): + delivery_state.pop("delivery_part_stall_notice_started", None) + write_delivery(delivery_path, delivery_state) + return result def deliver_manager_reply_parts( @@ -408,6 +404,7 @@ def deliver_manager_reply_parts( config_path: Path, message_id: str, content_format: str, + reply_options=None, ) -> Mapping[str, Any] | None: """Send the remaining parts in order, resuming from the recorded count. @@ -439,7 +436,8 @@ def deliver_manager_reply_parts( delivery_state.update( delivery_part_count=len(parts), delivery_parts_sent=sent, - format_degraded=True, + format_degraded=content_format != "markdown", + content_format=content_format, updated_at=datetime.now(timezone.utc).isoformat(), ) # A sequence that is still being sent is not a complete one, even when a @@ -448,6 +446,7 @@ def deliver_manager_reply_parts( write_delivery(delivery_path, delivery_state) last: Mapping[str, Any] | None = None for index in range(sent, len(parts)): + options = dict(reply_options(index) if reply_options is not None else {}) last = reconciled_part_reply( parts=parts, index=index, @@ -456,8 +455,13 @@ def deliver_manager_reply_parts( root=root, config_path=config_path, message_id=message_id, + reply_options={key: value for key, value in options.items() if key != "before_send"}, ) if last is None: + if delivery_state.get("delivery_part_started") == index: + # A crash between a provider write and its locator is ambiguous. + # The original intent stays recoverable; do not post it again. + return 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) @@ -467,11 +471,14 @@ def record_attempt(attempt: Mapping[str, Any], *, index=index) -> None: "index": index, "attempt": dict(attempt), } + delivery_state.setdefault("delivery_part_attempts", {})[str(index)] = dict(attempt) delivery_state["updated_at"] = datetime.now( timezone.utc ).isoformat() write_delivery(delivery_path, delivery_state) + delivery_state["delivery_part_started"] = index + write_delivery(delivery_path, delivery_state) last = reply_lark_event_inbox( project=root, config_path=config_path, @@ -484,7 +491,12 @@ def record_attempt(attempt: Mapping[str, Any], *, index=index) -> None: # Part delivery is the fallback for a body the provider itself # cannot take; a part is never cut by the notification length. short_message_limit=None, + **options, ) + if last.get("external_write_performed") is not True and last.get("provider_write_attempted") is not True: + delivery_state.pop("delivery_part_started", None) + elif last.get("provider_write_attempted") is True: + delivery_state["delivery_write_attempted"] = True if not _part_accepted(last): delivery_state.update( delivery_parts_sent=index, @@ -494,6 +506,7 @@ def record_attempt(attempt: Mapping[str, Any], *, index=index) -> None: write_delivery(delivery_path, delivery_state) return None delivery_state.pop(PART_ATTEMPT_KEY, None) + delivery_state.pop("delivery_part_started", None) delivery_state.update( delivery_parts_sent=index + 1, **( @@ -518,21 +531,9 @@ def deliver_manager_reply_after_length_failure( config_path: Path, message_id: str, ) -> tuple[Mapping[str, Any] | None, str | None]: - """Deliver one over-limit manager answer as bounded parts. + """Deliver the complete answer as ordered rich posts, with recovery.""" - Returns the last accepted reply, or ``None`` plus the reason to report when - 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) - if truncated: - delivery_state.update( - delivery_truncated=True, - delivery_source_char_count=len(reply_text), - ) + parts, _ = plan_manager_reply_parts(reply_text) reply = deliver_manager_reply_parts( parts=parts, delivery_state=delivery_state, @@ -542,7 +543,7 @@ def deliver_manager_reply_after_length_failure( root=root, config_path=config_path, message_id=message_id, - content_format="text", + content_format="markdown", ) if reply is not None: return reply, None @@ -591,7 +592,7 @@ def manager_part_delivery_pending_result( "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, + "format_degraded": delivery_state.get("content_format") != "markdown", "goal_id": goal_id, "inbox_config_ref": inbox_config_ref, "source_acknowledged": False, diff --git a/loopx/extensions/lark/manager_returns.py b/loopx/extensions/lark/manager_returns.py index dadf3ed1a6..00bc1f7635 100644 --- a/loopx/extensions/lark/manager_returns.py +++ b/loopx/extensions/lark/manager_returns.py @@ -201,8 +201,11 @@ def record_attempt(value): if attachments and result.get("reply_verified") is True: # The recorder persists the exact message locator before this read. # Failure retains that attempt; recovery only reads it, never resends. + from .post_delivery import attachment_reply + file_text, file_attempt = attachment_reply(project=project, config_path=config_path, + attempt=observed_attempt, text=text) verified = verify_result_files(profile=reply["sender_profile"], - message_id=observed_attempt["message_ref"], text=text, attachments=attachments, keys=keys, + message_id=file_attempt["message_ref"], text=file_text, attachments=attachments, keys=keys, runner=runner, private_transport=private_transport) result.update(verified) if verified.get("reply_verified") is True: @@ -226,7 +229,7 @@ def verify_return( private_transport: Any = None, ) -> dict[str, Any]: text = present_lark_conversation_text(text) - project, config_path, source_verifier, _destination, message_id = _return_inbox( + project, config_path, source_verifier, destination, message_id = _return_inbox( root=root, registry=registry, snapshot_provider=snapshot_provider, @@ -238,6 +241,12 @@ def verify_return( ) if runner is None and isinstance(session.get("steward_context"), dict): runner = private_transport._reply_runner + def before_resume(_intent: str) -> dict[str, bool]: + current = _return_inbox(root=root, registry=registry, snapshot_provider=snapshot_provider, + route=route, session=session, turn=turn, cancelled=cancelled, private_transport=private_transport) + return {"continue_delivery": current[:2] == (project, config_path) + and current[3:] == (destination, message_id)} + runner_kwargs: dict[str, Any] = {"runner": runner} if runner else {} attachments = route.get("result_attachments") or [] keys = () @@ -261,11 +270,16 @@ def verify_return( attachment_keys=keys, attachment_names=tuple(attachment["name"] for attachment in attachments), finalize_reactions=not attachments, + resume_parts=True, + before_send=before_resume, **runner_kwargs, ) if attachments and result.get("reply_verified") is True: + from .post_delivery import attachment_reply + file_text, file_attempt = attachment_reply(project=project, config_path=config_path, + attempt=attempt, text=text) result.update(verify_result_files(profile=reply["sender_profile"], - message_id=attempt["message_ref"], text=text, attachments=attachments, keys=keys, + message_id=file_attempt["message_ref"], text=file_text, attachments=attachments, keys=keys, runner=runner, private_transport=private_transport)) if result.get("reply_verified") is True: result.update(_verified_reply_result(project=project, config_path=config_path, diff --git a/loopx/extensions/lark/post_delivery.py b/loopx/extensions/lark/post_delivery.py new file mode 100644 index 0000000000..c27cd43169 --- /dev/null +++ b/loopx/extensions/lark/post_delivery.py @@ -0,0 +1,288 @@ +"""Rich-post continuation through the existing Lark part-delivery journal. + +The composite locator names private provider IO receipts, not a remote message +or a second Goal/work authority. Each concrete part keeps its own exact receipt. +""" + +from __future__ import annotations + +import json +import re +from pathlib import Path +from typing import Any + +from ...file_lock import exclusive_file_lock +from .event_inbox import load_lark_event_inbox_config +from .manager_reply_delivery import write_delivery +from .presentation.markdown_parts import split_lark_markdown_posts + +POST_PARTS_LOCATOR = re.compile(r"^lark-parts:([a-f0-9]{64})$") + + +def _path(project, config_path, intent_digest): + config = load_lark_event_inbox_config(project=project, config_path=config_path) + return ( + Path(config["inbox_path"]) + / "manager-delivery" + / "parts" + / f"{intent_digest.removeprefix('sha256:')}.json" + ) + + +def _batch_attempt(receipt: str, intent_digest: str) -> dict[str, Any]: + return { + "schema_version": "manager_return_delivery_attempt_v0", + "provider": "lark", + "message_ref": "lark-parts:" + intent_digest.removeprefix("sha256:"), + "intent_digest": intent_digest, + "provider_receipt": receipt, + } + + +def _result(state, receipt, verified=False, blocker=None): + return { + "ok": verified, + "status": "sent_verified" if verified else "sent_unverified", + "idempotency_key": receipt, + "content_format": "markdown", + "format_degraded": False, + "delivery_part_count": len(state["delivery_parts"]), + "delivery_parts_sent": state.get("delivery_parts_sent", 0), + "delivery_notice_sent": state.get("delivery_part_stall_notice") is True, + "external_write_performed": bool( + state.get("delivery_part_attempts") or state.get("delivery_write_attempted") + ), + "verification_performed": verified, + "reply_verified": verified, + **({"blocker": blocker} if blocker else {}), + } + + +def deliver_post_parts(*, receipt, intent_digest, template, reply_text, arguments): + from .manager_reply_parts import deliver_manager_reply_parts, deliver_stall_notice + from .inbox_reply import _verified_reply_result + + arguments = {**arguments, "text": reply_text or arguments.get("text", "")} + path = _path(arguments["project"], arguments["config_path"], intent_digest) + with exclusive_file_lock( + path.with_suffix(".lock"), agent_id="lark", operation="reply_parts" + ): + if path.is_file(): + state = json.loads(path.read_text(encoding="utf-8")) + if ( + state.get("intent_digest") != intent_digest + or state.get("provider_receipt") != receipt + ): + raise ValueError("rich reply delivery intent changed") + expected = reply_text or arguments.get("text", "") + if state["delivery_parts"] != split_lark_markdown_posts( + expected, state["request_template"] + ): + raise ValueError("rich reply part plan changed") + else: + state = { + "intent_digest": intent_digest, + "provider_receipt": receipt, + "delivery_parts": split_lark_markdown_posts(reply_text, template), + "request_template": dict(template), + "content_format": "markdown", + } + write_delivery(path, state) + if state.get("delivery_parts_complete"): + if arguments.get("delivery_attempt_recorder") is not None: + arguments["delivery_attempt_recorder"]( + _batch_attempt(receipt, intent_digest) + ) + return verify_post_parts( + receipt=receipt, + intent_digest=intent_digest, + attempt=_batch_attempt(receipt, intent_digest), + arguments=arguments, + ) + recorder = arguments.get("delivery_attempt_recorder") + outer_hook = arguments.get("before_send") + if outer_hook and (bind := getattr(outer_hook, "for_destination", None)): + config = load_lark_event_inbox_config( + project=arguments["project"], config_path=arguments["config_path"] + ) + outer_hook = bind(str(config["reply"]["chat_id"])) + + def record(path, value): + write_delivery(path, value) + if ( + value.get("delivery_part_attempts") + or value.get("delivery_write_attempted") + ) and recorder is not None: + recorder(_batch_attempt(receipt, intent_digest)) + + def options(index): + final = index == len(state["delivery_parts"]) - 1 + return { + "source_membership_verifier": arguments.get( + "source_membership_verifier" + ), + "source_authority_verifier": arguments.get("source_authority_verifier"), + "before_send": (lambda _part_intent: outer_hook(intent_digest)) + if outer_hook + else None, + "finalize_reactions": False, + "attachment_keys": arguments.get("attachment_keys", ()) + if final + else (), + "attachment_names": arguments.get("attachment_names", ()) + if final + else (), + } + + result = deliver_manager_reply_parts( + parts=state["delivery_parts"], + delivery_state=state, + delivery_path=path, + write_delivery=record, + reply_runner=arguments["runner"], + root=Path(arguments["project"]), + config_path=Path(arguments["config_path"]), + message_id=arguments.get("message_id"), + content_format="markdown", + reply_options=options, + ) + if result is None or result.get("reply_verified") is not True: + state["delivery_part_stall_count"] = ( + int(state.get("delivery_part_stall_count") or 0) + 1 + ) + notice_options = options(0) + notice_options.update(attachment_keys=(), attachment_names=()) + notice = deliver_stall_notice( + delivery_state=state, + delivery_path=path, + write_delivery=record, + reply_runner=arguments["runner"], + root=Path(arguments["project"]), + config_path=Path(arguments["config_path"]), + message_id=arguments.get("message_id"), + reply_options=notice_options, + ) + if notice and notice.get("reply_verified") is True: + state["delivery_part_stall_notice"] = True + record(path, state) + return _result(state, receipt, blocker="lark_inbox_reply_not_verified") + complete = _verified_reply_result( + project=arguments["project"], + config_path=arguments["config_path"], + message_id=arguments.get("message_id"), + runner=arguments["runner"], + finalize_reactions=bool(arguments.get("message_id")) + and arguments.get("finalize_reactions", True), + ) + return {**_result(state, receipt, True), **complete} + + +def verify_post_parts(*, receipt, intent_digest, attempt, arguments, resume=False): + from .inbox_reply import verify_lark_inbox_reply, _verified_reply_result + + if dict(attempt) != _batch_attempt(receipt, intent_digest): + return { + "ok": False, + "verification_performed": True, + "reply_verified": False, + "blocker": "provider_delivery_intent_conflict", + } + path = _path(arguments["project"], arguments["config_path"], intent_digest) + if not path.is_file(): + return { + "ok": False, + "verification_performed": False, + "reply_verified": False, + "blocker": "provider_verification_unavailable", + } + state = json.loads(path.read_text(encoding="utf-8")) + if ( + state.get("intent_digest") != intent_digest + or state.get("provider_receipt") != receipt + ): + return { + "ok": False, + "verification_performed": True, + "reply_verified": False, + "blocker": "provider_delivery_intent_conflict", + } + if state["delivery_parts"] != split_lark_markdown_posts( + arguments["text"], state["request_template"] + ): + return { + "ok": False, + "verification_performed": True, + "reply_verified": False, + "blocker": "provider_delivery_intent_conflict", + } + parts = state["delivery_parts"] + for index, part in enumerate(parts): + prior = state.get("delivery_part_attempts", {}).get(str(index)) + if prior is None: + if state.get("delivery_part_started") == index: + return { + "ok": False, + "verification_performed": True, + "reply_verified": False, + "blocker": "provider_message_missing", + } + if resume: + return deliver_post_parts( + receipt=receipt, + intent_digest=intent_digest, + template={}, + reply_text="", + arguments=arguments, + ) + return _result(state, receipt, blocker="provider_verification_unavailable") + final = index == len(parts) - 1 + result = verify_lark_inbox_reply( + project=arguments["project"], + config_path=arguments["config_path"], + message_id=arguments.get("message_id"), + text=part, + attempt=prior, + runner=arguments["runner"], + source_membership_verifier=arguments.get("source_membership_verifier"), + source_authority_verifier=arguments.get("source_authority_verifier"), + finalize_reactions=False, + attachment_keys=arguments.get("attachment_keys", ()) if final else (), + attachment_names=arguments.get("attachment_names", ()) if final else (), + ) + if not result.get("reply_verified"): + return result + if resume and state.get("delivery_parts_sent", 0) < len(parts): + return deliver_post_parts( + receipt=receipt, + intent_digest=intent_digest, + template={}, + reply_text="", + arguments=arguments, + ) + return { + **_result(state, receipt, True), + **_verified_reply_result( + project=arguments["project"], + config_path=arguments["config_path"], + message_id=arguments.get("message_id"), + runner=arguments["runner"], + finalize_reactions=bool(arguments.get("message_id")) + and arguments.get("finalize_reactions", True), + ), + } + + +def attachment_reply(*, project, config_path, attempt, text): + """Files belong to the last concrete post, never to the composite locator.""" + match = POST_PARTS_LOCATOR.fullmatch(str(attempt.get("message_ref") or "")) + if not match: + return text, attempt + state = json.loads( + _path(project, config_path, "sha256:" + match[1]).read_text(encoding="utf-8") + ) + if state.get("intent_digest") != attempt.get("intent_digest") or state.get( + "provider_receipt" + ) != attempt.get("provider_receipt"): + raise ValueError("rich reply attachment intent changed") + index = len(state["delivery_parts"]) - 1 + return state["delivery_parts"][index], state["delivery_part_attempts"][str(index)] diff --git a/loopx/extensions/lark/presentation/markdown_parts.py b/loopx/extensions/lark/presentation/markdown_parts.py new file mode 100644 index 0000000000..72d7a6923f --- /dev/null +++ b/loopx/extensions/lark/presentation/markdown_parts.py @@ -0,0 +1,217 @@ +"""Split post presentation against the actual provider request envelope.""" + +from __future__ import annotations + +import json +import re +from collections.abc import Mapping +from typing import Any + +from .markdown_post import lark_markdown_post_content +from ..outbound import LARK_POST_REQUEST_MAX_BYTES, LarkOutboundTextError + + +def post_request_bytes(text: str, template: Mapping[str, Any]) -> int: + content = json.loads(lark_markdown_post_content(text)) + authored = template.get("content", {}) + if isinstance(authored, str): + parsed = json.loads(authored) + # Preserve the provider preview's encoding and whitespace, including + # attachment merging. Escaped Unicode can cost twice its UTF-8 form. + old_text = parsed.get("zh_cn", {}).get("content", [[{}]])[0][0].get("text") + if isinstance(old_text, str): + new_text = content["zh_cn"]["content"][0][0]["text"] + for ascii_only in (False, True): + old_literal = json.dumps(old_text, ensure_ascii=ascii_only) + if old_literal in authored: + replacement = json.dumps(new_text, ensure_ascii=ascii_only) + body = { + **template, + "content": authored.replace(old_literal, replacement, 1), + } + return len( + json.dumps( + body, ensure_ascii=False, separators=(",", ":") + ).encode("utf-8") + ) + authored = parsed + if isinstance(authored, Mapping): + content.update( + {key: value for key, value in authored.items() if key != "zh_cn"} + ) + body = { + **template, + "content": json.dumps(content, ensure_ascii=False, separators=(",", ":")), + } + return len( + json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + ) + + +def _blocks(text: str) -> list[str]: + """Keep ordinary paragraphs, code blocks and tables together when possible.""" + lines = text.splitlines(keepends=True) + result: list[str] = [] + index = 0 + while index < len(lines): + start = index + fence = re.match(r"^ {0,3}(`{3,}|~{3,})", lines[index]) + if fence: + index += 1 + while index < len(lines): + close = re.match(r"^ {0,3}(`{3,}|~{3,})\s*$", lines[index]) + index += 1 + if ( + close + and close[1][0] == fence[1][0] + and len(close[1]) >= len(fence[1]) + ): + break + else: + index += 1 + while ( + index < len(lines) + and lines[index].strip() + and not re.match(r"^ {0,3}(`{3,}|~{3,})", lines[index]) + ): + index += 1 + while index < len(lines) and not lines[index].strip(): + index += 1 + result.append("".join(lines[start:index])) + return result + + +def _protected_spans(text: str) -> list[tuple[int, int]]: + # Do not cut a normal link, escaped character, inline code or emphasis. + return [ + (m.start(), m.end()) + for m in re.finditer( + r"\\.|(`+)(?:(?!\1).)*?\1|!?\[[^\]\n]*\]\((?:\\.|[^)\n])*\)|\*\*.+?\*\*|__.+?__", + text, + ) + ] + + +def _large_block(block: str, fits, first_prefix: str = "") -> list[str]: + """Repeat only presentation context for oversized code/table blocks.""" + lines = block.splitlines(keepends=True) + prefix = suffix = "" + fence = re.match(r"^ {0,3}(`{3,}|~{3,})[^\n]*\n", block) + if fence: + prefix = fence[0] + close = re.search( + r"(?m)^ {0,3}" + re.escape(fence[1]) + r"\s*$", block[fence.end() :] + ) + if close: + body = block[fence.end() : fence.end() + close.start()] + suffix = "\n" + fence[1] + else: + body = block[fence.end() :] + suffix = "\n" + fence[1] + elif len(lines) >= 3 and "|" in lines[0] and re.fullmatch(r"[ |:\-]+\n?", lines[1]): + # Each continuation repeats the column names, not the data rows. + prefix = "".join(lines[:2]) + rows, current = [], "" + + def row_fits(value): + return fits((first_prefix if not rows else "") + value) + + for line in lines[2:]: + if not fits(prefix + line): + raise LarkOutboundTextError( + "one rich table row exceeds the provider message bound" + ) + if current and not row_fits(prefix + current + line): + rows.append((first_prefix if not rows else "") + prefix + current) + current = "" + if not current and not row_fits(prefix + line) and not rows: + rows.append(first_prefix) + first_prefix = "" + current += line + if current: + rows.append((first_prefix if not rows else "") + prefix + current) + return rows + else: + body = block + wrapped = re.fullmatch(r"(\*\*|__|`+)([\s\S]+)\1\s*", block) + if wrapped: + prefix = suffix = wrapped[1] + body = wrapped[2] + result = [] + while body: + low, high = 0, len(body) + while low < high: + middle = (low + high + 1) // 2 + if fits( + (first_prefix if not result else "") + prefix + body[:middle] + suffix + ): + low = middle + else: + high = middle - 1 + cut = low + if fence: + newline = body.rfind("\n", 0, cut) + if newline >= 0: + cut = newline + 1 + else: + for start, end in _protected_spans(body): + if start < cut < end: + cut = start + break + if cut <= 0: + if first_prefix and not result: + # A protected element may fit a fresh post even when the + # preceding paragraph leaves too little room for it. + result.append(first_prefix) + first_prefix = "" + continue + raise LarkOutboundTextError( + "one rich inline element exceeds the provider message bound" + ) + result.append( + (first_prefix if not result else "") + prefix + body[:cut] + suffix + ) + body = body[cut:] + return result + + +def split_lark_markdown_posts(text: str, template: Mapping[str, Any]) -> list[str]: + """Use the provider's full byte budget, including JSON escapes and markers. + + No character ceiling, fixed part count, truncation or text fallback. The + ordinary single-post serializer and readback still own each message. + """ + if post_request_bytes(text, template) <= LARK_POST_REQUEST_MAX_BYTES: + return [text] + total = 2 + while True: + pieces: list[str] = [] + current = "" + + def fits(body: str) -> bool: + marked = f"（{total}/{total}）\n\n" + body + return post_request_bytes(marked, template) <= LARK_POST_REQUEST_MAX_BYTES + + for block in _blocks(text): + if fits(current + block): + current += block + continue + if not fits(block): + pieces.extend(_large_block(block, fits, first_prefix=current)) + current = "" + else: + if current: + pieces.append(current) + current = block + if current: + pieces.append(current) + if len(pieces) == total: + break + total = len(pieces) + parts = [f"（{i}/{total}）\n\n{part}" for i, part in enumerate(pieces, 1)] + if any( + post_request_bytes(part, template) > LARK_POST_REQUEST_MAX_BYTES + for part in parts + ): + raise LarkOutboundTextError("rich part exceeds the provider request bound") + return parts diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index e08a67296f..f213415c1c 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -880,7 +880,7 @@ def edit_attempt(value: Any) -> None: _atomic_write_json(path, record) return self._deliver(path, record, phase, text, inbox=inbox) elif phase_state.get("attempt"): - result = verify_lark_inbox_reply(**kwargs, attempt=phase_state["attempt"]) + result = verify_lark_inbox_reply(**kwargs, attempt=phase_state["attempt"], resume_parts=True) elif phase_state.get("started"): # A crash between provider write and its receipt is ambiguous. Keep # the original source recoverable; never resend blindly. @@ -946,6 +946,8 @@ def before_send(_intent: str) -> dict[str, bool]: return result.get("ok") is True def _close_progress(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], Path]) -> bool: + if self._retire_progress_parts(path, record, inbox=inbox): + return True text = "回答已生成，完整结果将单独发送。" if not self._edit_progress(path, record, text, inbox=inbox, finalizing=True): return False @@ -953,6 +955,27 @@ def _close_progress(self, path: Path, record: dict[str, Any], *, inbox: Callable _atomic_write_json(path, record) return True + def _retire_progress_parts(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], Path]) -> bool: + """Verify every concrete draft post before retiring a split draft.""" + from .post_delivery import POST_PARTS_LOCATOR + + progress = record["deliveries"].get(presentation_phase(record, "progress")) or {} + stream = presentation_stream(record) + attempt = stream.get("confirmed_attempt") or progress.get("attempt") or {} + if not POST_PARTS_LOCATOR.fullmatch(str(attempt.get("message_ref") or "")): + return False + result = verify_lark_inbox_reply(project=self.runtime_root, config_path=inbox(), + message_id=record["event"]["message_id"], + text=stream.get("confirmed_text") or progress["text"], attempt=attempt, + runner=self._reply_runner, finalize_reactions=False, + source_authority_verifier=lambda profile, chat, message: self._source_authority(record, profile, chat, message)) + stream["blocker"] = result.get("blocker") + verified = result.get("reply_verified") is True and result.get("ok") is True + if verified: + stream["closed"] = True + _atomic_write_json(path, record) + return verified + def _continue_progress(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], Path]) -> bool: """Retire a verified draft before creating the next source-bound one. @@ -965,7 +988,7 @@ def _continue_progress(self, path: Path, record: dict[str, Any], *, inbox: Calla if not progress.get("verified"): return False stream = presentation_stream(record) - if not stream.get("edit_limit_reached"): + if not stream.get("edit_limit_reached") and not self._retire_progress_parts(path, record, inbox=inbox): closed = self._edit_progress(path, record, "仍在处理，后续进展见下一条消息。", inbox=inbox, finalizing=True) if not closed and not stream.get("edit_limit_reached"): diff --git a/tests/extensions/test_lark_goal_topic_runtime.py b/tests/extensions/test_lark_goal_topic_runtime.py index aeeaeb4085..667102e1b6 100644 --- a/tests/extensions/test_lark_goal_topic_runtime.py +++ b/tests/extensions/test_lark_goal_topic_runtime.py @@ -14,8 +14,6 @@ import pytest 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 @@ -197,9 +195,18 @@ def run(args: list[str]) -> dict[str, Any]: ] } if "--dry-run" in args - else {"data": {"message_id": "om_reply_fixture"}} + else {"data": {"message_id": "om_reply_fixture" if not state.get("messages") else f"om_reply_{len(state['messages'])}"}} ) + if "--dry-run" not in args: + mid = payload["data"]["message_id"] + state.setdefault("messages", {})[mid] = { + "message_id": mid, "msg_type": state["reply_type"], + "body": {"content": state["reply_content"]}, + } elif "+messages-mget" in args: + mid = args[args.index("--message-ids") + 1] + if state.get("messages"): + return {"returncode": 0, "stdout": json.dumps({"data": {"items": [state["messages"][mid]]}})} payload = { "data": { "items": [ @@ -2767,7 +2774,6 @@ def reply(args): runtime_root=tmp_path / "runtime", answer=answer, reply_runner=reply, ) result = runtime.process_lark_goal_topic_event(**kwargs) - sendable = len(body.encode("utf-8")) < 150_000 if body == r"private\nformat": # Escaped newlines are repaired before the first send, so the answer is # delivered as markdown (a real bullet list) and is not reported as a @@ -2786,26 +2792,21 @@ def reply(args): assert [ incident["code"] for incident in result.get("rich_text_repairs") or [] ] == expected_repairs - if sendable: - assert state["reply_text"] == expected - assert result["ok"] is reply_ok - assert result.get("format_degraded") is False - else: - # Longer than one deliverable message: the persisted answer is sent as a - # bounded, ordered sequence of parts that ends with the overflow note, - # instead of being left undelivered as `reply_format_invalid`. - assert result["ok"] is reply_ok - assert result.get("format_degraded") is True - if reply_ok: - assert result["delivery_part_count"] == MANAGER_REPLY_MAX_PARTS - assert result["delivery_parts_sent"] == MANAGER_REPLY_MAX_PARTS - assert state["reply_text"].startswith("(8/8) ") - assert MANAGER_REPLY_OVERFLOW_NOTE in state["reply_text"] + assert result["ok"] is reply_ok + assert result.get("format_degraded") is False + if reply_ok: + messages = list(state["messages"].values()) + bodies = [json.loads(m["body"]["content"])["zh_cn"]["content"][0][0]["text"] for m in messages] + assert all(m["msg_type"] == "post" for m in messages) + if len(messages) == 1: + assert bodies == [expected] else: - assert result["status"] == "reply_delivery_pending" - assert result["reason"] == "reply_part_delivery_incomplete" - assert result["delivery_parts_sent"] == 0 - assert result["source_acknowledged"] is False + assert "".join(part.split("\n\n", 1)[1] for part in bodies) == expected + assert result["delivery_part_count"] == len(messages) + assert result["delivery_parts_sent"] == len(messages) + else: + assert result["status"] == "reply_delivery_pending" + assert result["source_acknowledged"] is False pending = inspect_lark_event_inbox(project=kwargs["runtime_root"], config_path=Path(result["inbox_config_ref"])) if reply_ok: @@ -2934,7 +2935,7 @@ def no_duplicate_send(args): assert second["ok"] is True assert second["status"] == "replied_and_acknowledged" assert second["saved_response_reused"] is True - assert second["format_degraded"] is True + assert second["format_degraded"] is False receipt = json.loads(delivery_files[0].read_text()) assert receipt["status"] == "acknowledged" assert "delivery_text" not in receipt @@ -3127,15 +3128,17 @@ def interrupted_receipt_write(path, payload): ) 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 + journals = list((tmp_path / "runtime").glob("**/manager-delivery/parts/*.json")) + assert len(journals) == 1 + parts_state = json.loads(journals[0].read_text()) + part_count = len(parts_state["delivery_parts"]) + assert part_count > 1 + assert parts_state["delivery_parts_sent"] == part_count + assert parts_state[PART_DELIVERY_COMPLETE_KEY] is True + delivered_once = [call for call in state["calls"] + if "+messages-reply" in call and "--dry-run" not in call] + assert len(delivered_once) == part_count + assert all("--content" in call and "--text" not in call for call in delivered_once) monkeypatch.setattr(runtime, "_write_manager_delivery", real_write) second = runtime.process_lark_goal_topic_event(**kwargs) @@ -3153,7 +3156,7 @@ def interrupted_receipt_write(path, payload): 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 + assert settled["delivery_parts_sent"] == part_count pending = inspect_lark_event_inbox(project=kwargs["runtime_root"], config_path=config_path) assert pending["items"] == [] @@ -3244,12 +3247,13 @@ def ambiguous_runner(args: list[str]) -> dict[str, Any]: ) 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 - ] + journals = list((tmp_path / "runtime").glob("**/manager-delivery/parts/*.json")) + assert len(journals) == 1 + parts_state = json.loads(journals[0].read_text()) + part_count = len(parts_state["delivery_parts"]) + assert parts_state[PART_ATTEMPT_KEY]["index"] == 0 + sent_once = [call[call.index("--content") + 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 @@ -3259,15 +3263,15 @@ def ambiguous_runner(args: list[str]) -> dict[str, Any]: assert second["status"] == "replied_and_acknowledged" assert len(answered) == 1 sent_total = [ - call[call.index("--text") + 1] + call[call.index("--content") + 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 + assert len(sent_total) == part_count settled = json.loads(state_path.read_text()) assert settled["status"] == "acknowledged" - assert settled["delivery_parts_sent"] == MANAGER_REPLY_MAX_PARTS + assert settled["delivery_parts_sent"] == part_count assert PART_ATTEMPT_KEY not in settled @@ -3277,7 +3281,7 @@ def test_a_long_manager_answer_is_delivered_as_one_message( """A 3000-character steward answer is one message, not a bounded sequence. The self-imposed 1200-character cap is not a provider limit: the transport - already accepts up to the 150 KB provider bound, and every reader-visible + already accepts the rich-post provider byte bound, and every reader-visible split costs an extra message plus a part-sequence record. This pins the single-message outcome so a later change cannot silently reintroduce the split for an ordinary long answer. @@ -3395,12 +3399,12 @@ def manager_decision(**kwargs): working_runner = _reply_runner(state) def part_index(text: str) -> int | None: - match = re.match(r"\((\d+)/(\d+)\) ", text) + 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] + if "+messages-reply" in args and "--dry-run" in args: + text = json.loads(args[args.index("--content") + 1])["zh_cn"]["content"][0][0]["text"] index = part_index(text) if index is not None and index >= 3: return {"returncode": 1} @@ -3429,7 +3433,7 @@ def stalling_runner(args: list[str]) -> dict[str, Any]: def delivered_texts() -> list[str]: return [ - call[call.index("--text") + 1] + json.loads(call[call.index("--content") + 1])["zh_cn"]["content"][0][0]["text"] for call in state["calls"] if "+messages-reply" in call and "--dry-run" not in call ] @@ -3451,7 +3455,7 @@ def delivered_texts() -> list[str]: 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] + assert f"2/{third['delivery_part_count']}" in notices[0] fourth = runtime.process_lark_goal_topic_event(**kwargs) @@ -3466,7 +3470,8 @@ def delivered_texts() -> list[str]: message_id="om_incoming", ).read_text() ) - assert saved[PART_STALL_NOTICE_KEY] is True + parts_path = next((tmp_path / "runtime").glob("**/manager-delivery/parts/*.json")) + assert json.loads(parts_path.read_text())[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 854b291540..bdeaa60a4e 100644 --- a/tests/extensions/test_lark_manager_reply_parts.py +++ b/tests/extensions/test_lark_manager_reply_parts.py @@ -29,7 +29,7 @@ ) -BODY = "\n".join(f"- 条目 {index} " + "x" * 60 for index in range(1, 200)) +BODY = "\n".join(f"- 条目 {index} " + "x" * 60 for index in range(1, 2000)) @pytest.fixture @@ -350,8 +350,8 @@ def readback(**kwargs): assert PART_ATTEMPT_KEY not in delivery["state"] -def test_an_unconfirmed_locator_still_sends_the_part(monkeypatch, delivery): - """A locator the provider cannot confirm must not drop the reader's text.""" +def test_an_unconfirmed_locator_stays_pending_without_resending(monkeypatch, delivery): + """Missing readback is not proof that a recorded write never happened.""" parts, _ = plan_manager_reply_parts(BODY) delivery["state"].update( @@ -372,9 +372,10 @@ def test_an_unconfirmed_locator_still_sends_the_part(monkeypatch, delivery): ) monkeypatch.setattr(parts_module, "reply_lark_event_inbox", delivery["install"]()) - assert delivery["deliver"](parts)["ok"] is True + assert delivery["deliver"](parts) is None - assert delivery["sends"][0] == parts[0] + assert delivery["sends"] == [] + assert delivery["state"][PART_ATTEMPT_KEY]["attempt"] == ATTEMPT def test_a_confirmed_locator_settles_a_sequence_with_no_new_write( @@ -427,7 +428,7 @@ def test_a_stalled_sequence_tells_the_reader_once_after_real_retries( def failing_parts(**kwargs): sent_texts.append(kwargs["text"]) - if kwargs["text"].startswith("("): + if kwargs["text"].startswith("（"): return {"ok": False, "status": "reply_provider_failed", "idempotency_key": None} return {"ok": True, "status": "sent_verified", "idempotency_key": "sha256:n", @@ -547,7 +548,7 @@ def test_a_notice_the_provider_took_but_did_not_read_back_is_confirmed( def ambiguous_send(**kwargs): sends.append(kwargs["text"]) - if kwargs["text"].startswith("("): + if kwargs["text"].startswith("（"): return { "ok": False, "status": "reply_provider_failed", @@ -638,7 +639,7 @@ def test_a_stall_notice_without_a_locator_survives_its_own_writeback( def locator_less_send(**kwargs): sends.append(kwargs["text"]) - if kwargs["text"].startswith("("): + if kwargs["text"].startswith("（"): return { "ok": False, "status": "reply_provider_failed", @@ -697,3 +698,20 @@ def deliver_with_stalls(current: dict): "attempt": LOCATOR_LESS_ATTEMPT } assert written[-1]["last_delivery_notice_status"] == "sent_unverified" + + +def test_stall_notice_unknown_write_is_not_repeated(monkeypatch, delivery): + parts, _ = plan_manager_reply_parts(BODY) + state = _stalled_state(parts, stalls=3, sent=2) + writes = [] + def unknown(**kwargs): + writes.append(kwargs["text"]) + return {"ok": False, "reply_verified": False, "provider_write_attempted": True} + monkeypatch.setattr(parts_module, "reply_lark_event_inbox", unknown) + for _ in range(2): + result = parts_module.deliver_stall_notice(delivery_state=state, + delivery_path=delivery["tmp"] / "delivery.json", write_delivery=lambda *_: None, + reply_runner=object(), root=delivery["tmp"], config_path=delivery["tmp"] / "config.json", + message_id="om_fixture") + assert not result["reply_verified"] + assert len(writes) == 1 diff --git a/tests/extensions/test_lark_manager_returns.py b/tests/extensions/test_lark_manager_returns.py index 31b042a18c..e40ab2ee85 100644 --- a/tests/extensions/test_lark_manager_returns.py +++ b/tests/extensions/test_lark_manager_returns.py @@ -5,7 +5,7 @@ import test_lark_goal_topic_connections as fixtures from loopx.chat_store import ChatSessionStore -from loopx.extensions.lark.manager_returns import send_return +from loopx.extensions.lark.manager_returns import send_return, verify_return from loopx.extensions.lark.goal_topic_runtime import _inbox_config from loopx.extensions.lark.goal_topic_connections import decide_lark_topic_event from loopx.extensions.lark.goal_channel_targets import read_goal_channel_targets @@ -27,9 +27,10 @@ ) +@pytest.mark.parametrize("text", ["Concrete conclusion", "**Complete field**\n\n" + "测" * 24000], ids=["one-post", "split-posts"]) @pytest.mark.parametrize("revoke_before_send", [False, True]) def test_original_source_reply_waits_for_ack_and_rechecks_authority( - tmp_path, revoke_before_send + tmp_path, revoke_before_send, text ): kwargs, state, bindings = fixtures._manager_fixture(tmp_path) root = tmp_path / "runtime" @@ -98,6 +99,8 @@ def test_original_source_reply_waits_for_ack_and_rechecks_authority( ) calls = [] sent = [] + attempts = [] + messages = {} def transport(args): calls.append(args) @@ -116,7 +119,6 @@ def transport(args): assert args[args.index("--message-id") + 1] == "om_return_source" assert args[args.index("--msg-type") + 1] == "post" content = args[args.index("--content") + 1] - assert json.loads(content)["zh_cn"]["content"][0][0] == {"tag": "md", "text": "Concrete conclusion"} if "--dry-run" in args: if revoke_before_send: _write( @@ -126,19 +128,12 @@ def transport(args): data = {"api": [{"body": {"msg_type": "post", "content": content}}]} else: sent.append(args) - data = {"data": {"message_id": "om_return_result"}} + mid = f"om_return_result_{len(sent)}" + messages[mid] = {"message_id": mid, "msg_type": "post", "body": {"content": content}} + data = {"data": {"message_id": mid}} elif "+messages-mget" in args: - data = { - "data": { - "items": [ - { - "message_id": "om_return_result", - "msg_type": "post", - "content": "Concrete conclusion", - } - ] - } - } + mid = args[args.index("--message-ids") + 1] + data = {"data": {"items": [messages[mid]]}} else: raise AssertionError(args) return {"returncode": 0, "stdout": json.dumps(data)} @@ -151,8 +146,8 @@ def invoke(): route=return_route, session=session, turn=turn, - text="Concrete conclusion", - runner=transport, + text=text, + delivery_attempt_recorder=attempts.append, runner=transport, ) with pytest.raises(ValueError, match="initial reply") as pending: @@ -178,15 +173,20 @@ def invoke(): else: result = invoke() assert result["reply_verified"] and result["external_write_performed"], result - assert len(sent) == 1 and "--idempotency-key" in sent[0] + part_count = len(sent) + assert part_count >= 1 and "--idempotency-key" in sent[0] + if text != "Concrete conclusion": + assert part_count > 1 + parts = [json.loads(m["body"]["content"])["zh_cn"]["content"][0][0]["text"] for m in messages.values()] + assert "".join(part.split("\n\n", 1)[1] for part in parts) == text assert load_lark_event_inbox_config(project=root, config_path=config)["reply"][ "received_reaction_policy" ] == "retain" assert lark_inbox_reaction_receipts( inbox=inbox, message_id=event["message_id"] )["received"]["reaction_id"] == "reaction_Get" - assert invoke()["reply_verified"] + assert verify_return(root=root, registry=registry, snapshot_provider=lambda: snapshot, + route=return_route, session=session, turn=turn, text=text, attempt=attempts[-1], + runner=transport)["reply_verified"] + assert len(sent) == part_count assert not any("reactions" in call for call in calls) - assert sent[0][sent[0].index("--idempotency-key") + 1] == sent[1][ - sent[1].index("--idempotency-key") + 1 - ] diff --git a/tests/extensions/test_lark_markdown_reply.py b/tests/extensions/test_lark_markdown_reply.py index ec0fd47d06..e91647ec0b 100644 --- a/tests/extensions/test_lark_markdown_reply.py +++ b/tests/extensions/test_lark_markdown_reply.py @@ -20,6 +20,47 @@ TEXT = "进展\n\n- **结果**\n - 证据\n\n```python\nif ok:\n done()\n```" +def test_historical_raw_markdown_text_receipt_still_recovers_without_resending(tmp_path): + config, _, project = _fixture(tmp_path, lifecycle=False) + text = "| Case | Score |\n| --- | --- |\n| original | **0** |" + attempts = [] + first = reply_lark_event_inbox(project=project, config_path=config, + message_id="om_reaction_fixture", text=text, content_format="text", + execute=True, runner=ReplyRunner(readback_text="unavailable"), + delivery_attempt_recorder=attempts.append) + assert first["status"] == "sent_unverified" + recovery = ReplyRunner(readback_text=text) + verified = verify_lark_inbox_reply(project=project, config_path=config, + message_id="om_reaction_fixture", text=text, attempt=attempts[0], runner=recovery) + assert verified["reply_verified"] is True + assert not any("+messages-send" in call or "+messages-reply" in call for call in recovery.calls) + + +def test_rich_parts_recheck_native_source_authority_instead_of_using_legacy_membership(tmp_path): + config, _, project = _fixture(tmp_path, lifecycle=False) + text = "**完整内容**\n" * 2000 + provider = ReplyRunner() + checks = [] + + def source(profile, chat, message): + checks.append((profile, chat, message)) + return "Project Review Bot", len(checks) == 1 + + def runner(args): + if "--content" in args: + return {"returncode": 0, "stdout": json.dumps({"api": [{"body": { + "msg_type": "post", "content": args[args.index("--content") + 1], + }}]})} + return provider(args) + + result = reply_lark_event_inbox(project=project, config_path=config, + message_id="om_reaction_fixture", text=text, content_format="markdown", + execute=True, runner=runner, source_authority_verifier=source) + assert len(checks) == 2 and checks[0] == checks[1] + assert not result["ok"] + assert not any(("+messages-send" in a or "+messages-reply" in a) and "--dry-run" not in a for a in provider.calls) + + def test_result_attachment_zone_matches_raw_post_and_flattened_cli_readback(): keys = ("file_first", "file_second") content = json.loads(lark_markdown_post_content(TEXT)) @@ -135,84 +176,6 @@ def runner(args): assert ("--reply-in-thread" in sent[0]) == (placement == "source_thread") -def test_large_post_falls_back_before_send_without_truncating(tmp_path): - config, _, project = _fixture(tmp_path, lifecycle=False) - text = "- 完整内容\n" * 2000 - fallback = ReplyRunner(readback_text=text.strip()) - calls = [] - def runner(args): - calls.append(args) - if "--content" in args: - assert "--dry-run" in args # oversized post must never be sent - return {"returncode": 0, "stdout": json.dumps({"api": [{"body": { - "msg_type": "post", "content": args[args.index("--content") + 1], - }}]})} - return fallback(args) - result = reply_lark_event_inbox(project=project, config_path=config, - message_id="om_reaction_fixture", text=text, content_format="markdown", - execute=True, runner=runner) - assert result["reply_verified"] - assert result["content_format"] == "text" - assert result["format_fallback"] == "post_size_limit" - sent = [args for args in calls if "--text" in args and "--dry-run" not in args] - assert len(sent) == 1 - assert sent[0][sent[0].index("--text") + 1] == text.strip() - - -def test_large_post_fallback_can_be_verified_without_resending(tmp_path): - config, _, project = _fixture(tmp_path, lifecycle=False) - text = "- 完整内容\n" * 2000 - fallback = ReplyRunner(readback_text="different text") - attempts = [] - - def runner(args): - if "--content" in args: - return { - "returncode": 0, - "stdout": json.dumps( - { - "api": [ - { - "body": { - "msg_type": "post", - "content": args[args.index("--content") + 1], - } - } - ] - } - ), - } - return fallback(args) - - first = reply_lark_event_inbox( - project=project, - config_path=config, - message_id="om_reaction_fixture", - text=text, - content_format="markdown", - execute=True, - runner=runner, - delivery_attempt_recorder=attempts.append, - ) - assert first["status"] == "sent_unverified" - assert first["content_format"] == "text" - - recovery = ReplyRunner(readback_text=text.strip()) - verified = verify_lark_inbox_reply( - project=project, - config_path=config, - message_id="om_reaction_fixture", - text=text, - attempt=attempts[0], - runner=recovery, - ) - assert verified["reply_verified"] is True - assert not any( - "+messages-send" in call or "+messages-reply" in call - for call in recovery.calls - ) - - def test_markdown_request_with_mentions_keeps_verified_text_transport(tmp_path): config, _, project = _fixture(tmp_path, lifecycle=False) text = 'Reviewer please review' diff --git a/tests/extensions/test_lark_private_result_files.py b/tests/extensions/test_lark_private_result_files.py index dc6de68bdd..6b20f026a4 100644 --- a/tests/extensions/test_lark_private_result_files.py +++ b/tests/extensions/test_lark_private_result_files.py @@ -255,3 +255,47 @@ async def exercise(): drain(sender.root, sender.server.registry_path, transport.core.controller.store, sender) assert reply_status(sender.root, {"request_id": route["request_id"]})[0]["status"] == "delivered" assert len(replies) == 1 and uploads == [raw] + + +def test_split_file_result_recovers_all_posts_and_downloads_only_the_last(file_return): + from datetime import datetime, timedelta, timezone + + sender, session, turn, route, _, provider, transport, replies, workspace, raw, _, uploads, downloads, state = file_return + original = transport.runner + def unique_posts(args, cwd=None, timeout=None): + # Older one-message fixtures reuse an ID. Model distinct remote posts + # here without replacing the real source, Core return pump or file IO. + prior = provider.messages.get("om_out_worker") + result = original(args, cwd, timeout) + if ("+messages-reply" in args or "+messages-send" in args) and "--content" in args and "--dry-run" not in args: + ref = f"om_out_split_{len(replies)}" + provider.messages[ref] = {**provider.messages["om_out_worker"], "message_id": ref} + if prior is not None: + provider.messages["om_out_worker"] = prior + result = {**result, "stdout": json.dumps({"data": {"message_id": ref}})} + return result + transport.runner = unique_posts + body = "## Complete result\n\n" + "完整字段。" * 3000 + report(sender.root, route["goal_id"], route["agent_id"], route["request_id"], + "conclusion", body, registry=sender.server.registry_path, + attachment_refs=["result-fixture.bin"], workspace=workspace) + state["download_available"] = False + store = transport.core.controller.store + drain(sender.root, sender.server.registry_path, store, sender) + count = len(replies) + assert count > 1 and uploads == [raw], { + "status": [{k: item.get(k) for k in ("status", "error", "blocker")} for item in reply_status(sender.root, route)], + "journals": [{k: json.loads(p.read_text()).get(k) for k in ("delivery_parts_sent", "last_delivery_status", "delivery_part_started", "delivery_part_attempt")} for p in transport.runtime_root.glob("**/manager-delivery/parts/*.json")], + } + assert reply_status(sender.root, route)[0]["status"] == "verification_required" + messages = [provider.messages[f"om_out_split_{i}"] for i in range(1, count + 1)] + content = [json.loads(message["body"]["content"]) for message in messages] + assert all(message["msg_type"] == "post" for message in messages) + assert all("files" not in part for part in content[:-1]) + assert content[-1]["files"][0]["key"] == "file_result_fixture" + state["download_available"] = True + drain(sender.root, sender.server.registry_path, store, sender, + now=datetime.now(timezone.utc) + timedelta(minutes=10)) + assert reply_status(sender.root, route)[0]["status"] == "delivered" + assert len(replies) == count and uploads == [raw] and len(downloads) == 2 + assert all(call[call.index("--message-id") + 1] == f"om_out_split_{count}" for call in downloads) diff --git a/tests/extensions/test_lark_rich_parts.py b/tests/extensions/test_lark_rich_parts.py new file mode 100644 index 0000000000..534feb72ca --- /dev/null +++ b/tests/extensions/test_lark_rich_parts.py @@ -0,0 +1,361 @@ +"""Complete rich replies use provider bytes and survive partial delivery.""" + +import json +from pathlib import Path + +import pytest + +from loopx.extensions.lark.inbox_reply import ( + reply_lark_event_inbox, + verify_lark_inbox_reply, +) +from loopx.extensions.lark.outbound import ( + LARK_POST_REQUEST_MAX_BYTES, + LarkOutboundTextError, +) +from loopx.extensions.lark.presentation.markdown_parts import ( + split_lark_markdown_posts, + post_request_bytes, +) +from loopx.extensions.lark.post_delivery import attachment_reply +from test_lark_inbox_reactions import ReplyRunner, _fixture + + +class RichProvider(ReplyRunner): + def __init__(self): + super().__init__() + self.messages = {} + self.sent = [] + self.unreadable = set() + self.no_locator = False + self.fail_after_write = False + self.ascii_wire = False + + def __call__(self, args): + if "+messages-send" in args or "+messages-reply" in args: + self.calls.append(list(args)) + assert "--text" not in args + content = json.loads(args[args.index("--content") + 1]) + keys = [args[i + 1] for i, arg in enumerate(args) if arg == "--attachment"] + if keys: + content["files"] = [{"key": key} for key in keys] + body = { + "msg_type": "post", + "content": json.dumps( + content, ensure_ascii=self.ascii_wire, separators=(",", ":") + ), + "uuid": args[args.index("--idempotency-key") + 1], + "receive_id": "oc_project_review", + } + if "--dry-run" in args: + return { + "returncode": 0, + "stdout": json.dumps({"api": [{"body": body}]}), + } + assert ( + len( + json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode() + ) + <= LARK_POST_REQUEST_MAX_BYTES + ) + mid = f"om_part_{len(self.sent)}" + self.sent.append(content) + self.messages[mid] = { + "message_id": mid, + "msg_type": "post", + "body": {"content": content}, + } + return { + "returncode": int(self.fail_after_write), + "stdout": json.dumps({} if self.no_locator else {"message_id": mid}), + } + if "+messages-mget" in args: + self.calls.append(list(args)) + mid = args[args.index("--message-ids") + 1] + return { + "returncode": int(mid in self.unreadable), + "stdout": json.dumps({"items": [self.messages[mid]]}), + } + return super().__call__(args) + + +def send(tmp_path, provider, text, **options): + config, _, project = _fixture(tmp_path, lifecycle=False) + args = dict( + project=project, + config_path=config, + message_id="om_reaction_fixture", + text=text, + content_format="markdown", + execute=True, + runner=provider, + ) + return args, reply_lark_event_inbox(**args, **options) + + +def resume(args, attempt, **options): + return verify_lark_inbox_reply( + **{ + key: val + for key, val in args.items() + if key not in {"content_format", "execute"} + }, + attempt=attempt, + **options, + ) + + +@pytest.mark.parametrize( + "body", + ["测😀" * 9000, '"\\' * 12000, "x" * 310000], + ids=["unicode", "escaping", "over-eight"], +) +def test_byte_packing_keeps_every_character_without_a_part_cap(body): + template = {"msg_type": "post", "uuid": "a" * 50, "receive_id": "oc_fixture"} + parts = split_lark_markdown_posts(body, template) + assert "".join(part.split("\n\n", 1)[1] for part in parts) == body + assert all(post_request_bytes(part, template) <= 30000 for part in parts) + assert post_request_bytes(parts[0], template) >= 29990 + if len(body) > 300000: + assert len(parts) > 8 + + +def test_split_keeps_code_fences_tables_links_and_complete_data(): + code = "```python\n" + " print('测')\n" * 5000 + "```" + parts = split_lark_markdown_posts(code, {"msg_type": "post"}) + assert len(parts) > 1 + assert all(part.count("```") == 2 for part in parts) + assert sum(part.count("print('测')") for part in parts) == 5000 + table = "| Case | Score |\n| --- | --- |\n" + "".join( + f"| item-{i} | **0 / unknown** |\n" for i in range(4000) + ) + parts = split_lark_markdown_posts(table, {"msg_type": "post"}) + assert all("| Case | Score |\n| --- | --- |" in part for part in parts) + assert sum(part.count("| item-") for part in parts) == 4000 + link = "[source](https://example.org/" + "x" * 5000 + ")" + parts = split_lark_markdown_posts( + "a" * 29000 + link + "b" * 2000, {"msg_type": "post"} + ) + assert sum(link in part for part in parts) == 1 + + +@pytest.mark.parametrize("element", [ + "[source](https://example.org/" + "x" * 5000 + ")", + "`" + "x" * 5000 + "`", +], ids=["link", "inline-code"]) +def test_real_entry_moves_protected_element_to_a_fresh_post(tmp_path, element): + text = "P" * 28000 + "\n\n" + element + "z" * 35000 + provider = RichProvider() + _, result = send(tmp_path, provider, text) + assert result["ok"] and result["reply_verified"] + bodies = [m["zh_cn"]["content"][0][0]["text"] for m in provider.sent] + # The paragraph break becomes a message boundary; no authored data may go. + assert bodies[0].split("\n\n", 1)[1] == "P" * 28000 + assert "".join(body.split("\n\n", 1)[1] for body in bodies[1:]) == element + "z" * 35000 + assert sum(element in body for body in bodies) == 1 + + +@pytest.mark.parametrize("block", [ + "[source](https://example.org/" + "x" * 31000 + ")", + "| Case | Value |\n| --- | --- |\n| row | " + "x" * 31000 + " |\n", +], ids=["link", "table-row"]) +def test_unrepresentable_atomic_block_stays_pending_without_any_send(tmp_path, block): + provider = RichProvider() + with pytest.raises(LarkOutboundTextError, match="exceeds the provider"): + send(tmp_path, provider, "P" * 28000 + "\n\n" + block) + assert provider.sent == [] + + +@pytest.mark.parametrize("escaped_unicode", [False, True]) +def test_real_entry_splits_rich_posts_and_recovers_only_unsent_parts( + tmp_path, escaped_unicode +): + provider = RichProvider() + provider.ascii_wire = escaped_unicode + provider.unreadable.add("om_part_0") + text = "## Findings\n\n" + "完整字段与来源。" * 6000 + attempts = [] + args, first = send( + tmp_path, provider, text, delivery_attempt_recorder=attempts.append + ) + assert first["reply_verified"] is False and len(provider.sent) == 1 + assert attempts and attempts[0]["message_ref"].startswith("lark-parts:") + assert len({json.dumps(item, sort_keys=True) for item in attempts}) == 1 + # Verify-only recovery cannot send a tail, including after a restart. + assert resume(args, attempts[0])["reply_verified"] is False + assert len(provider.sent) == 1 + provider.unreadable.clear() + complete = resume(args, attempts[0], resume_parts=True) + assert complete["reply_verified"] and complete["ok"] + assert complete["content_format"] == "markdown" + sent = len(provider.sent) + assert sent > 1 + assert resume(args, attempts[0], resume_parts=True)["reply_verified"] + assert len(provider.sent) == sent + assert all("--text" not in call for call in provider.calls) + content = "".join( + m["zh_cn"]["content"][0][0]["text"].split("\n\n", 1)[1] for m in provider.sent + ) + assert content == text + + +@pytest.mark.parametrize("loss", ["locator", "response"]) +def test_ambiguous_part_never_resends_or_reports_completion(tmp_path, loss): + provider = RichProvider() + provider.no_locator = loss == "locator" + provider.fail_after_write = loss == "response" + attempts = [] + args, result = send( + tmp_path, provider, "测" * 20000, delivery_attempt_recorder=attempts.append + ) + assert not result["reply_verified"] and len(provider.sent) == 1 and attempts + assert not resume(args, attempts[0], resume_parts=True)["reply_verified"] + assert len(provider.sent) == 1 + + +def test_each_part_rechecks_source_and_keeps_the_original_review_hook(tmp_path): + provider = RichProvider() + observations = [] + reviews = [] + + def authority(profile, chat, message): + observations.append((profile, chat, message)) + return "Project Review Bot", len(observations) < 3 + + _, result = send( + tmp_path, + provider, + "测" * 24000, + source_authority_verifier=authority, + before_send=lambda intent: ( + reviews.append(intent) or {"continue_delivery": True} + ), + ) + assert not result["reply_verified"] and len(provider.sent) == 1 + assert len(set(observations)) == 1 and len(reviews) == 1 + + +def test_changed_intent_or_part_plan_is_rejected_without_a_write(tmp_path): + provider = RichProvider() + provider.unreadable.add("om_part_0") + attempts = [] + args, _ = send( + tmp_path, provider, "测" * 20000, delivery_attempt_recorder=attempts.append + ) + changed = {**args, "text": args["text"] + "changed"} + assert ( + resume(changed, attempts[0], resume_parts=True)["blocker"] + == "provider_delivery_intent_conflict" + ) + journal = next( + Path(args["project"]).glob( + ".loopx/inbox/feedback/manager-delivery/parts/*.json" + ) + ) + state = json.loads(journal.read_text()) + state["delivery_parts"][1] = "other content" + journal.write_text(json.dumps(state)) + assert ( + resume(args, attempts[0], resume_parts=True)["blocker"] + == "provider_delivery_intent_conflict" + ) + assert len(provider.sent) == 1 + + +def test_attachments_stay_on_last_concrete_post(tmp_path): + provider = RichProvider() + attempts = [] + args, result = send( + tmp_path, + provider, + "测" * 20000, + delivery_attempt_recorder=attempts.append, + attachment_keys=("file_fixture",), + ) + assert result["reply_verified"] + assert all("files" not in content for content in provider.sent[:-1]) + assert provider.sent[-1]["files"] == [{"key": "file_fixture"}] + part, last = attachment_reply( + project=args["project"], + config_path=args["config_path"], + attempt=attempts[-1], + text=args["text"], + ) + assert last["message_ref"] == f"om_part_{len(provider.sent) - 1}" + assert part.endswith("测") + + +def test_preview_and_denied_review_do_not_write_or_record_an_attempt(tmp_path): + provider = RichProvider() + config, _, project = _fixture(tmp_path, lifecycle=False) + args = dict( + project=project, + config_path=config, + message_id="om_reaction_fixture", + text="测" * 20000, + content_format="markdown", + runner=provider, + ) + preview = reply_lark_event_inbox(**args, provider_preflight=True) + assert ( + preview["content_format"] == "markdown" and preview["delivery_part_count"] > 1 + ) + assert not provider.sent + assert not list(Path(project).glob("**/manager-delivery/parts/*.json")) + attempts, reviewed = [], [] + + class Review: + def __call__(self, intent): + pytest.fail("review must bind the actual destination") + + def for_destination(self, chat): + reviewed.append(chat) + return lambda intent: {"continue_delivery": False} + + result = reply_lark_event_inbox( + **args, + execute=True, + before_send=Review(), + delivery_attempt_recorder=attempts.append, + ) + assert not result["reply_verified"] and not provider.sent and not attempts + assert reviewed == ["oc_project_review"] + + +@pytest.mark.parametrize("fault", ["source", "destination", "missing-journal"]) +def test_recovery_fences_unavailable_source_destination_or_journal(tmp_path, fault): + provider = RichProvider() + provider.unreadable.add("om_part_0") + attempts = [] + args, _ = send( + tmp_path, provider, "测" * 20000, delivery_attempt_recorder=attempts.append + ) + provider.unreadable.clear() + options = {} + if fault == "source": + options["source_authority_verifier"] = lambda *_: ("Project Review Bot", False) + elif fault == "destination": + config = json.loads(Path(args["config_path"]).read_text()) + config["reply"]["chat_id"] = "oc_other" + Path(args["config_path"]).write_text(json.dumps(config)) + else: + next(Path(args["project"]).glob("**/manager-delivery/parts/*.json")).unlink() + assert not resume(args, attempts[0], resume_parts=True, **options)["reply_verified"] + assert len(provider.sent) == 1 + + +def test_unexpected_preview_encoding_fails_without_recursing_or_text_fallback(tmp_path): + provider = RichProvider() + + def html_escaped_preview(args): + result = provider(args) + if "--dry-run" in args: + value = json.loads(result["stdout"]) + body = value["api"][0]["body"] + body["content"] = body["content"].replace("<", "\\u003c") + return {**result, "stdout": json.dumps(value)} + return result + + with pytest.raises(LarkOutboundTextError, match="serialization"): + send(tmp_path, html_escaped_preview, "<" * 6000) + assert provider.sent == [] diff --git a/tests/test_lark_private_progress.py b/tests/test_lark_private_progress.py index c7c9c4a618..c8276a57fb 100644 --- a/tests/test_lark_private_progress.py +++ b/tests/test_lark_private_progress.py @@ -668,3 +668,43 @@ def test_ambiguous_progress_edit_counts_against_budget_before_provider_write(ord assert len(provider.edits) == 1 finally: runtime.close() + + +@pytest.mark.parametrize("rollover", ["terminal", "progress"]) +def test_split_draft_retires_only_after_readback_and_keeps_the_full_next_reply(ordinary, rollover): # noqa: F811 + store, runtime, provider, transport = streaming(ordinary) + try: + event = provider.event("notes-app", "split-draft", "wait for interrupt") + transport.admit("notes-app", event) + row = transport.core.pending()[0] + active(store, row) + path = transport.root / f"{row['request_ref']}.json" + record = _read_json(path) + def inbox(): + return transport._inbox(record) + draft = "## Draft\n\n" + "完整草稿字段。" * 4000 + assert transport._deliver(path, record, "progress", draft, inbox=inbox) + count = len(provider.writes) + assert count > 1 + assert record["deliveries"]["progress"]["attempt"]["message_ref"].startswith("lark-parts:") + provider.verify_replies = False + if rollover == "terminal": + assert not transport._deliver(path, record, "terminal", draft + "Final.", inbox=inbox) + else: + assert not transport._continue_progress(path, record, inbox=inbox) + assert len(provider.writes) == count + provider.verify_replies = True + if rollover == "progress": + assert transport._continue_progress(path, record, inbox=inbox) + assert record["deliveries"]["progress.0"]["verified"] + else: + assert transport._deliver(path, record, "terminal", draft + "Final.", inbox=inbox) + assert record["stream"]["closed"] and record["deliveries"]["terminal"]["verified"] + sent = len(provider.writes) + assert transport._deliver(path, record, "terminal", draft + "Final.", inbox=inbox) + assert len(provider.writes) == sent + final_parts = [text.split("\n\n", 1)[1] for _, text in provider.writes[count:]] + assert "".join(final_parts) == draft + "Final." + assert all(message["msg_type"] == "post" for ref, message in provider.messages.items() if ref.startswith("om_out_")) + finally: + runtime.close()