fix(wechatclawbot): 识别 item_list 媒体入站消息并修正 CDN 探测判定 (#7000)

* fix(wechatclawbot): 识别 item_list 媒体入站消息并修正 CDN 探测判定 (#6992)

iLink 下行媒体报文把附件放在 item_list 子项中，原解析存在三处缺口，
图片/语音/文件入站消息会被判为空消息丢弃：

- item_list 仅按数字 type 处理 1 文本/2 图片/5、6 文件，缺少 3 语音/
  4 文件/5 视频分派；改为按实际出现的子项键名分派，同时兼容历史
  type=5/6 文件报文。
- 附件引用只读取顶层 url/aeskey，识别不了 image_item.media.full_url /
  media.aes_key 等嵌套字段；统一按“顶层优先、media 子对象兜底”提取。
- 消息被解析为空时直接返回 None，日志只剩“未识别到有效消息”；补充
  渠道级诊断日志，输出 message_id 与截断后的原始报文。

另外把“微信 ClawBot 媒体 CDN”网络探测的 400 纳入成功状态码：该 CDN
对无参数裸请求固定返回 400，此前探测项恒定误报失败。

tests/test_wechatclawbot.py 整文件迁移为 pytest 原生，并新增 item_list
媒体解析、仅图片消息可入链、空消息诊断日志回归；nettest 契约测试补充
CDN 400 探测用例。复杂度与 mypy 低水位随债务下降同步固化。

* fix(wechatclawbot): 同步架构清单中的 mypy 债务数字 (#6992)

app/application/nettest/catalogue.py

@@ -308,7 +308,9 @@ def _build_notification_module_rules() -> list[NetworkTestRule]:
             proxy=True,
             allowed_redirect_prefixes=("https://novac2c.cdn.weixin.qq.com/",),
             module_ids=("WechatClawBotModule",),
-            success_status_codes=(200, 403, 404),
+            # CDN 对不带下载参数的裸请求固定返回 400，能返回 400 即证明连通正常，
+            # 不能据此判定探测失败。
+            success_status_codes=(200, 400, 403, 404),
         ),
     ]
 

app/modules/wechatclawbot/__init__.py

@@ -292,6 +292,15 @@ class WechatClawBotModule(_MessageChannelModuleBase[WechatClawBot]):
         audio_refs = self._normalize_audio_refs(message.get("audio_refs"))
         files = self._normalize_files(message.get("files"))
         if not text and not images and not audio_refs and not files:
+            # 渠道报文被解析为空消息时必须留下诊断信息，否则只能在消息入口看到
+            # “未识别到有效消息”，无法定位是哪种入站内容被丢弃。
+            logger.info(
+                "微信 ClawBot 消息未识别到有效内容：source=%s, userid=%s, message_id=%s, body=%s",
+                client_config.name,
+                user_id,
+                message_id,
+                WechatClawBot._short_text(message, max_len=500),
+            )
             return None
         if self._is_duplicate_message(client_config.name, message_id):
             logger.info(

app/modules/wechatclawbot/wechatclawbot.py

@@ -10,7 +10,7 @@ import time
 import uuid
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional, Tuple
 from urllib.parse import quote
 
 from Crypto.Cipher import AES
@@ -975,30 +975,36 @@ class ILinkClient:
         attachment: Dict[str, Any],
         default_name: Optional[str] = None,
     ) -> Optional[Tuple[str, Dict[str, Any]]]:
-        """从附件信息构建 wxclaw 协议引用。"""
+        """
+        从附件信息构建 wxclaw 协议引用。
+
+        下行报文可能把下载地址与密钥放在附件的 `media` 子对象中
+        （如 `image_item.media.full_url` / `media.aes_key`），也可能直接放在附件顶层，
+        这里统一按“附件顶层优先、media 子对象兜底”的顺序提取字段。
+        """
         if not isinstance(attachment, dict):
             return None
-        download_url = (
-            attachment.get("download_url")
-            or attachment.get("url")
-            or attachment.get("cdnurl")
-            or attachment.get("cdn_url")
-            or attachment.get("file_url")
+        media = attachment.get("media")
+        media = media if isinstance(media, dict) else {}
+        url_keys = ["full_url", "download_url", "url", "cdnurl", "cdn_url", "file_url"]
+        download_url = self._pick_value(attachment, url_keys) or self._pick_value(
+            media, url_keys
         )
         if not download_url:
             return None
+        aeskey_keys = ["aeskey", "aes_key", "encoding_aes_key", "encrypt_key"]
         payload = {
             "url": download_url,
-            "aeskey": attachment.get("aeskey")
-            or attachment.get("encoding_aes_key")
-            or attachment.get("encrypt_key"),
-            "encrypt_type": attachment.get("encrypt_type"),
-            "mime_type": attachment.get("mime_type")
-            or attachment.get("content_type"),
-            "name": attachment.get("name")
-            or attachment.get("filename")
+            "aeskey": self._pick_value(attachment, aeskey_keys)
+            or self._pick_value(media, aeskey_keys),
+            "encrypt_type": self._pick_value(attachment, ["encrypt_type"])
+            or self._pick_value(media, ["encrypt_type"]),
+            "mime_type": self._pick_value(attachment, ["mime_type", "content_type"])
+            or self._pick_value(media, ["mime_type", "content_type"]),
+            "name": self._pick_value(attachment, ["name", "filename", "file_name"])
             or default_name,
-            "size": attachment.get("size") or attachment.get("file_size"),
+            "size": self._pick_value(attachment, ["size", "file_size"])
+            or self._pick_value(media, ["size", "file_size"]),
         }
         ref = self._encode_ref_payload(kind=kind, payload=payload)
         return ref, payload
@@ -1012,6 +1018,58 @@ class ILinkClient:
             return None
         return value
 
+    def _append_item_list_entry(
+        self,
+        one: Dict[str, Any],
+        append_text: Callable[[Any], None],
+        images: List[Dict[str, Any]],
+        audio_refs: List[str],
+        files: List[Dict[str, Any]],
+    ) -> None:
+        """
+        把 item_list 的单个子项分派为文本或媒体附件。
+
+        iLink 下行报文的 item_list 存在两套类型约定：数字 `type` 与子项键名。
+        这里按实际出现的子项键名分派，兼容公开资料的 `2 图片/3 语音/4 文件/5 视频`
+        以及历史实现里 `file_item` 使用 `type=5/6` 的报文，避免媒体消息被静默丢弃。
+        """
+        candidates = (
+            ("text_item", "text", None),
+            ("image_item", "image", None),
+            ("voice_item", "voice", "voice.amr"),
+            ("file_item", "file", None),
+            ("video_item", "file", "video.mp4"),
+        )
+        for key, kind, default_name in candidates:
+            child = one.get(key)
+            if not isinstance(child, dict):
+                continue
+            if kind == "text":
+                append_text(child.get("text"))
+                return
+            if kind == "voice":
+                append_text(child.get("content"))
+            attachment_ref = self._build_attachment_ref(
+                kind, child, default_name=default_name
+            )
+            if not attachment_ref:
+                return
+            ref, payload = attachment_ref
+            if kind == "voice":
+                audio_refs.append(ref)
+                return
+            entry = {
+                "ref": ref,
+                "name": payload.get("name"),
+                "mime_type": payload.get("mime_type"),
+                "size": payload.get("size"),
+            }
+            if kind == "image":
+                images.append(entry)
+            else:
+                files.append(entry)
+            return
+
     def _build_poll_result(
         self,
         success: bool,
@@ -1228,41 +1286,7 @@ class ILinkClient:
         for one in item_list:
             if not isinstance(one, dict):
                 continue
-            item_type = one.get("type")
-            if item_type == 1 and isinstance(one.get("text_item"), dict):
-                append_text((one.get("text_item") or {}).get("text"))
-            elif item_type == 2 and isinstance(one.get("image_item"), dict):
-                image_ref = self._build_attachment_ref(
-                    "image", (one.get("image_item") or {}).get("media") or {}
-                )
-                if image_ref:
-                    ref, payload = image_ref
-                    images.append(
-                        {
-                            "ref": ref,
-                            "name": payload.get("name"),
-                            "mime_type": payload.get("mime_type"),
-                            "size": payload.get("size"),
-                        }
-                    )
-            elif item_type in {5, 6}:
-                file_ref = self._build_attachment_ref(
-                    "file", (one.get("file_item") or {}).get("media") or {}
-                )
-                if file_ref:
-                    ref, payload = file_ref
-                    files.append(
-                        {
-                            "ref": ref,
-                            "name": (one.get("file_item") or {}).get("name")
-                            or (one.get("file_item") or {}).get("file_name")
-                            or payload.get("name"),
-                            "mime_type": (one.get("file_item") or {}).get("mime_type")
-                            or payload.get("mime_type"),
-                            "size": (one.get("file_item") or {}).get("size")
-                            or payload.get("size"),
-                        }
-                    )
+            self._append_item_list_entry(one, append_text, images, audio_refs, files)
 
         if not text_parts:
             append_text(

docs/refactor/optimization-checklist.md

@@ -108,7 +108,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
 | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 |
 | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 |
-| 全量 mypy 历史债务 | 9,163 / 508 文件 | 通知排版补齐 Slack 消息块类型，历史债务减少 2 项；工作流资源过滤改用领域对象调用过滤链，恢复的节点输出在消费前归一化，并收紧可空资源列表的类型低水位；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
+| 全量 mypy 历史债务 | 9,159 / 508 文件 | 通知排版补齐 Slack 消息块类型，历史债务减少 2 项；工作流资源过滤改用领域对象调用过滤链，恢复的节点输出在消费前归一化，并收紧可空资源列表的类型低水位；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降；微信 ClawBot 入站 item_list 媒体解析改为按子项键名分派后，类型错误减少 4 项 |
 | Ruff 历史诊断 | 100 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
 | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 

tests/fixtures/architecture/complexity-baseline.json

@@ -110,7 +110,7 @@
       "app/modules/trimemedia/api.py:Api.request": 16,
       "app/modules/ugreen/ugreen.py:Ugreen.get_tv_episodes": 21,
       "app/modules/wechat/__init__.py:WechatModule.message_parser": 20,
-      "app/modules/wechatclawbot/wechatclawbot.py:ILinkClient._parse_incoming": 34,
+      "app/modules/wechatclawbot/wechatclawbot.py:ILinkClient._parse_incoming": 29,
       "app/modules/zspace/zspace.py:ZSpace.get_items": 16,
       "app/modules/zspace/zspace.py:ZSpace.get_tv_episodes": 18,
       "app/modules/zspace/zspace.py:ZSpace.get_webhook_message": 17,

tests/fixtures/architecture/mypy-baseline.json

@@ -2352,8 +2352,7 @@
   },
   "app/modules/wechatclawbot/wechatclawbot.py": {
     "arg-type": 2,
-    "assignment": 2,
-    "comparison-overlap": 3,
+    "assignment": 1,
     "no-any-return": 3,
     "no-untyped-def": 4,
     "type-arg": 2,

tests/test_nettest_probe_contract.py

@@ -121,6 +121,25 @@ def test_network_proxy_probe_uses_configured_github_identity(github_headers) ->
     assert "github-token" not in public_target.address
 
 
+def test_wechatclawbot_cdn_probe_accepts_parameterless_400() -> None:
+    """微信 ClawBot 媒体 CDN 对无参数裸请求返回 400，探针应视为连通正常。"""
+    captured = {}
+
+    async def respond(method, url, **options):
+        """模拟 CDN 对裸地址请求返回 400 空响应。"""
+        captured.update(method=method, url=url, **options)
+        return SimpleNamespace(status_code=400, headers={}, text="", aclose=AsyncMock())
+
+    transport = SimpleNamespace(request=AsyncMock(side_effect=respond))
+    result = asyncio.run(
+        _network_test_service(transport).execute(target_id="wechatclawbot_cdn")
+    )
+
+    assert result.success
+    assert captured["method"] == "GET"
+    assert captured["url"] == "https://novac2c.cdn.weixin.qq.com/c2c"
+
+
 def test_tmdb_web_probe_identifies_as_moviepilot() -> None:
     """TMDB 网站探针应使用程序自身 UA，浏览器 UA 会被站点人机校验页拦成 403。"""
     captured = {}

tests/test_wechatclawbot.py

@@ -1,95 +1,155 @@
 import json
 import tempfile
-import unittest
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 from app.modules.wechatclawbot import WechatClawBotModule
 from app.modules.wechatclawbot.wechatclawbot import ILinkClient, WechatClawBot
 
+CLIENT_BASE_URL = "https://ilinkai.weixin.qq.com"
 
-class WechatClawBotTest(unittest.TestCase):
-    def test_ilink_parse_incoming_uses_seq_as_message_id_fallback(self):
-        client = ILinkClient(base_url="https://ilinkai.weixin.qq.com")
 
-        message = client._parse_incoming(
-            {
-                "seq": 123456,
-                "from_user_id": "wxid_user_1",
-                "item_list": [{"type": 1, "text_item": {"text": "你好"}}],
-            }
-        )
+def _create_client(**kwargs) -> ILinkClient:
+    """构造指向 iLink 服务端的客户端实例。"""
+    return ILinkClient(base_url=CLIENT_BASE_URL, **kwargs)
 
-        self.assertIsNotNone(message)
-        self.assertEqual(message.message_id, "123456")
-        self.assertEqual(message.text, "你好")
 
-    def test_wechatclawbot_message_parser_deduplicates_message_id(self):
-        module = WechatClawBotModule()
-        body = json.dumps(
-            {
-                "__channel__": "wechatclawbot",
-                "userid": "wxid_user_1",
-                "username": "tester",
-                "message_id": "msg-1001",
-                "text": "刷新订阅",
-            }
-        )
+def _empty_state() -> dict:
+    """构造未登录的微信 ClawBot 持久化状态。"""
+    return {
+        "bot_token": None,
+        "account_id": None,
+        "sync_buf": None,
+        "qrcode": {},
+        "known_targets": {},
+        "user_context_tokens": {},
+        "base_url": CLIENT_BASE_URL,
+    }
 
-        with patch.object(
-            module,
-            "get_config",
-            return_value=SimpleNamespace(name="wechatclawbot-test", config={}),
-        ):
-            first = module.message_parser(
-                source="wechatclawbot-test",
-                body=body,
-                form={},
-                args={},
-            )
-            second = module.message_parser(
-                source="wechatclawbot-test",
-                body=body,
-                form={},
-                args={},
-            )
 
-        self.assertIsNotNone(first)
-        self.assertEqual(first.message_id, "msg-1001")
-        self.assertIsNone(second)
+def _create_bot() -> WechatClawBot:
+    """构造不自动轮询的微信 ClawBot 实例。"""
+    with patch.object(WechatClawBot, "_load_state", return_value=_empty_state()):
+        return WechatClawBot(name="wechatclawbot-test", auto_start_polling=False)
 
-    def test_ilink_extract_updates_keeps_empty_sync_buf(self):
-        client = ILinkClient(
-            base_url="https://ilinkai.weixin.qq.com",
-            bot_token="token",
-            sync_buf="cursor-old",
-        )
 
-        items, sync_buf = client._extract_updates(
-            {
-                "ret": 0,
-                "get_updates_buf": "",
-                "msgs": [
-                    {
-                        "message_id": "msg-1001",
-                        "from_user_id": "wxid_user_1",
-                        "item_list": [{"type": 1, "text_item": {"text": "你好"}}],
-                    }
-                ],
-            }
-        )
+def _module_config() -> SimpleNamespace:
+    """构造消息解析所需的渠道配置。"""
+    return SimpleNamespace(name="wechatclawbot-test", config={})
 
-        self.assertEqual(sync_buf, "")
-        self.assertEqual(len(items), 1)
 
-    def test_ilink_poll_updates_resets_sync_buf_when_server_returns_empty_cursor(self):
-        client = ILinkClient(
-            base_url="https://ilinkai.weixin.qq.com",
-            bot_token="token",
-            sync_buf="cursor-old",
-        )
-        response = MagicMock()
-        response.json.return_value = {
+def test_ilink_parse_incoming_uses_seq_as_message_id_fallback():
+    client = _create_client()
+
+    message = client._parse_incoming(
+        {
+            "seq": 123456,
+            "from_user_id": "wxid_user_1",
+            "item_list": [{"type": 1, "text_item": {"text": "你好"}}],
+        }
+    )
+
+    assert message is not None
+    assert message.message_id == "123456"
+    assert message.text == "你好"
+
+
+def test_ilink_parse_incoming_extracts_item_list_media():
+    """iLink 下行媒体放在 item_list 子项的 media 中，应解析为可直接下载的附件引用。"""
+    client = _create_client()
+    image_url = "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=image-1"
+    voice_url = "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=voice-1"
+    file_url = "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=file-1"
+    video_url = "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=video-1"
+    media_aes_key = "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY="
+
+    message = client._parse_incoming(
+        {
+            "message_id": "msg-media-1",
+            "from_user_id": "wxid_user_1",
+            "message_type": 1,
+            "item_list": [
+                {"type": 1, "text_item": {"text": "看这张图"}},
+                {
+                    "type": 2,
+                    "image_item": {
+                        "aeskey": "0123456789abcdef0123456789abcdef",
+                        "media": {"full_url": image_url, "aes_key": media_aes_key},
+                    },
+                },
+                {
+                    "type": 3,
+                    "voice_item": {"media": {"full_url": voice_url, "aes_key": media_aes_key}},
+                },
+                {
+                    "type": 4,
+                    "file_item": {
+                        "file_name": "报告.pdf",
+                        "size": 1024,
+                        "media": {"full_url": file_url, "aes_key": media_aes_key},
+                    },
+                },
+                {
+                    "type": 5,
+                    "video_item": {"media": {"full_url": video_url, "aes_key": media_aes_key}},
+                },
+            ],
+        }
+    )
+
+    assert message is not None
+    assert message.text == "看这张图"
+    assert len(message.images) == 1
+    image_payload = WechatClawBot._decode_ref_payload(message.images[0]["ref"], "image")
+    assert image_payload["url"] == image_url
+    assert image_payload["aeskey"] == "0123456789abcdef0123456789abcdef"
+    assert len(message.audio_refs) == 1
+    voice_payload = WechatClawBot._decode_ref_payload(message.audio_refs[0], "voice")
+    assert voice_payload["url"] == voice_url
+    assert voice_payload["aeskey"] == media_aes_key
+    assert len(message.files) == 2
+    file_payload = WechatClawBot._decode_ref_payload(message.files[0]["ref"], "file")
+    assert file_payload["url"] == file_url
+    assert file_payload["name"] == "报告.pdf"
+    assert message.files[0]["size"] == 1024
+    video_payload = WechatClawBot._decode_ref_payload(message.files[1]["ref"], "file")
+    assert video_payload["url"] == video_url
+    assert message.files[1]["name"] == "video.mp4"
+
+
+def test_ilink_parse_incoming_keeps_legacy_file_item_types():
+    """历史报文把文件放在 type=5/6 的 file_item 中，仍需保持兼容。"""
+    client = _create_client()
+    file_url = "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=legacy-1"
+
+    message = client._parse_incoming(
+        {
+            "message_id": "msg-file-legacy",
+            "from_user_id": "wxid_user_1",
+            "item_list": [
+                {
+                    "type": 6,
+                    "file_item": {
+                        "name": "archive.zip",
+                        "media": {"full_url": file_url},
+                    },
+                }
+            ],
+        }
+    )
+
+    assert message is not None
+    assert len(message.files) == 1
+    file_payload = WechatClawBot._decode_ref_payload(message.files[0]["ref"], "file")
+    assert file_payload["url"] == file_url
+    assert file_payload["name"] == "archive.zip"
+
+
+def test_ilink_extract_updates_keeps_empty_sync_buf():
+    client = _create_client(bot_token="token", sync_buf="cursor-old")
+
+    items, sync_buf = client._extract_updates(
+        {
             "ret": 0,
             "get_updates_buf": "",
             "msgs": [
@@ -100,276 +160,342 @@ class WechatClawBotTest(unittest.TestCase):
                 }
             ],
         }
+    )
 
-        with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response):
-            messages, sync_buf, result = client.poll_updates()
+    assert sync_buf == ""
+    assert len(items) == 1
 
-        self.assertTrue(result["success"])
-        self.assertEqual(sync_buf, "")
-        self.assertEqual(client.sync_buf, "")
-        self.assertEqual(len(messages), 1)
 
-    def test_ilink_poll_updates_accepts_canonical_payload_without_explicit_success_flag(self):
-        client = ILinkClient(
-            base_url="https://ilinkai.weixin.qq.com",
-            bot_token="token",
-            sync_buf="cursor-old",
-        )
-        response = MagicMock()
-        response.json.return_value = {
-            "sync_buf": "cursor-new",
-            "msgs": [
+def test_ilink_poll_updates_resets_sync_buf_when_server_returns_empty_cursor():
+    client = _create_client(bot_token="token", sync_buf="cursor-old")
+    response = MagicMock()
+    response.json.return_value = {
+        "ret": 0,
+        "get_updates_buf": "",
+        "msgs": [
+            {
+                "message_id": "msg-1001",
+                "from_user_id": "wxid_user_1",
+                "item_list": [{"type": 1, "text_item": {"text": "你好"}}],
+            }
+        ],
+    }
+
+    with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response):
+        messages, sync_buf, result = client.poll_updates()
+
+    assert result["success"]
+    assert sync_buf == ""
+    assert client.sync_buf == ""
+    assert len(messages) == 1
+
+
+def test_ilink_poll_updates_accepts_canonical_payload_without_explicit_success_flag():
+    client = _create_client(bot_token="token", sync_buf="cursor-old")
+    response = MagicMock()
+    response.json.return_value = {
+        "sync_buf": "cursor-new",
+        "msgs": [
+            {
+                "message_id": "msg-1002",
+                "from_user_id": "wxid_user_2",
+                "item_list": [{"type": 1, "text_item": {"text": "收到"}}],
+            }
+        ],
+    }
+
+    with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response):
+        messages, sync_buf, result = client.poll_updates()
+
+    assert result["success"]
+    assert sync_buf == "cursor-new"
+    assert client.sync_buf == "cursor-new"
+    assert len(messages) == 1
+    assert messages[0].text == "收到"
+
+
+def test_ilink_poll_updates_rejects_noncanonical_nested_success_payload():
+    client = _create_client(bot_token="token", sync_buf="cursor-old")
+    response = MagicMock()
+    response.json.return_value = {
+        "ret": 0,
+        "data": {
+            "get_updates_buf": "cursor-new",
+            "messages": [
                 {
-                    "message_id": "msg-1002",
-                    "from_user_id": "wxid_user_2",
-                    "item_list": [{"type": 1, "text_item": {"text": "收到"}}],
+                    "message_id": "msg-1001",
+                    "from_user_id": "wxid_user_1",
+                    "item_list": [{"type": 1, "text_item": {"text": "你好"}}],
+                }
+            ],
+        },
+    }
+
+    with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response):
+        messages, sync_buf, result = client.poll_updates()
+
+    assert not result["success"]
+    assert result["message"] == "轮询响应结构非官方，缺少顶层 msgs 字段"
+    assert sync_buf == "cursor-old"
+    assert client.sync_buf == "cursor-old"
+    assert messages == []
+
+
+def test_ilink_poll_updates_rejects_failed_payload_even_if_it_contains_messages():
+    client = _create_client(bot_token="token", sync_buf="cursor-old")
+    failed_response = MagicMock()
+    failed_response.json.return_value = {
+        "ret": -2,
+        "errmsg": "cursor invalid",
+        "data": {
+            "sync_buf": "cursor-old",
+            "messages": [
+                {
+                    "message_id": "msg-dup-1",
+                    "from_user_id": "wxid_user_1",
+                    "item_list": [{"type": 1, "text_item": {"text": "旧消息"}}],
+                }
+            ],
+        },
+    }
+
+    with patch(
+        "app.modules.wechatclawbot.wechatclawbot.RequestUtils.post",
+        return_value=failed_response,
+    ) as mock_post:
+        messages, sync_buf, result = client.poll_updates()
+
+    assert not result["success"]
+    assert result["message"] == "cursor invalid"
+    assert sync_buf == "cursor-old"
+    assert client.sync_buf == "cursor-old"
+    assert messages == []
+    mock_post.assert_called_once()
+    request_body = mock_post.call_args.kwargs["json"]
+    assert "get_updates_buf" in request_body
+    assert "sync_buf" not in request_body
+    assert "syncBuf" not in request_body
+
+
+def test_ilink_test_connection_accepts_getconfig_ilink_user_id_limitation():
+    client = _create_client(bot_token="token")
+    response = MagicMock()
+    response.json.return_value = {
+        "ret": -1,
+        "errmsg": "ilink_user_id required",
+    }
+
+    with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response):
+        ok, message = client.test_connection()
+
+    # `ilink_user_id required` 仅表示自检接口缺少额外参数，不代表连接失败：视为连接正常
+    assert ok
+    assert message == "连接正常"
+
+
+def test_wechatclawbot_message_parser_deduplicates_message_id():
+    module = WechatClawBotModule()
+    body = json.dumps(
+        {
+            "__channel__": "wechatclawbot",
+            "userid": "wxid_user_1",
+            "username": "tester",
+            "message_id": "msg-1001",
+            "text": "刷新订阅",
+        }
+    )
+
+    with patch.object(module, "get_config", return_value=_module_config()):
+        first = module.message_parser(
+            source="wechatclawbot-test",
+            body=body,
+            form={},
+            args={},
+        )
+        second = module.message_parser(
+            source="wechatclawbot-test",
+            body=body,
+            form={},
+            args={},
+        )
+
+    assert first is not None
+    assert first.message_id == "msg-1001"
+    assert second is None
+
+
+def test_wechatclawbot_message_parser_accepts_image_only_message():
+    """图片入站消息（无文本）应进入 Agent 链路，而不是被判为无效消息丢弃。"""
+    module = WechatClawBotModule()
+    client = _create_client()
+    parsed_message = client._parse_incoming(
+        {
+            "message_id": "msg-image-only",
+            "from_user_id": "wxid_user_1",
+            "item_list": [
+                {
+                    "type": 2,
+                    "image_item": {
+                        "aeskey": "0123456789abcdef0123456789abcdef",
+                        "media": {
+                            "full_url": "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=image-only",
+                            "aes_key": "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=",
+                        },
+                    },
                 }
             ],
         }
+    )
+    assert parsed_message is not None
 
-        with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response):
-            messages, sync_buf, result = client.poll_updates()
-
-        self.assertTrue(result["success"])
-        self.assertEqual(sync_buf, "cursor-new")
-        self.assertEqual(client.sync_buf, "cursor-new")
-        self.assertEqual(len(messages), 1)
-        self.assertEqual(messages[0].text, "收到")
-
-    def test_ilink_poll_updates_rejects_noncanonical_nested_success_payload(self):
-        client = ILinkClient(
-            base_url="https://ilinkai.weixin.qq.com",
-            bot_token="token",
-            sync_buf="cursor-old",
-        )
-        response = MagicMock()
-        response.json.return_value = {
-            "ret": 0,
-            "data": {
-                "get_updates_buf": "cursor-new",
-                "messages": [
-                    {
-                        "message_id": "msg-1001",
-                        "from_user_id": "wxid_user_1",
-                        "item_list": [{"type": 1, "text_item": {"text": "你好"}}],
-                    }
-                ],
-            },
-        }
-
-        with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response):
-            messages, sync_buf, result = client.poll_updates()
-
-        self.assertFalse(result["success"])
-        self.assertEqual(result["message"], "轮询响应结构非官方，缺少顶层 msgs 字段")
-        self.assertEqual(sync_buf, "cursor-old")
-        self.assertEqual(client.sync_buf, "cursor-old")
-        self.assertEqual(messages, [])
-
-    def test_ilink_poll_updates_rejects_failed_payload_even_if_it_contains_messages(self):
-        client = ILinkClient(
-            base_url="https://ilinkai.weixin.qq.com",
-            bot_token="token",
-            sync_buf="cursor-old",
-        )
-        failed_response = MagicMock()
-        failed_response.json.return_value = {
-            "ret": -2,
-            "errmsg": "cursor invalid",
-            "data": {
-                "sync_buf": "cursor-old",
-                "messages": [
-                    {
-                        "message_id": "msg-dup-1",
-                        "from_user_id": "wxid_user_1",
-                        "item_list": [{"type": 1, "text_item": {"text": "旧消息"}}],
-                    }
-                ],
-            },
-        }
-
-        with patch(
-            "app.modules.wechatclawbot.wechatclawbot.RequestUtils.post",
-            return_value=failed_response,
-        ) as mock_post:
-            messages, sync_buf, result = client.poll_updates()
-
-        self.assertFalse(result["success"])
-        self.assertEqual(result["message"], "cursor invalid")
-        self.assertEqual(sync_buf, "cursor-old")
-        self.assertEqual(client.sync_buf, "cursor-old")
-        self.assertEqual(messages, [])
-        mock_post.assert_called_once()
-        request_body = mock_post.call_args.kwargs["json"]
-        self.assertIn("get_updates_buf", request_body)
-        self.assertNotIn("sync_buf", request_body)
-        self.assertNotIn("syncBuf", request_body)
-
-    def test_ilink_test_connection_accepts_getconfig_ilink_user_id_limitation(self):
-        client = ILinkClient(
-            base_url="https://ilinkai.weixin.qq.com",
-            bot_token="token",
-        )
-        response = MagicMock()
-        response.json.return_value = {
-            "ret": -1,
-            "errmsg": "ilink_user_id required",
-        }
-
-        with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response):
-            ok, message = client.test_connection()
-
-        # `ilink_user_id required` 仅表示自检接口缺少额外参数，不代表连接失败：视为连接正常
-        self.assertTrue(ok)
-        self.assertEqual(message, "连接正常")
-
-    def test_wechatclawbot_send_msg_uses_plain_text_payload(self):
-        state = {
-            "bot_token": None,
-            "account_id": None,
-            "sync_buf": None,
-            "qrcode": {},
-            "known_targets": {},
-            "user_context_tokens": {},
-            "base_url": "https://ilinkai.weixin.qq.com",
-        }
-        with patch.object(WechatClawBot, "_load_state", return_value=state):
-            bot = WechatClawBot(name="wechatclawbot-test", auto_start_polling=False)
-
-        mock_client = MagicMock()
-        mock_client.send_text.return_value = True
-
-        with patch.object(bot, "_build_client", return_value=mock_client):
-            result = bot.send_msg(
-                title="测试标题",
-                text="测试正文",
-                userid="wxid_user_1",
-                link="https://example.com/detail",
-            )
-
-        self.assertTrue(result)
-        mock_client.send_text.assert_called_once_with(
-            to_user="wxid_user_1",
-            text="测试标题\n\n测试正文\n\n查看详情：https://example.com/detail",
-            context_token=None,
+    with patch.object(module, "get_config", return_value=_module_config()):
+        parsed = module.message_parser(
+            source="wechatclawbot-test",
+            body=json.dumps(parsed_message.to_message_payload()),
+            form={},
+            args={},
         )
 
-    def test_wechatclawbot_send_msg_prefers_text_when_text_and_image_coexist(self):
-        state = {
-            "bot_token": None,
-            "account_id": None,
-            "sync_buf": None,
-            "qrcode": {},
-            "known_targets": {},
-            "user_context_tokens": {},
-            "base_url": "https://ilinkai.weixin.qq.com",
+    assert parsed is not None
+    assert parsed.images is not None
+    assert parsed.images[0].ref.startswith("wxclaw://image/")
+    assert not parsed.text
+
+
+def test_wechatclawbot_message_parser_logs_unrecognized_content():
+    """无法识别的入站内容应留下诊断日志，而不是静默丢弃。"""
+    module = WechatClawBotModule()
+    body = json.dumps(
+        {
+            "__channel__": "wechatclawbot",
+            "userid": "wxid_user_1",
+            "message_id": "msg-unknown-1",
+            "item_list": [{"type": 99, "unknown_item": {"foo": "bar"}}],
         }
-        with patch.object(WechatClawBot, "_load_state", return_value=state):
-            bot = WechatClawBot(name="wechatclawbot-test", auto_start_polling=False)
+    )
 
-        mock_client = MagicMock()
-        mock_client.send_text.return_value = True
-        mock_client.send_image_png.return_value = True
-        mock_client.send_image_text_png.return_value = True
+    with (
+        patch.object(module, "get_config", return_value=_module_config()),
+        patch("app.modules.wechatclawbot.logger") as mock_logger,
+    ):
+        parsed = module.message_parser(
+            source="wechatclawbot-test",
+            body=body,
+            form={},
+            args={},
+        )
 
+    assert parsed is None
+    mock_logger.info.assert_called_once()
+    assert "未识别到有效内容" in mock_logger.info.call_args.args[0]
+
+
+def test_wechatclawbot_send_msg_uses_plain_text_payload():
+    bot = _create_bot()
+    mock_client = MagicMock()
+    mock_client.send_text.return_value = True
+
+    with patch.object(bot, "_build_client", return_value=mock_client):
+        result = bot.send_msg(
+            title="测试标题",
+            text="测试正文",
+            userid="wxid_user_1",
+            link="https://example.com/detail",
+        )
+
+    assert result
+    mock_client.send_text.assert_called_once_with(
+        to_user="wxid_user_1",
+        text="测试标题\n\n测试正文\n\n查看详情：https://example.com/detail",
+        context_token=None,
+    )
+
+
+def test_wechatclawbot_send_msg_prefers_text_when_text_and_image_coexist():
+    bot = _create_bot()
+    mock_client = MagicMock()
+    mock_client.send_text.return_value = True
+    mock_client.send_image_png.return_value = True
+    mock_client.send_image_text_png.return_value = True
+
+    with (
+        patch.object(bot, "_build_client", return_value=mock_client),
+        patch.object(bot, "_load_remote_image", return_value=b"image-bytes") as mock_load_image,
+    ):
+        result = bot.send_msg(
+            title="测试标题",
+            text="测试正文",
+            image="https://example.com/test.png",
+            userid="wxid_user_1",
+        )
+
+    assert result
+    mock_load_image.assert_not_called()
+    mock_client.send_text.assert_called_once_with(
+        to_user="wxid_user_1",
+        text="测试标题\n\n测试正文",
+        context_token=None,
+    )
+    mock_client.send_image_png.assert_not_called()
+    mock_client.send_image_text_png.assert_not_called()
+
+
+def test_wechatclawbot_send_file_prefers_image_when_image_file_has_caption():
+    bot = _create_bot()
+    mock_client = MagicMock()
+    mock_client.send_text.return_value = True
+    mock_client.send_image_png.return_value = True
+
+    with tempfile.NamedTemporaryFile(suffix=".png") as image_file:
+        image_file.write(b"\x89PNG\r\n\x1a\nfake-png")
+        image_file.flush()
         with (
             patch.object(bot, "_build_client", return_value=mock_client),
-            patch.object(bot, "_load_remote_image", return_value=b"image-bytes") as mock_load_image,
+            patch.object(bot, "_guess_mime_type", return_value="image/png"),
         ):
-            result = bot.send_msg(
-                title="测试标题",
-                text="测试正文",
-                image="https://example.com/test.png",
+            result = bot.send_file(
+                file_path=image_file.name,
+                title="图片标题",
+                text="图片说明",
                 userid="wxid_user_1",
             )
 
-        self.assertTrue(result)
-        mock_load_image.assert_not_called()
-        mock_client.send_text.assert_called_once_with(
-            to_user="wxid_user_1",
-            text="测试标题\n\n测试正文",
-            context_token=None,
-        )
-        mock_client.send_image_png.assert_not_called()
-        mock_client.send_image_text_png.assert_not_called()
+    assert result
+    mock_client.send_text.assert_not_called()
+    mock_client.send_image_png.assert_called_once_with(
+        to_user="wxid_user_1",
+        image_bytes=b"\x89PNG\r\n\x1a\nfake-png",
+        context_token=None,
+    )
 
-    def test_wechatclawbot_send_file_prefers_image_when_image_file_has_caption(self):
-        state = {
-            "bot_token": None,
-            "account_id": None,
-            "sync_buf": None,
-            "qrcode": {},
-            "known_targets": {},
-            "user_context_tokens": {},
-            "base_url": "https://ilinkai.weixin.qq.com",
-        }
-        with patch.object(WechatClawBot, "_load_state", return_value=state):
-            bot = WechatClawBot(name="wechatclawbot-test", auto_start_polling=False)
 
-        mock_client = MagicMock()
-        mock_client.send_text.return_value = True
-        mock_client.send_image_png.return_value = True
+def test_wechatclawbot_send_file_prefers_file_when_generic_file_has_caption():
+    bot = _create_bot()
+    mock_client = MagicMock()
+    mock_client.send_text.return_value = True
+    mock_client.send_file_bytes.return_value = True
 
-        with tempfile.NamedTemporaryFile(suffix=".png") as image_file:
-            image_file.write(b"\x89PNG\r\n\x1a\nfake-png")
-            image_file.flush()
-            with (
-                patch.object(bot, "_build_client", return_value=mock_client),
-                patch.object(bot, "_guess_mime_type", return_value="image/png"),
-            ):
-                result = bot.send_file(
-                    file_path=image_file.name,
-                    title="图片标题",
-                    text="图片说明",
-                    userid="wxid_user_1",
-                )
+    with tempfile.NamedTemporaryFile(suffix=".txt") as text_file:
+        text_file.write(b"plain-text")
+        text_file.flush()
+        with (
+            patch.object(bot, "_build_client", return_value=mock_client),
+            patch.object(bot, "_guess_mime_type", return_value="text/plain"),
+        ):
+            result = bot.send_file(
+                file_path=text_file.name,
+                file_name="report.txt",
+                title="文件标题",
+                text="文件说明",
+                userid="wxid_user_1",
+            )
 
-        self.assertTrue(result)
-        mock_client.send_text.assert_not_called()
-        mock_client.send_image_png.assert_called_once_with(
-            to_user="wxid_user_1",
-            image_bytes=b"\x89PNG\r\n\x1a\nfake-png",
-            context_token=None,
-        )
-
-    def test_wechatclawbot_send_file_prefers_file_when_generic_file_has_caption(self):
-        state = {
-            "bot_token": None,
-            "account_id": None,
-            "sync_buf": None,
-            "qrcode": {},
-            "known_targets": {},
-            "user_context_tokens": {},
-            "base_url": "https://ilinkai.weixin.qq.com",
-        }
-        with patch.object(WechatClawBot, "_load_state", return_value=state):
-            bot = WechatClawBot(name="wechatclawbot-test", auto_start_polling=False)
-
-        mock_client = MagicMock()
-        mock_client.send_text.return_value = True
-        mock_client.send_file_bytes.return_value = True
-
-        with tempfile.NamedTemporaryFile(suffix=".txt") as text_file:
-            text_file.write(b"plain-text")
-            text_file.flush()
-            with (
-                patch.object(bot, "_build_client", return_value=mock_client),
-                patch.object(bot, "_guess_mime_type", return_value="text/plain"),
-            ):
-                result = bot.send_file(
-                    file_path=text_file.name,
-                    file_name="report.txt",
-                    title="文件标题",
-                    text="文件说明",
-                    userid="wxid_user_1",
-                )
-
-        self.assertTrue(result)
-        mock_client.send_text.assert_not_called()
-        mock_client.send_file_bytes.assert_called_once_with(
-            to_user="wxid_user_1",
-            file_bytes=b"plain-text",
-            file_name="report.txt",
-            mime_type="text/plain",
-            context_token=None,
-        )
+    assert result
+    mock_client.send_text.assert_not_called()
+    mock_client.send_file_bytes.assert_called_once_with(
+        to_user="wxid_user_1",
+        file_bytes=b"plain-text",
+        file_name="report.txt",
+        mime_type="text/plain",
+        context_token=None,
+    )