fix(feishu): route tables and code blocks to CardKit 2.0 with post/text fallback

gateway/platforms/feishu.py

@@ -160,6 +160,11 @@
 _MARKDOWN_FENCE_OPEN_RE = re.compile(r"^```([^\n`]*)\s*$")
 _MARKDOWN_FENCE_CLOSE_RE = re.compile(r"^```\s*$")
 _MENTION_RE = re.compile(r"@_user_\d+")
+# GFM table: a row of |cells| followed by a separator row (---|:---|...)
+_TABLE_MARKDOWN_RE = re.compile(r"^\|.+\|.*\n\|[-: |]+\|", re.MULTILINE)
+# Multi-line fenced code block (opening fence + 2+ content lines + closing fence)
+# Empty or 1-line blocks render fine in post/md; route 2+ content lines to CardKit 2.0.
+_CODE_BLOCK_RE = re.compile(r"^```[^\n]*\n(.*\n){2,}```", re.MULTILINE)
 _MULTISPACE_RE = re.compile(r"[ \t]{2,}")
 _POST_CONTENT_INVALID_RE = re.compile(r"content format of the post type is incorrect", re.IGNORECASE)
 # ---------------------------------------------------------------------------
@@ -1724,9 +1729,28 @@ async def send(
                         metadata=metadata,
                     )
                 except Exception as exc:
-                    if msg_type != "post" or not _POST_CONTENT_INVALID_RE.search(str(exc)):
+                    if msg_type == "interactive":
+                        logger.warning("[Feishu] Interactive card send failed; falling back to plain text")
+                        response = await self._feishu_send_with_retry(
+                            chat_id=chat_id,
+                            msg_type="text",
+                            payload=json.dumps({"text": _strip_markdown_to_plain_text(chunk)}, ensure_ascii=False),
+                            reply_to=reply_to,
+                            metadata=metadata,
+                        )
+                    elif msg_type != "post" or not _POST_CONTENT_INVALID_RE.search(str(exc)):
                         raise
-                    logger.warning("[Feishu] Invalid post payload rejected by API; falling back to plain text")
+                    else:
+                        logger.warning("[Feishu] Invalid post payload rejected by API; falling back to plain text")
+                        response = await self._feishu_send_with_retry(
+                            chat_id=chat_id,
+                            msg_type="text",
+                            payload=json.dumps({"text": _strip_markdown_to_plain_text(chunk)}, ensure_ascii=False),
+                            reply_to=reply_to,
+                            metadata=metadata,
+                        )
+                if msg_type == "interactive" and not self._response_succeeded(response):
+                    logger.warning("[Feishu] Interactive card rejected by API; falling back to plain text")
                     response = await self._feishu_send_with_retry(
                         chat_id=chat_id,
                         msg_type="text",
@@ -1773,6 +1797,15 @@ async def edit_message(
             request = self._build_update_message_request(message_id=message_id, request_body=body)
             response = await asyncio.to_thread(self._client.im.v1.message.update, request)
             result = self._finalize_send_result(response, "update failed")
+            if not result.success and msg_type == "interactive":
+                logger.warning("[Feishu] Interactive card update rejected by API; falling back to plain text")
+                fallback_body = self._build_update_message_body(
+                    msg_type="text",
+                    content=json.dumps({"text": _strip_markdown_to_plain_text(content)}, ensure_ascii=False),
+                )
+                fallback_request = self._build_update_message_request(message_id=message_id, request_body=fallback_body)
+                fallback_response = await asyncio.to_thread(self._client.im.v1.message.update, fallback_request)
+                result = self._finalize_send_result(fallback_response, "update failed")
             if not result.success and msg_type == "post" and _POST_CONTENT_INVALID_RE.search(result.error or ""):
                 logger.warning("[Feishu] Invalid post update payload rejected by API; falling back to plain text")
                 fallback_body = self._build_update_message_body(
@@ -4175,6 +4208,11 @@ def _is_duplicate(self, message_id: str) -> bool:
     # =========================================================================
 
     def _build_outbound_payload(self, content: str) -> tuple[str, str]:
+        # Route tables and multi-line code blocks to CardKit 2.0 interactive format.
+        # Feishu's post/md tag does not support table syntax and truncates code
+        # blocks after ~6 lines; CardKit 2.0's markdown element handles both.
+        if _TABLE_MARKDOWN_RE.search(content) or _CODE_BLOCK_RE.search(content):
+            return "interactive", self._build_card_payload(content)
         # Feishu post-type 'md' elements do not render markdown tables; sending
         # table content as post causes the message to appear blank on the client.
         # Force plain text for anything that looks like a markdown table.
@@ -4466,7 +4504,9 @@ async def _feishu_send_with_retry(
                 return response
             except Exception as exc:
                 last_error = exc
-                if msg_type == "post" and _POST_CONTENT_INVALID_RE.search(str(exc)):
+                if msg_type in ("post", "interactive") and _POST_CONTENT_INVALID_RE.search(str(exc)):
+                    raise
+                if msg_type == "interactive":
                     raise
                 if attempt >= _FEISHU_SEND_ATTEMPTS - 1:
                     raise
@@ -4655,6 +4695,22 @@ def _build_media_post_payload(self, *, caption: str, media_tag: Dict[str, str])
         content.append([media_tag])
         return json.dumps(payload, ensure_ascii=False)
 
+    def _build_card_payload(self, content: str) -> str:
+        """Build a CardKit 2.0 interactive card with a markdown element.
+
+        CardKit 2.0's ``tag: markdown`` element natively supports GFM tables and
+        fenced code blocks with scrollable rendering — unlike the ``post/md`` tag
+        which silently drops tables and truncates code after ~6 lines.
+        """
+        card = {
+            "schema": "2.0",
+            "config": {"wide_screen_mode": True},
+            "body": {
+                "elements": [{"tag": "markdown", "content": content}]
+            },
+        }
+        return json.dumps(card, ensure_ascii=False)
+
     @staticmethod
     def _resolve_outbound_file_routing(
         *,

tests/gateway/test_feishu.py

@@ -4823,3 +4823,117 @@ def test_scenario_post_bot_plus_alice_filters_self_from_hint(self):
         # Body: leading @Hermes stripped, Alice preserved, trailing text intact.
         self.assertIn("@Alice review the spec with Alice", event.text)
         self.assertNotIn("@Hermes @Alice", event.text)
+
+
+# ---------------------------------------------------------------------------
+# Tests for CardKit 2.0 routing — tables and multi-line code blocks
+# ---------------------------------------------------------------------------
+
+
+class TestOutboundCardkitRouting(unittest.TestCase):
+    """_build_outbound_payload routes tables / code blocks to interactive."""
+
+    def setUp(self):
+        from gateway.config import PlatformConfig
+        from gateway.platforms.feishu import FeishuAdapter
+
+        self.adapter = FeishuAdapter(PlatformConfig())
+
+    # -- Table detection --------------------------------------------------
+
+    def test_gfm_table_routes_to_interactive(self):
+        msg_type, payload = self.adapter._build_outbound_payload(
+            "| 名称 | 价格 |\n|------|------|\n| AAPL | 150  |"
+        )
+        self.assertEqual(msg_type, "interactive")
+        card = json.loads(payload)
+        self.assertEqual(card["schema"], "2.0")
+        self.assertIn("tag", card["body"]["elements"][0])
+        self.assertEqual(card["body"]["elements"][0]["tag"], "markdown")
+
+    def test_plain_text_no_table_no_codeblock_routes_to_text(self):
+        msg_type, payload = self.adapter._build_outbound_payload(
+            "Hello, this is a normal message without any tables or code blocks."
+        )
+        self.assertEqual(msg_type, "text")
+
+    def test_pipe_in_inline_text_does_not_trigger_table(self):
+        """Pipes in prose should not match the table regex."""
+        msg_type, _ = self.adapter._build_outbound_payload(
+            "This is a message with a | pipe character in it, but no real table."
+        )
+        self.assertNotEqual(msg_type, "interactive")
+
+    # -- Code block detection ---------------------------------------------
+
+    def test_multi_line_code_block_routes_to_interactive(self):
+        msg_type, payload = self.adapter._build_outbound_payload(
+            "```python\nimport os\nprint('hello')\n```"
+        )
+        self.assertEqual(msg_type, "interactive")
+        card = json.loads(payload)
+        self.assertIn("import os", card["body"]["elements"][0]["content"])
+
+    def test_short_code_block_stays_post_or_text(self):
+        """Code blocks with only 1 content line stay in post or text."""
+        msg_type, _ = self.adapter._build_outbound_payload(
+            "```python\nprint('hi')\n```"
+        )
+        self.assertIn(msg_type, ("post", "text"))
+
+    def test_inline_code_does_not_trigger_code_block(self):
+        msg_type, _ = self.adapter._build_outbound_payload(
+            "Use the `print()` function to output text."
+        )
+        self.assertNotEqual(msg_type, "interactive")
+
+    # -- _build_card_payload output --------------------------------------
+
+    def test_build_card_payload_produces_valid_cardkit2_json(self):
+        payload = self.adapter._build_card_payload("**bold** and _italic_")
+        card = json.loads(payload)
+        self.assertEqual(card["schema"], "2.0")
+        self.assertTrue(card["config"]["wide_screen_mode"])
+        elements = card["body"]["elements"]
+        self.assertEqual(len(elements), 1)
+        self.assertEqual(elements[0]["tag"], "markdown")
+        self.assertEqual(elements[0]["content"], "**bold** and _italic_")
+
+    def test_build_card_payload_preserves_table_syntax(self):
+        table_md = "| A | B |\n|---|---|\n| 1 | 2 |"
+        payload = self.adapter._build_card_payload(table_md)
+        card = json.loads(payload)
+        self.assertIn("| A | B |", card["body"]["elements"][0]["content"])
+
+    # -- Regex correctness -----------------------------------------------
+
+    def test_table_regex_matches_gfm_separator_variants(self):
+        import re
+        from gateway.platforms.feishu import _TABLE_MARKDOWN_RE
+
+        # Left, center, right alignment
+        self.assertTrue(_TABLE_MARKDOWN_RE.search("|col|\n|---|"))
+        self.assertTrue(_TABLE_MARKDOWN_RE.search("|col|\n|:---|"))
+        self.assertTrue(_TABLE_MARKDOWN_RE.search("|col|\n|:---:|"))
+        self.assertTrue(_TABLE_MARKDOWN_RE.search("|col|\n|---:|"))
+
+    def test_code_block_regex_matches_only_four_plus_total_lines(self):
+        import re
+        from gateway.platforms.feishu import _CODE_BLOCK_RE
+
+        # 4 lines total (open + 2 content + close) → matches `(.*\n){2,}`
+        self.assertTrue(_CODE_BLOCK_RE.search("```\na\nb\n```"))
+        # 6 lines total
+        self.assertTrue(_CODE_BLOCK_RE.search("```\na\nb\nc\nd\n```"))
+        # 3 lines total (open + 1 content + close) → doesn't match
+        self.assertIsNone(_CODE_BLOCK_RE.search("```\na\n```"))
+        # 2 lines total (empty block) → doesn't match
+        self.assertIsNone(_CODE_BLOCK_RE.search("```\n```"))
+        # Inline backticks → doesn't match
+        self.assertIsNone(_CODE_BLOCK_RE.search("`code`"))
+
+    def test_table_then_interactive_fallback_on_code_block(self):
+        """Combined content: table + code block → interactive."""
+        content = "| X | Y |\n|---|---|\n| 1 | 2 |\n\n```\nline1\nline2\n```"
+        msg_type, payload = self.adapter._build_outbound_payload(content)
+        self.assertEqual(msg_type, "interactive")