fix(lark): split oversized Markdown into complete rich replies

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

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 {}

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.

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