feat(imessage): 实现完整的 iMessage 适配器基础模块

新增 iMessage 通道适配器完整实现，包含：
1. 核心适配器与工具工厂导出
2. 运行时存储、反射防护、会话路由等基础组件
3. 消息信封、线程管理、回复上下文格式化
4. Tapback 表情反应处理、自定义异常体系
5. 审批按钮、联系人解析、速率限制功能
6. 文本净化、目标解析、缓存管理模块
7. 配置 schema、多账户支持、安装向导等配置模块
8. 审计日志、媒体AI处理等扩展功能

backend/package/yuxi/channels/adapters/imessage/__init__.py

@@ -0,0 +1,4 @@
+from yuxi.channels.adapters.imessage.adapter import IMessageAdapter
+from yuxi.channels.adapters.imessage.agent_tools import IMessageAgentToolFactory
+
+__all__ = ["IMessageAdapter", "IMessageAgentToolFactory"]

backend/package/yuxi/channels/adapters/imessage/accounts.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class IMessageAccount:
+    account_id: str
+    name: str = ""
+    server_url: str = "http://localhost:1234"
+    password: str = ""
+    dm_policy: str = "pairing"
+    group_policy: str = "allowlist"
+    allow_from: list[str] = field(default_factory=list)
+    group_allow_from: list[str] = field(default_factory=list)
+    require_mention: bool = False
+    default_to: str = ""
+    enabled: bool = True
+
+
+def resolve_accounts(config: dict[str, Any]) -> list[IMessageAccount]:
+    """解析多账户配置。
+
+    支持两种配置模式：
+    1. 顶层单账户：config 直接包含 server_url / password
+    2. accounts 块：config["accounts"] 为 {accountId: {...}, ...}
+
+    返回扁平化的 IMessageAccount 列表。
+    """
+    accounts_raw = config.get("accounts")
+
+    if not accounts_raw:
+        account = _account_from_top_level(config)
+        return [account] if account.server_url else []
+
+    result: list[IMessageAccount] = []
+    default_account_id = config.get("defaultAccount", config.get("default_account", ""))
+
+    for account_id, acct_config in accounts_raw.items():
+        if not isinstance(acct_config, dict):
+            continue
+        merged = _merge_account_config(config, account_id, acct_config)
+        result.append(merged)
+
+    if default_account_id:
+        result.sort(key=lambda a: 0 if a.account_id == default_account_id else 1)
+
+    return result
+
+
+def list_enabled_accounts(config: dict[str, Any]) -> list[IMessageAccount]:
+    accounts = resolve_accounts(config)
+    return [a for a in accounts if a.enabled]
+
+
+def resolve_default_account(config: dict[str, Any]) -> IMessageAccount | None:
+    accounts = list_enabled_accounts(config)
+    if not accounts:
+        return None
+    default_id = config.get("defaultAccount", config.get("default_account", ""))
+    for acct in accounts:
+        if acct.account_id == default_id:
+            return acct
+    return accounts[0]
+
+
+def _account_from_top_level(config: dict[str, Any]) -> IMessageAccount:
+    return IMessageAccount(
+        account_id="default",
+        name=config.get("name", "iMessage"),
+        server_url=config.get("server_url", "http://localhost:1234"),
+        password=config.get("password", ""),
+        dm_policy=config.get("dmPolicy", config.get("dm_policy", "pairing")),
+        group_policy=config.get("groupPolicy", config.get("group_policy", "allowlist")),
+        allow_from=config.get("allowFrom", config.get("allow_from", [])),
+        group_allow_from=config.get("groupAllowFrom", config.get("group_allow_from", [])),
+        require_mention=config.get("requireMention", config.get("require_mention", False)),
+        default_to=config.get("defaultTo", config.get("default_to", "")),
+    )
+
+
+def _merge_account_config(top: dict[str, Any], account_id: str, acct: dict[str, Any]) -> IMessageAccount:
+    return IMessageAccount(
+        account_id=account_id,
+        name=acct.get("name", account_id),
+        server_url=acct.get("server_url", top.get("server_url", "http://localhost:1234")),
+        password=acct.get("password", top.get("password", "")),
+        dm_policy=acct.get("dmPolicy", acct.get("dm_policy", top.get("dmPolicy", top.get("dm_policy", "pairing")))),
+        group_policy=acct.get(
+            "groupPolicy", acct.get("group_policy", top.get("groupPolicy", top.get("group_policy", "allowlist")))
+        ),
+        allow_from=acct.get("allowFrom", acct.get("allow_from", top.get("allowFrom", top.get("allow_from", [])))),
+        group_allow_from=acct.get(
+            "groupAllowFrom", acct.get("group_allow_from", top.get("groupAllowFrom", top.get("group_allow_from", [])))
+        ),
+        require_mention=acct.get(
+            "requireMention", acct.get("require_mention", top.get("requireMention", top.get("require_mention", False)))
+        ),
+        default_to=acct.get("defaultTo", acct.get("default_to", top.get("defaultTo", top.get("default_to", "")))),
+        enabled=acct.get("enabled", True),
+    )

backend/package/yuxi/channels/adapters/imessage/agent_tools.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def describe_imessage_pairing_tool() -> dict:
+    return {
+        "type": "function",
+        "function": {
+            "name": "imessage_manage_pairing",
+            "description": "管理 iMessage 配对请求：列出待审批的用户、批准或拒绝配对",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "action": {
+                        "type": "string",
+                        "enum": ["list", "approve", "reject"],
+                        "description": "操作类型：list=列出待审批, approve=批准配对, reject=拒绝配对",
+                    },
+                    "code": {
+                        "type": "string",
+                        "description": "配对码（approve/reject 操作需要）",
+                    },
+                },
+                "required": ["action"],
+            },
+        },
+    }
+
+
+def describe_imessage_allowlist_tool() -> dict:
+    return {
+        "type": "function",
+        "function": {
+            "name": "imessage_manage_allowlist",
+            "description": "管理 iMessage 白名单：添加或移除联系人/群组白名单",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "action": {
+                        "type": "string",
+                        "enum": ["add_dm", "remove_dm", "add_group", "remove_group", "list"],
+                        "description": "操作类型",
+                    },
+                    "handle": {
+                        "type": "string",
+                        "description": "电话号码或 chat_guid",
+                    },
+                },
+                "required": ["action"],
+            },
+        },
+    }
+
+
+def describe_imessage_security_status_tool() -> dict:
+    return {
+        "type": "function",
+        "function": {
+            "name": "imessage_security_status",
+            "description": "查看 iMessage 渠道的安全状态和警告信息",
+            "parameters": {
+                "type": "object",
+                "properties": {},
+                "required": [],
+            },
+        },
+    }
+
+
+class IMessageAgentToolFactory:
+    def __init__(self, adapter: Any):
+        self._adapter = adapter
+
+    def list_tools(self) -> list[dict]:
+        return [
+            describe_imessage_pairing_tool(),
+            describe_imessage_allowlist_tool(),
+            describe_imessage_security_status_tool(),
+        ]
+
+    async def execute_tool(self, tool_name: str, params: dict) -> Any:
+        if tool_name == "imessage_manage_pairing":
+            return await self._manage_pairing(params)
+        if tool_name == "imessage_manage_allowlist":
+            return await self._manage_allowlist(params)
+        if tool_name == "imessage_security_status":
+            return self._security_status()
+        raise ValueError(f"Unknown tool: {tool_name}")
+
+    async def _manage_pairing(self, params: dict) -> dict:
+        action = params.get("action", "list")
+
+        if action == "list":
+            pending = self._adapter.list_pending_pairings()
+            paired = self._adapter._pairing.get_paired_handles()
+            return {"pending": pending, "paired": paired}
+
+        code = params.get("code", "")
+        if action == "approve":
+            ok = self._adapter.approve_pairing(code)
+            if ok:
+                await self._adapter._pairing.save_allowlist()
+            return {"success": ok, "message": "Pairing approved" if ok else "Invalid or expired code"}
+
+        if action == "reject":
+            ok = self._adapter.reject_pairing(code)
+            return {"success": ok, "message": "Pairing rejected" if ok else "Invalid or expired code"}
+
+        return {"success": False, "error": f"Unknown action: {action}"}
+
+    async def _manage_allowlist(self, params: dict) -> dict:
+        action = params.get("action", "list")
+
+        if action == "list":
+            return {
+                "dm_allowlist": self._adapter._security.allow_list,
+                "group_allowlist": self._adapter._security.group_allow_list,
+            }
+
+        handle = params.get("handle", "")
+        if not handle:
+            return {"success": False, "error": "handle is required"}
+
+        if action == "add_dm":
+            self._adapter._security.add_to_allow_list(handle)
+            return {"success": True, "message": f"Added {handle} to DM allowlist"}
+
+        if action == "remove_dm":
+            ok = self._adapter._security.remove_from_allow_list(handle)
+            return {"success": ok, "message": f"{'Removed' if ok else 'Not found'}: {handle}"}
+
+        if action == "add_group":
+            self._adapter._security.add_to_group_allow_list(handle)
+            return {"success": True, "message": f"Added {handle} to group allowlist"}
+
+        if action == "remove_group":
+            ok = self._adapter._security.remove_from_group_allow_list(handle)
+            return {"success": ok, "message": f"{'Removed' if ok else 'Not found'}: {handle}"}
+
+        return {"success": False, "error": f"Unknown action: {action}"}
+
+    def _security_status(self) -> dict:
+        warnings = self._adapter.get_security_warnings()
+        return {
+            "dm_policy": self._adapter._security.dm_policy.value,
+            "group_policy": self._adapter._security.group_policy.value,
+            "require_mention": self._adapter._security.require_mention,
+            "warnings": warnings,
+            "paired_count": len(self._adapter._pairing.get_paired_handles()),
+        }

backend/package/yuxi/channels/adapters/imessage/allowlist.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class IMessageAllowlist:
+    def __init__(self, config: dict[str, Any]):
+        self._dm_allowlist: list[str] = _normalize_entries(config.get("allowFrom", config.get("allow_from", [])))
+        self._group_allowlist: list[str] = _normalize_chat_entries(
+            config.get("groupAllowFrom", config.get("group_allow_from", []))
+        )
+
+    def check_dm(self, sender_handle: str) -> bool:
+        return _match_handle(sender_handle, self._dm_allowlist)
+
+    def check_group(self, chat_guid: str) -> bool:
+        return _match_chat(chat_guid, self._group_allowlist)
+
+    def add_dm(self, handle: str) -> None:
+        handle_clean = _normalize_handle(handle)
+        if handle_clean not in self._dm_allowlist:
+            self._dm_allowlist.append(handle_clean)
+
+    def remove_dm(self, handle: str) -> bool:
+        handle_clean = _normalize_handle(handle)
+        if handle_clean in self._dm_allowlist:
+            self._dm_allowlist.remove(handle_clean)
+            return True
+        return False
+
+    def add_group(self, chat_guid: str) -> None:
+        if chat_guid not in self._group_allowlist:
+            self._group_allowlist.append(chat_guid)
+
+    def remove_group(self, chat_guid: str) -> bool:
+        if chat_guid in self._group_allowlist:
+            self._group_allowlist.remove(chat_guid)
+            return True
+        return False
+
+    @property
+    def dm_entries(self) -> list[str]:
+        return list(self._dm_allowlist)
+
+    @property
+    def group_entries(self) -> list[str]:
+        return list(self._group_allowlist)
+
+    @property
+    def dm_count(self) -> int:
+        return len(self._dm_allowlist)
+
+    @property
+    def group_count(self) -> int:
+        return len(self._group_allowlist)
+
+
+def _normalize_handle(handle: str) -> str:
+    return handle.strip().replace(" ", "").lstrip("+")
+
+
+def _normalize_entries(entries: list[str]) -> list[str]:
+    return [_normalize_handle(e) for e in entries if e]
+
+
+def _normalize_chat_entries(entries: list[str]) -> list[str]:
+    return [e.strip() for e in entries if e]
+
+
+def _match_handle(target: str, allowlist: list[str]) -> bool:
+    if not allowlist or "*" in allowlist:
+        return True
+    target_clean = _normalize_handle(target)
+    for entry in allowlist:
+        if _normalize_handle(entry) == target_clean:
+            return True
+    return False
+
+
+def _match_chat(target: str, allowlist: list[str]) -> bool:
+    if not allowlist or "*" in allowlist:
+        return True
+    return target in allowlist

backend/package/yuxi/channels/adapters/imessage/approval.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+
+class ApprovalStatus(StrEnum):
+    PENDING = "pending"
+    APPROVED = "approved"
+    REJECTED = "rejected"
+    EXPIRED = "expired"
+
+
+@dataclass
+class ApprovalRequest:
+    request_id: str
+    channel_chat_id: str
+    channel_user_id: str
+    action: str
+    action_payload: dict[str, Any] = field(default_factory=dict)
+    status: ApprovalStatus = ApprovalStatus.PENDING
+    created_at: float = field(default_factory=time.time)
+    expires_at: float = field(default_factory=lambda: time.time() + 300)
+    approved_by: str = ""
+    result: Any = None
+
+
+class ExecApprovalManager:
+    def __init__(self, approval_ttl_s: float = 300.0, max_pending: int = 10):
+        self._pending: dict[str, ApprovalRequest] = {}
+        self._history: list[ApprovalRequest] = []
+        self._approval_ttl = approval_ttl_s
+        self._max_pending = max_pending
+
+    def request_approval(
+        self,
+        request_id: str,
+        channel_chat_id: str,
+        channel_user_id: str,
+        action: str,
+        action_payload: dict[str, Any] | None = None,
+    ) -> ApprovalRequest:
+        self._cleanup()
+
+        if len(self._pending) >= self._max_pending:
+            oldest = min(self._pending.values(), key=lambda r: r.created_at)
+            self._pending.pop(oldest.request_id, None)
+
+        req = ApprovalRequest(
+            request_id=request_id,
+            channel_chat_id=channel_chat_id,
+            channel_user_id=channel_user_id,
+            action=action,
+            action_payload=action_payload or {},
+        )
+        self._pending[request_id] = req
+        logger.info(f"[iMessage/Approval] New request: {request_id} action={action}")
+        return req
+
+    def approve(self, request_id: str, approved_by: str = "") -> ApprovalRequest | None:
+        self._cleanup()
+        req = self._pending.get(request_id)
+        if req is None:
+            return None
+        req.status = ApprovalStatus.APPROVED
+        req.approved_by = approved_by
+        self._pending.pop(request_id, None)
+        self._history.append(req)
+        logger.info(f"[iMessage/Approval] Approved: {request_id}")
+        return req
+
+    def reject(self, request_id: str, reason: str = "") -> ApprovalRequest | None:
+        self._cleanup()
+        req = self._pending.get(request_id)
+        if req is None:
+            return None
+        req.status = ApprovalStatus.REJECTED
+        self._pending.pop(request_id, None)
+        self._history.append(req)
+        logger.info(f"[iMessage/Approval] Rejected: {request_id} ({reason})")
+        return req
+
+    def get(self, request_id: str) -> ApprovalRequest | None:
+        return self._pending.get(request_id)
+
+    def list_pending(self) -> list[ApprovalRequest]:
+        self._cleanup()
+        return list(self._pending.values())
+
+    def _cleanup(self) -> None:
+        now = time.time()
+        expired = [rid for rid, req in self._pending.items() if req.expires_at < now]
+        for rid in expired:
+            req = self._pending.pop(rid)
+            req.status = ApprovalStatus.EXPIRED
+            self._history.append(req)

backend/package/yuxi/channels/adapters/imessage/approval_buttons.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def format_approval_message(request_id: str, action: str, payload: dict[str, Any] | None = None) -> str:
+    """生成审批请求消息文本。
+
+    iMessage 不支持内联按钮，使用文本命令模拟审批。
+    """
+    lines = [
+        "Pending Approval Required",
+        f"Action: {action}",
+        f"Request ID: {request_id}",
+        "",
+    ]
+
+    if payload:
+        for key, val in payload.items():
+            val_str = str(val)[:100]
+            lines.append(f"  {key}: {val_str}")
+
+    lines.extend(
+        [
+            "",
+            "Reply with:",
+            f"  approve {request_id}  — to approve",
+            f"  reject {request_id}   — to reject",
+        ]
+    )
+
+    return "\n".join(lines)
+
+
+def format_approval_result(request_id: str, approved: bool, message: str = "") -> str:
+    status = "APPROVED" if approved else "REJECTED"
+    lines = [f"Approval {status}: {request_id}"]
+    if message:
+        lines.append(f"  {message}")
+    return "\n".join(lines)
+
+
+def parse_approval_command(text: str) -> tuple[str | None, str | None]:
+    """解析审批文本命令。
+
+    返回 (action, request_id)，其中 action 为 "approve" 或 "reject"。
+    """
+    text_lower = text.strip().lower()
+    parts = text_lower.split(maxsplit=1)
+    if len(parts) < 2:
+        return None, None
+    action = parts[0]
+    request_id = parts[1].strip()
+    if action in ("approve", "reject"):
+        return action, request_id
+    return None, None

backend/package/yuxi/channels/adapters/imessage/audit.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+
+class AuditLogger:
+    def __init__(self, log_dir: str | None = None):
+        self._log_dir = Path(log_dir or str(Path.home() / ".yuxi" / "imessage" / "audit"))
+        self._log_dir.mkdir(parents=True, exist_ok=True)
+        self._buffer: list[dict[str, Any]] = []
+
+    def record(
+        self,
+        event: str,
+        channel_chat_id: str = "",
+        channel_user_id: str = "",
+        decision: str = "",
+        reason: str = "",
+        metadata: dict[str, Any] | None = None,
+    ) -> None:
+        entry = {
+            "timestamp": time.time(),
+            "event": event,
+            "channel_chat_id": channel_chat_id,
+            "channel_user_id": channel_user_id,
+            "decision": decision,
+            "reason": reason,
+            "metadata": metadata or {},
+        }
+        self._buffer.append(entry)
+        logger.info(f"[iMessage/Audit] {event}: {decision} — {reason}")
+
+    def record_dm_access(self, sender: str, allowed: bool, reason: str = "") -> None:
+        self.record(
+            event="dm_access",
+            channel_user_id=sender,
+            decision="allowed" if allowed else "blocked",
+            reason=reason,
+        )
+
+    def record_group_access(self, chat_guid: str, allowed: bool, reason: str = "") -> None:
+        self.record(
+            event="group_access",
+            channel_chat_id=chat_guid,
+            decision="allowed" if allowed else "blocked",
+            reason=reason,
+        )
+
+    def record_pairing(self, code: str, action: str, sender: str) -> None:
+        self.record(
+            event="pairing",
+            channel_user_id=sender,
+            decision=action,
+            reason=f"code={code}",
+        )
+
+    async def flush(self) -> None:
+        if not self._buffer:
+            return
+        date_str = time.strftime("%Y-%m-%d")
+        file_path = self._log_dir / f"audit-{date_str}.jsonl"
+        with file_path.open("a", encoding="utf-8") as f:
+            for entry in self._buffer:
+                f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+        count = len(self._buffer)
+        self._buffer.clear()
+        logger.debug(f"[iMessage/Audit] Flushed {count} audit entries")
+
+    def get_recent(self, limit: int = 100) -> list[dict[str, Any]]:
+        result: list[dict[str, Any]] = []
+        files = sorted(self._log_dir.glob("audit-*.jsonl"), reverse=True)
+        for fp in files:
+            if len(result) >= limit:
+                break
+            try:
+                lines = fp.read_text(encoding="utf-8").strip().split("\n")
+                for line in reversed(lines):
+                    if not line.strip():
+                        continue
+                    result.append(json.loads(line))
+                    if len(result) >= limit:
+                        break
+            except (json.JSONDecodeError, OSError):
+                continue
+        return result[:limit]

backend/package/yuxi/channels/adapters/imessage/bridge_client.py

@@ -0,0 +1,327 @@
+from __future__ import annotations
+
+import asyncio
+import os
+import time
+from typing import Any
+from urllib.parse import urlparse
+
+import aiohttp
+
+from yuxi.channels.adapters.imessage.exceptions import BlueBubblesHTTPError
+from yuxi.channels.infra.circuit_breaker import CircuitBreaker
+from yuxi.channels.models import DeliveryResult
+from yuxi.utils.logging_config import logger
+
+
+def is_test_env() -> bool:
+    return os.environ.get("NODE_ENV") == "test" or os.environ.get("PYTEST_VERSION", "") != ""
+
+
+class BlueBubblesClient:
+    def __init__(self, config: dict[str, Any]):
+        self._server_url = config.get("server_url", "http://localhost:1234").rstrip("/")
+        self._password = config.get("password", "")
+        self._http_timeout_s = config.get("http_timeout_s", config.get("httpTimeoutS", 30.0))
+        self._session: aiohttp.ClientSession | None = None
+        self._max_retries = config.get("max_retries", 3)
+        self._breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30.0)
+        self._probe_timeout_ms = config.get("probeTimeoutMs", config.get("probe_timeout_ms", 10000))
+        self._remote_attachment_roots: list[str] = config.get(
+            "remoteAttachmentRoots", config.get("remote_attachment_roots", [])
+        )
+        self._auth_backoff_s = 1.0
+        self._last_auth_failure: float | None = None
+
+    async def __aenter__(self) -> BlueBubblesClient:
+        self._session = aiohttp.ClientSession(
+            headers={"X-BlueBubbles-Password": self._password},
+            timeout=aiohttp.ClientTimeout(total=self._http_timeout_s),
+        )
+        return self
+
+    async def __aexit__(self, *args) -> None:
+        await self.close()
+
+    async def close(self) -> None:
+        if self._session and not self._session.closed:
+            await self._session.close()
+        self._session = None
+
+    async def _ensure_session(self) -> aiohttp.ClientSession:
+        if self._session is None or self._session.closed:
+            self._session = aiohttp.ClientSession(
+                headers={"X-BlueBubbles-Password": self._password},
+                timeout=aiohttp.ClientTimeout(total=self._http_timeout_s),
+            )
+        return self._session
+
+    async def probe_server(self) -> dict[str, Any]:
+        session = await self._ensure_session()
+        try:
+            timeout = aiohttp.ClientTimeout(total=self._probe_timeout_ms / 1000.0)
+            async with session.get(f"{self._server_url}/api/v1/server/info", timeout=timeout) as resp:
+                return await resp.json()
+        except Exception as e:
+            logger.error(f"[iMessage] probe_server failed: {e}")
+            raise
+
+    async def get_handle(self) -> dict[str, Any]:
+        session = await self._ensure_session()
+        try:
+            async with session.get(f"{self._server_url}/api/v1/handle") as resp:
+                return await resp.json()
+        except Exception as e:
+            logger.error(f"[iMessage] get_handle failed: {e}")
+            raise
+
+    async def send_text_message(
+        self,
+        chat_guid: str,
+        content: str,
+        reply_to: str | None = None,
+        is_html: bool = False,
+        mentions: list[str] | None = None,
+        disable_notification: bool = False,
+    ) -> DeliveryResult:
+        payload: dict[str, Any] = {"chatGuid": chat_guid, "message": content}
+        if is_html:
+            payload["isHTML"] = True
+        if reply_to:
+            payload["replyTo"] = reply_to
+        if mentions:
+            payload["mentions"] = [{"type": "mention", "mention": m} for m in mentions]
+        if disable_notification:
+            payload["disableNotification"] = True
+
+        return await self._send_with_retry(
+            method="POST",
+            path="/api/v1/message/text",
+            json_payload=payload,
+        )
+
+    async def send_typing_indicator(self, chat_guid: str, display: bool = True) -> DeliveryResult:
+        return await self._send_with_retry(
+            method="POST",
+            path=f"/api/v1/chat/{chat_guid}/typing",
+            json_payload={"display": display},
+        )
+
+    async def send_read_receipt(self, chat_guid: str) -> DeliveryResult:
+        return await self._send_with_retry(
+            method="POST",
+            path=f"/api/v1/chat/{chat_guid}/readreceipt",
+            json_payload={},
+        )
+
+    async def download_attachment(self, attachment_path: str) -> bytes:
+        session = await self._ensure_session()
+
+        if attachment_path.startswith(("http://", "https://")):
+            parsed = urlparse(attachment_path)
+            if parsed.netloc not in urlparse(self._server_url).netloc:
+                raise BlueBubblesHTTPError(403, "Attachment URL origin mismatch")
+            url = attachment_path
+        else:
+            sanitized = attachment_path.lstrip("/")
+            url = f"{self._server_url}/{sanitized}"
+
+        try:
+            async with session.get(url) as resp:
+                if resp.status != 200:
+                    raise BlueBubblesHTTPError(resp.status, f"Failed to download attachment: {resp.status}")
+                return await resp.read()
+        except BlueBubblesHTTPError:
+            raise
+        except Exception as e:
+            logger.error(f"[iMessage] download_attachment failed: {e}")
+            raise
+
+    async def send_attachment(
+        self,
+        chat_guid: str,
+        file_path: str,
+        caption: str = "",
+        reply_to: str | None = None,
+    ) -> DeliveryResult:
+        import mimetypes
+
+        mime_type, _ = mimetypes.guess_type(file_path)
+
+        try:
+            loop = asyncio.get_running_loop()
+
+            def _read_file():
+                with open(file_path, "rb") as f:
+                    return f.read()
+
+            file_data = await loop.run_in_executor(None, _read_file)
+
+            form = aiohttp.FormData()
+            form.add_field(
+                "file",
+                file_data,
+                filename=os.path.basename(file_path),
+                content_type=mime_type or "application/octet-stream",
+            )
+            form.add_field("chatGuid", chat_guid)
+            if caption:
+                form.add_field("caption", caption)
+            if reply_to:
+                form.add_field("replyTo", reply_to)
+
+            return await self._send_with_retry(
+                method="POST",
+                path="/api/v1/message/attachment",
+                data_payload=form,
+            )
+        except Exception as e:
+            return DeliveryResult(success=False, error=str(e))
+
+    async def send_reaction(
+        self,
+        chat_guid: str,
+        message_guid: str,
+        tapback: int,
+    ) -> DeliveryResult:
+        payload = {"messageGuid": message_guid, "tapback": tapback}
+
+        return await self._send_with_retry(
+            method="POST",
+            path=f"/api/v1/message/{chat_guid}/tapback",
+            json_payload=payload,
+        )
+
+    async def edit_message(self, message_guid: str, new_content: str) -> DeliveryResult:
+        payload = {"message": new_content}
+        return await self._send_with_retry(
+            method="PUT",
+            path=f"/api/v1/message/{message_guid}",
+            json_payload=payload,
+        )
+
+    async def delete_message(self, chat_guid: str, message_guid: str) -> DeliveryResult:
+        payload = {"chatGuid": chat_guid, "messageGuid": message_guid}
+        return await self._send_with_retry(
+            method="POST",
+            path="/api/v1/message/delete",
+            json_payload=payload,
+        )
+
+    async def get_chats(self) -> list[dict[str, Any]]:
+        try:
+            session = await self._ensure_session()
+            async with session.get(f"{self._server_url}/api/v1/chat") as resp:
+                data = await resp.json()
+                return data.get("data", [])
+        except Exception:
+            logger.warning("[iMessage] Failed to fetch chats")
+            return []
+
+    async def get_chat_messages(self, chat_guid: str, limit: int = 50) -> list[dict[str, Any]]:
+        try:
+            session = await self._ensure_session()
+            params = {"limit": limit}
+            async with session.get(f"{self._server_url}/api/v1/chat/{chat_guid}/message", params=params) as resp:
+                data = await resp.json()
+                return data.get("data", [])
+        except Exception:
+            logger.warning(f"[iMessage] Failed to fetch messages for chat {chat_guid}")
+            return []
+
+    async def query_contacts(self, addresses: list[str]) -> list[dict[str, Any]]:
+        session = await self._ensure_session()
+        url = f"{self._server_url}/api/v1/contact/query"
+        async with session.post(url, json={"addresses": addresses}) as resp:
+            data = await resp.json()
+            return data.get("data", [])
+
+    async def unsend_message(self, message_guid: str) -> DeliveryResult:
+        return await self._send_with_retry(
+            method="POST",
+            path=f"/api/v1/message/{message_guid}/unsend",
+            json_payload={},
+        )
+
+    async def _send_with_retry(
+        self,
+        method: str,
+        path: str,
+        json_payload: dict[str, Any] | None = None,
+        data_payload: aiohttp.FormData | None = None,
+    ) -> DeliveryResult:
+        async def _attempt() -> DeliveryResult:
+            if self._last_auth_failure is not None:
+                elapsed = time.time() - self._last_auth_failure
+                wait_s = min(self._auth_backoff_s, 120.0)
+                if elapsed < wait_s:
+                    remaining = wait_s - elapsed
+                    logger.debug(f"[iMessage] Auth backoff: waiting {remaining:.1f}s")
+                    await asyncio.sleep(remaining)
+                else:
+                    self._last_auth_failure = None
+                    self._auth_backoff_s = 1.0
+
+            session = await self._ensure_session()
+            url = f"{self._server_url}{path}"
+            kwargs = {}
+            if json_payload is not None:
+                kwargs["json"] = json_payload
+            if data_payload is not None:
+                kwargs["data"] = data_payload
+
+            try:
+                async with session.request(method, url, **kwargs) as resp:
+                    data = await resp.json()
+                    if resp.status == 200:
+                        return DeliveryResult(
+                            success=True,
+                            message_id=data.get("data", {}).get("guid") or data.get("guid"),
+                        )
+                    if resp.status == 401:
+                        self._last_auth_failure = time.time()
+                        self._auth_backoff_s = min(self._auth_backoff_s * 2, 120.0)
+                        raise _RetryableError(f"Authentication failed (401): {data.get('message', '')}")
+                    if resp.status == 429:
+                        raise _RetryableError(f"Rate limited: {data.get('message', '')}")
+                    if resp.status >= 500:
+                        raise _RetryableError(f"Server error {resp.status}: {data.get('message', '')}")
+                    return DeliveryResult(success=False, error=data.get("message", "Unknown error"))
+            except _RetryableError:
+                raise
+            except Exception as e:
+                raise _RetryableError(str(e))
+
+        async def _breaker_attempt() -> DeliveryResult:
+            return await self._breaker.call(lambda: _attempt())
+
+        base_delay = 1.0
+        for attempt in range(self._max_retries):
+            try:
+                return await _breaker_attempt()
+            except _RetryableError as e:
+                if attempt < self._max_retries - 1:
+                    delay = base_delay * (2**attempt)
+                    logger.warning(
+                        f"[iMessage] {method} {path} failed (attempt {attempt + 1}/{self._max_retries}), "
+                        f"retrying in {delay:.1f}s: {e}"
+                    )
+                    await asyncio.sleep(delay)
+                    continue
+                logger.error(f"[iMessage] {method} {path} failed after {self._max_retries} attempts: {e}")
+                return DeliveryResult(success=False, error=str(e))
+
+        return DeliveryResult(success=False, error=f"Failed after {self._max_retries} retries")
+
+    @property
+    def ws_url(self) -> str:
+        url = self._server_url.replace("http://", "ws://").replace("https://", "wss://")
+        return f"{url}/ws"
+
+    @property
+    def password(self) -> str:
+        return self._password
+
+
+class _RetryableError(Exception):
+    pass

backend/package/yuxi/channels/adapters/imessage/commands.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def parse_command(text: str) -> dict[str, Any] | None:
+    """解析控制命令。
+
+    支持的格式：/cmd <action> [args...]
+
+    返回 {"action": str, "args": list[str]} 或 None。
+    """
+    if not text.startswith("/"):
+        return None
+
+    parts = text[1:].strip().split()
+    if not parts:
+        return None
+
+    action = parts[0].lower()
+    args = parts[1:] if len(parts) > 1 else []
+    return {"action": action, "args": args}
+
+
+def is_authorized_for_commands(sender_handle: str, allow_list: list[str]) -> bool:
+    """检查发送者是否有权限执行控制命令。"""
+    if "*" in allow_list:
+        return True
+    sender_clean = sender_handle.strip().replace(" ", "").lstrip("+")
+    for entry in allow_list:
+        entry_clean = entry.strip().replace(" ", "").lstrip("+")
+        if entry_clean == sender_clean:
+            return True
+    return False
+
+
+AVAILABLE_COMMANDS = {
+    "status": {"description": "Show adapter status", "requires_owner": False},
+    "health": {"description": "Show health check result", "requires_owner": False},
+    "pairing": {"description": "Show pairing status", "requires_owner": False},
+    "approve": {"description": "Approve a pending approval request", "requires_owner": True},
+    "reject": {"description": "Reject a pending approval request", "requires_owner": True},
+    "allowlist": {"description": "Manage allowlist entries", "requires_owner": True},
+    "help": {"description": "Show this help message", "requires_owner": False},
+}
+
+
+def get_command_help(include_owner: bool = False) -> str:
+    lines = ["Available commands:"]
+    for name, info in AVAILABLE_COMMANDS.items():
+        if info["requires_owner"] and not include_owner:
+            continue
+        lines.append(f"  /{name} — {info['description']}")
+    return "\n".join(lines)

backend/package/yuxi/channels/adapters/imessage/config_schema.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+from enum import StrEnum
+
+from pydantic import BaseModel, Field, field_validator
+
+
+class DmPolicyEnum(StrEnum):
+    PAIRING = "pairing"
+    ALLOWLIST = "allowlist"
+    OPEN = "open"
+    DISABLED = "disabled"
+
+
+class GroupPolicyEnum(StrEnum):
+    OPEN = "open"
+    ALLOWLIST = "allowlist"
+    DISABLED = "disabled"
+
+
+class IMessageConnectionConfig(BaseModel):
+    server_url: str = "http://localhost:1234"
+    password: str = ""
+    probe_timeout_ms: int = Field(default=10000, ge=1000, le=60000)
+    max_retries: int = Field(default=3, ge=1, le=10)
+    http_timeout_s: float = Field(default=30.0, ge=1.0, le=300.0)
+
+
+class IMessageSecurityConfigSchema(BaseModel):
+    dm_policy: DmPolicyEnum = DmPolicyEnum.PAIRING
+    group_policy: GroupPolicyEnum = GroupPolicyEnum.ALLOWLIST
+    allow_from: list[str] = Field(default_factory=list)
+    group_allow_from: list[str] = Field(default_factory=list)
+    require_mention: bool = False
+
+    @field_validator("dm_policy", mode="before")
+    @classmethod
+    def coerce_dm_policy(cls, v):
+        if isinstance(v, str):
+            try:
+                return DmPolicyEnum(v)
+            except ValueError:
+                return DmPolicyEnum.PAIRING
+        return v
+
+    @field_validator("group_policy", mode="before")
+    @classmethod
+    def coerce_group_policy(cls, v):
+        if isinstance(v, str):
+            try:
+                return GroupPolicyEnum(v)
+            except ValueError:
+                return GroupPolicyEnum.ALLOWLIST
+        return v
+
+
+class GroupOverrideConfig(BaseModel):
+    enabled: bool | None = None
+    require_mention: bool | None = None
+    tools: list[str] | None = None
+
+
+class IMessageAccountConfig(BaseModel):
+    account_id: str = "default"
+    name: str = ""
+    connection: IMessageConnectionConfig = Field(default_factory=IMessageConnectionConfig)
+    security: IMessageSecurityConfigSchema = Field(default_factory=IMessageSecurityConfigSchema)
+    enabled: bool = True
+    default_to: str = ""
+
+
+class IMessageFullConfig(BaseModel):
+    name: str = "iMessage"
+    connection: IMessageConnectionConfig = Field(default_factory=IMessageConnectionConfig)
+    security: IMessageSecurityConfigSchema = Field(default_factory=IMessageSecurityConfigSchema)
+
+    config_writes: bool = True
+    block_streaming: bool = False
+    text_chunk_limit: int = Field(default=4096, ge=100, le=16000)
+    media_max_mb: int = Field(default=100, ge=1, le=1000)
+    history_limit: int = Field(default=20, ge=1, le=200)
+
+    default_account: str = ""
+    default_to: str = ""
+    accounts: dict[str, IMessageAccountConfig] = Field(default_factory=dict)
+
+    include_attachments: bool = False
+    attachment_roots: list[str] = Field(default_factory=list)
+    remote_attachment_roots: list[str] = Field(default_factory=list)
+
+    groups: dict[str, GroupOverrideConfig] = Field(default_factory=dict)
+
+    ws_reconnect_initial_delay: float = Field(default=5.0, ge=0.5, le=120.0)
+    ws_reconnect_max_delay: float = Field(default=60.0, ge=1.0, le=600.0)
+
+    loop_rate_limit: int = Field(default=5, ge=0, le=50)
+    loop_rate_window_s: float = Field(default=60.0, ge=1.0, le=600.0)
+    loop_cooldown_s: float = Field(default=120.0, ge=1.0, le=3600.0)
+
+    @field_validator("accounts", mode="before")
+    @classmethod
+    def coerce_accounts(cls, v):
+        if v is None:
+            return {}
+        if isinstance(v, list):
+            return {a.account_id: a for a in v}
+        return v
+
+
+def validate_config(config: dict) -> IMessageFullConfig:
+    """校验并标准化 iMessage 配置。
+
+    Raises:
+        ValidationError: 配置不符合 Schema 时抛出。
+    """
+    return IMessageFullConfig(**config)

backend/package/yuxi/channels/adapters/imessage/contact_resolver.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+
+async def resolve_contact(bridge, address: str) -> dict[str, Any]:
+    try:
+        contacts = await bridge.query_contacts([address])
+        if contacts:
+            contact = contacts[0]
+            return {
+                "address": contact.get("address", address),
+                "display_name": contact.get("displayName", ""),
+                "first_name": contact.get("firstName", ""),
+                "last_name": contact.get("lastName", ""),
+            }
+        return {"address": address, "display_name": ""}
+    except Exception as e:
+        logger.warning(f"[iMessage] Contact resolution failed for {address}: {e}")
+        return {"address": address, "display_name": ""}
+
+
+async def resolve_contacts_batch(bridge, addresses: list[str]) -> dict[str, dict[str, Any]]:
+    if not addresses:
+        return {}
+
+    try:
+        contacts = await bridge.query_contacts(addresses)
+        resolved: dict[str, dict[str, Any]] = {}
+        for contact in contacts:
+            addr = contact.get("address", "")
+            resolved[addr] = {
+                "address": addr,
+                "display_name": contact.get("displayName", ""),
+                "first_name": contact.get("firstName", ""),
+                "last_name": contact.get("lastName", ""),
+            }
+        for addr in addresses:
+            if addr not in resolved:
+                resolved[addr] = {"address": addr, "display_name": ""}
+        return resolved
+    except Exception as e:
+        logger.warning(f"[iMessage] Batch contact resolution failed: {e}")
+        return {addr: {"address": addr, "display_name": ""} for addr in addresses}

backend/package/yuxi/channels/adapters/imessage/conversation_bindings.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+
+@dataclass
+class ConversationBinding:
+    channel_chat_id: str
+    agent_id: str = "main"
+    label: str = ""
+    created_at: float = field(default_factory=time.time)
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+class ConversationBindingManager:
+    def __init__(self, storage_dir: str | None = None, account_id: str = "default"):
+        self._bindings: dict[str, ConversationBinding] = {}
+        self._storage_dir = storage_dir or str(Path.home() / ".yuxi" / "imessage")
+        self._account_id = account_id
+
+    def create(
+        self, channel_chat_id: str, agent_id: str = "main", label: str = "", metadata: dict[str, Any] | None = None
+    ) -> ConversationBinding:
+        binding = ConversationBinding(
+            channel_chat_id=channel_chat_id,
+            agent_id=agent_id,
+            label=label,
+            metadata=metadata or {},
+        )
+        self._bindings[channel_chat_id] = binding
+        logger.info(f"[iMessage/Bindings] Created binding: {channel_chat_id} -> agent:{agent_id}")
+        return binding
+
+    def get(self, channel_chat_id: str) -> ConversationBinding | None:
+        return self._bindings.get(channel_chat_id)
+
+    def list_all(self) -> list[ConversationBinding]:
+        return list(self._bindings.values())
+
+    def delete(self, channel_chat_id: str) -> bool:
+        if channel_chat_id in self._bindings:
+            del self._bindings[channel_chat_id]
+            logger.info(f"[iMessage/Bindings] Deleted binding: {channel_chat_id}")
+            return True
+        return False
+
+    def get_agent_for_chat(self, channel_chat_id: str, default: str = "main") -> str:
+        binding = self._bindings.get(channel_chat_id)
+        return binding.agent_id if binding else default
+
+    async def save_bindings(self) -> None:
+        dir_path = Path(self._storage_dir) / self._account_id
+        dir_path.mkdir(parents=True, exist_ok=True)
+        data = {
+            cid: {
+                "agent_id": b.agent_id,
+                "label": b.label,
+                "created_at": b.created_at,
+                "metadata": b.metadata,
+            }
+            for cid, b in self._bindings.items()
+        }
+        file_path = dir_path / "conversation-bindings.json"
+        file_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    async def load_bindings(self) -> None:
+        file_path = Path(self._storage_dir) / self._account_id / "conversation-bindings.json"
+        if not file_path.exists():
+            return
+        try:
+            data = json.loads(file_path.read_text(encoding="utf-8"))
+            for cid, info in data.items():
+                if isinstance(info, dict):
+                    self._bindings[cid] = ConversationBinding(
+                        channel_chat_id=cid,
+                        agent_id=info.get("agent_id", "main"),
+                        label=info.get("label", ""),
+                        created_at=info.get("created_at", time.time()),
+                        metadata=info.get("metadata", {}),
+                    )
+        except (json.JSONDecodeError, OSError) as e:
+            logger.warning(f"[iMessage/Bindings] Failed to load bindings: {e}")

backend/package/yuxi/channels/adapters/imessage/conversation_route.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+
+@dataclass
+class ConversationRoute:
+    agent_id: str
+    channel_chat_id: str
+    source: str
+
+
+def resolve_agent_route(
+    channel_chat_id: str,
+    configured_bindings: dict[str, str] | None = None,
+    runtime_bindings: dict[str, str] | None = None,
+    default_agent: str = "main",
+) -> ConversationRoute:
+    if runtime_bindings and channel_chat_id in runtime_bindings:
+        return ConversationRoute(
+            agent_id=runtime_bindings[channel_chat_id],
+            channel_chat_id=channel_chat_id,
+            source="runtime_binding",
+        )
+
+    if configured_bindings and channel_chat_id in configured_bindings:
+        return ConversationRoute(
+            agent_id=configured_bindings[channel_chat_id],
+            channel_chat_id=channel_chat_id,
+            source="configured_binding",
+        )
+
+    return ConversationRoute(
+        agent_id=default_agent,
+        channel_chat_id=channel_chat_id,
+        source="default",
+    )
+
+
+def resolve_configured_binding(
+    channel_chat_id: str,
+    bindings: dict[str, str] | None = None,
+) -> str | None:
+    if bindings and channel_chat_id in bindings:
+        return bindings[channel_chat_id]
+    return None
+
+
+def resolve_runtime_binding(
+    channel_chat_id: str,
+    bindings: dict[str, str] | None = None,
+) -> str | None:
+    if bindings and channel_chat_id in bindings:
+        return bindings[channel_chat_id]
+    return None

backend/package/yuxi/channels/adapters/imessage/directory.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from typing import Any
+
+from yuxi.channels.adapters.imessage.bridge_client import BlueBubblesClient
+from yuxi.utils.logging_config import logger
+
+
+async def search_contacts(bridge: BlueBubblesClient, query: str, limit: int = 20) -> list[dict[str, Any]]:
+    """搜索联系人。"""
+    try:
+        contacts = await bridge.query_contacts([query])
+        return contacts[:limit]
+    except Exception as e:
+        logger.warning(f"[iMessage/Directory] Contact search failed: {e}")
+        return []
+
+
+async def list_recent_chats(bridge: BlueBubblesClient, limit: int = 30) -> list[dict[str, Any]]:
+    """获取最近聊天列表。"""
+    try:
+        chats = await bridge.get_chats()
+        chats.sort(key=lambda c: c.get("lastMessageDate", 0), reverse=True)
+        return chats[:limit]
+    except Exception as e:
+        logger.warning(f"[iMessage/Directory] Failed to list recent chats: {e}")
+        return []
+
+
+async def search_groups(bridge: BlueBubblesClient, name_query: str, limit: int = 20) -> list[dict[str, Any]]:
+    """搜索群组。"""
+    try:
+        chats = await bridge.get_chats()
+        groups = [c for c in chats if ";-;" in c.get("guid", "") or "@chat" in c.get("guid", "")]
+        if name_query:
+            name_lower = name_query.lower()
+            groups = [
+                g
+                for g in groups
+                if name_lower in g.get("displayName", "").lower() or name_lower in g.get("guid", "").lower()
+            ]
+        groups.sort(key=lambda c: c.get("lastMessageDate", 0), reverse=True)
+        return groups[:limit]
+    except Exception as e:
+        logger.warning(f"[iMessage/Directory] Group search failed: {e}")
+        return []
+
+
+async def get_chat_info(bridge: BlueBubblesClient, chat_guid: str) -> dict[str, Any]:
+    """获取聊天详细信息。"""
+    try:
+        messages = await bridge.get_chat_messages(chat_guid, limit=1)
+        chats = await bridge.get_chats()
+        for chat in chats:
+            if chat.get("guid") == chat_guid:
+                return {
+                    "guid": chat_guid,
+                    "display_name": chat.get("displayName", ""),
+                    "participants": chat.get("participants", []),
+                    "last_message": messages[0] if messages else None,
+                }
+        return {"guid": chat_guid, "display_name": "", "participants": []}
+    except Exception as e:
+        logger.warning(f"[iMessage/Directory] Failed to get chat info for {chat_guid}: {e}")
+        return {"guid": chat_guid, "display_name": ""}

backend/package/yuxi/channels/adapters/imessage/echo_cache.py

@@ -0,0 +1,146 @@
+from __future__ import annotations
+
+import hashlib
+import time
+from dataclasses import dataclass
+
+TEXT_TTL_S = 4.0
+TEXT_BACKED_BY_ID_TTL_S = 4.0
+MESSAGE_ID_TTL_S = 60.0
+SELF_CHAT_TTL_S = 10.0
+MAX_SELF_CHAT_ENTRIES = 512
+
+
+@dataclass
+class _SentEntry:
+    message_id: str
+    text: str
+    chat_guid: str
+    sent_at: float
+
+
+@dataclass
+class _SelfChatEntry:
+    text_hash: str
+    chat_guid: str
+    recorded_at: float
+
+
+class EchoCache:
+    def __init__(
+        self,
+        text_ttl_s: float = TEXT_TTL_S,
+        text_backed_by_id_ttl_s: float = TEXT_BACKED_BY_ID_TTL_S,
+        message_id_ttl_s: float = MESSAGE_ID_TTL_S,
+        self_chat_ttl_s: float = SELF_CHAT_TTL_S,
+        max_self_chat_entries: int = MAX_SELF_CHAT_ENTRIES,
+    ):
+        self._text_ttl = text_ttl_s
+        self._text_backed_by_id_ttl = text_backed_by_id_ttl_s
+        self._msg_id_ttl = message_id_ttl_s
+        self._self_chat_ttl = self_chat_ttl_s
+        self._max_self_chat_entries = max_self_chat_entries
+
+        self._by_message_id: dict[str, _SentEntry] = {}
+        self._by_text: dict[str, list[_SentEntry]] = {}
+        self._by_text_backed_by_id: dict[str, _SentEntry] = {}
+
+        self._self_chat_cache: list[_SelfChatEntry] = []
+        self._last_self_chat_cleanup = time.monotonic()
+
+    def record_sent(self, message_id: str, text: str, chat_guid: str) -> None:
+        now = time.monotonic()
+        entry = _SentEntry(message_id=message_id, text=text, chat_guid=chat_guid, sent_at=now)
+        self._by_message_id[message_id] = entry
+
+        text_key = _text_key(text, chat_guid)
+        if text_key not in self._by_text:
+            self._by_text[text_key] = []
+        self._by_text[text_key].append(entry)
+
+        if message_id:
+            self._by_text_backed_by_id[message_id] = entry
+
+        self._cleanup()
+
+    def is_echo(self, message_id: str | None = None, text: str | None = None, chat_guid: str = "") -> bool:
+        self._cleanup()
+
+        if message_id and message_id in self._by_message_id:
+            return True
+
+        if message_id and message_id in self._by_text_backed_by_id:
+            return True
+
+        if text:
+            text_key = _text_key(text, chat_guid)
+            entries = self._by_text.get(text_key, [])
+            if entries:
+                return True
+
+        return False
+
+    def is_self_chat_echo(
+        self,
+        sender_handle: str,
+        self_handle: str,
+        chat_guid: str,
+    ) -> bool:
+        if not self_handle:
+            return False
+        sender_clean = sender_handle.strip().replace(" ", "").lstrip("+")
+        self_clean = self_handle.strip().replace(" ", "").lstrip("+")
+        return sender_clean == self_clean
+
+    def record_self_chat_text(self, text: str, chat_guid: str) -> None:
+        now = time.monotonic()
+        text_hash = hashlib.sha256(f"{chat_guid}:{text.strip()[:200]}".encode()).hexdigest()
+        entry = _SelfChatEntry(text_hash=text_hash, chat_guid=chat_guid, recorded_at=now)
+        self._self_chat_cache.append(entry)
+        self._cleanup_self_chat()
+        if len(self._self_chat_cache) > self._max_self_chat_entries:
+            self._self_chat_cache = self._self_chat_cache[-self._max_self_chat_entries :]
+
+    def is_in_self_chat_cache(self, text: str, chat_guid: str) -> bool:
+        self._cleanup_self_chat()
+        text_hash = hashlib.sha256(f"{chat_guid}:{text.strip()[:200]}".encode()).hexdigest()
+        for entry in self._self_chat_cache:
+            if entry.text_hash == text_hash and entry.chat_guid == chat_guid:
+                return True
+        return False
+
+    def _cleanup_self_chat(self) -> None:
+        now = time.monotonic()
+        self._self_chat_cache = [
+            entry for entry in self._self_chat_cache if now - entry.recorded_at <= self._self_chat_ttl
+        ]
+
+    def _cleanup(self) -> None:
+        now = time.monotonic()
+
+        expired_msg_ids = [mid for mid, entry in self._by_message_id.items() if now - entry.sent_at > self._msg_id_ttl]
+        for mid in expired_msg_ids:
+            del self._by_message_id[mid]
+
+        expired_text_backed = [
+            mid
+            for mid, entry in self._by_text_backed_by_id.items()
+            if now - entry.sent_at > self._text_backed_by_id_ttl
+        ]
+        for mid in expired_text_backed:
+            del self._by_text_backed_by_id[mid]
+
+        for text_key in list(self._by_text):
+            self._by_text[text_key] = [
+                entry for entry in self._by_text[text_key] if now - entry.sent_at <= self._text_ttl
+            ]
+            if not self._by_text[text_key]:
+                del self._by_text[text_key]
+
+        if now - self._last_self_chat_cleanup > self._self_chat_ttl:
+            self._cleanup_self_chat()
+            self._last_self_chat_cleanup = now
+
+
+def _text_key(text: str, chat_guid: str) -> str:
+    return f"{chat_guid}:{text.strip()[:200]}"

backend/package/yuxi/channels/adapters/imessage/envelope.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from yuxi.channels.models import ChannelIdentity
+
+
+def format_inbound_envelope(
+    identity: ChannelIdentity,
+    sender_name: str = "",
+    chat_name: str = "",
+    reply_context: dict | None = None,
+) -> str:
+    """生成标准入站信封格式。
+
+    格式示例：
+        [From: Alice (alice@icloud.com)] @Family Chat
+    """
+    parts: list[str] = []
+
+    sender_display = sender_name or identity.channel_user_id
+    parts.append(f"[From: {sender_display}")
+
+    if identity.channel_user_id != sender_display:
+        parts.append(f" ({identity.channel_user_id})")
+
+    parts.append("]")
+
+    if chat_name:
+        parts.append(f" @{chat_name}")
+
+    if reply_context:
+        reply_to_id = reply_context.get("reply_to_message_id")
+        reply_to_text = reply_context.get("reply_to_text")
+        if reply_to_text:
+            preview = reply_to_text[:80].replace("\n", " ")
+            parts.append(f"\n  > Replying to: {preview}")
+        elif reply_to_id:
+            parts.append(f"\n  > Replying to message id:{reply_to_id}")
+
+    return "".join(parts)

backend/package/yuxi/channels/adapters/imessage/exceptions.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from yuxi.channels.exceptions import ChannelException
+
+
+class IMessageError(ChannelException):
+    def __init__(self, message: str, retryable: bool = False, retry_after_ms: int = 0):
+        super().__init__(message, retryable=retryable, retry_after_ms=retry_after_ms)
+
+
+class BlueBubblesHTTPError(IMessageError):
+    def __init__(self, status_code: int, message: str = "", headers: dict | None = None):
+        self.status_code = status_code
+        self.headers = headers or {}
+        retryable = status_code >= 500 or status_code == 429
+        detail = f"HTTP {status_code}: {message}" if message else f"HTTP {status_code}"
+        super().__init__(detail, retryable=retryable)
+
+
+class BlueBubblesConnectionError(IMessageError):
+    def __init__(self, message: str = ""):
+        super().__init__(message or "BlueBubbles connection failed", retryable=True, retry_after_ms=5000)
+
+
+class BlueBubblesAuthenticationError(IMessageError):
+    def __init__(self, message: str = ""):
+        super().__init__(message or "BlueBubbles authentication failed", retryable=False)
+
+
+class TapbackError(IMessageError):
+    def __init__(self, message: str = ""):
+        super().__init__(message or "Tapback operation failed", retryable=False)

backend/package/yuxi/channels/adapters/imessage/formatter.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import re
+
+TABLE_ROW_RE = re.compile(r"^\|(.+)\|$")
+TABLE_SEP_RE = re.compile(r"^\|[\s\-:]+\|$")
+DIRECTIVE_RE = re.compile(r"<\s*/?directive[^>]*>", re.IGNORECASE)
+
+
+def convert_markdown_tables(text: str) -> str:
+    """将 Markdown 表格转换为可读的 bullets 格式。
+
+    iMessage 不支持 Markdown 表格渲染，此函数将表格行转为可读的列表。
+    """
+    lines = text.split("\n")
+    result: list[str] = []
+    i = 0
+    while i < len(lines):
+        line = lines[i]
+        if i + 2 < len(lines) and TABLE_ROW_RE.match(line) and TABLE_SEP_RE.match(lines[i + 1]):
+            header = line
+            sep = lines[i + 1]
+            body_rows: list[str] = []
+            j = i + 2
+            while j < len(lines) and TABLE_ROW_RE.match(lines[j]):
+                body_rows.append(lines[j])
+                j += 1
+
+            converted = _table_to_bullets(header, sep, body_rows)
+            result.extend(converted)
+            i = j
+            continue
+
+        result.append(line)
+        i += 1
+
+    return "\n".join(result)
+
+
+def _table_to_bullets(header: str, sep: str, body_rows: list[str]) -> list[str]:
+    header_cells = _parse_cells(header)
+    sep_cells = _parse_cells(sep)
+
+    if not header_cells:
+        return [header, sep] + body_rows
+
+    alignments = _detect_alignments(sep_cells, len(header_cells))
+    body_cells = [_parse_cells(row) for row in body_rows]
+
+    output: list[str] = []
+    for row_idx, cells in enumerate([header_cells] + body_cells):
+        if row_idx == 0:
+            output.append(f"• {', '.join(cells)}")
+        elif len(cells) <= 2:
+            output.append("  " + ": ".join(cells))
+        else:
+            parts: list[str] = []
+            for ci, (cell, hdr, align) in enumerate(zip(cells, header_cells, alignments)):
+                prefix = f"  {hdr}: " if ci == 0 or align != "skip" else f"  {hdr}: "
+                parts.append(f"{prefix}{cell}")
+            output.append("")
+            output.extend(parts)
+
+    return output
+
+
+def _parse_cells(row: str) -> list[str]:
+    return [cell.strip() for cell in row.strip("|").split("|")]
+
+
+def _detect_alignments(sep_cells: list[str], col_count: int) -> list[str]:
+    aligns: list[str] = []
+    for i in range(col_count):
+        if i < len(sep_cells):
+            cell = sep_cells[i]
+            left = cell.startswith(":")
+            right = cell.endswith(":")
+            if left and right:
+                aligns.append("center")
+            elif right:
+                aligns.append("right")
+            else:
+                aligns.append("left")
+        else:
+            aligns.append("left")
+    return aligns
+
+
+def sanitize_imessage_text(text: str) -> str:
+    """净化出站文本，移除 iMessage 中不支持的格式。
+
+    - 去除 \\r 字符（换行标准化）
+    - 去除 <directive> 内联指令标签
+    - 去除长度前缀损坏（如 \"123\\nThis is message\"）
+    - 过滤控制字符
+    - 合并多余空白行
+    """
+    text = text.replace("\r", "")
+
+    text = DIRECTIVE_RE.sub("", text)
+
+    text = _strip_length_prefix(text)
+
+    text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]", "", text)
+
+    text = re.sub(r"\n{3,}", "\n\n", text)
+    text = text.strip()
+
+    return text
+
+
+def _strip_length_prefix(text: str) -> str:
+    """去除 iMessage 长度前缀损坏格式。
+
+    BlueBubbles 可能在某些消息前附加长度前缀如 \"42\\nHello world\"，
+    去除数字换行前缀。
+    """
+    match = re.match(r"^(\d+)\n", text)
+    if match:
+        prefix_len = int(match.group(1))
+        remaining = text[match.end() :]
+        if abs(len(remaining) - prefix_len) <= 5:
+            return remaining
+    return text

backend/package/yuxi/channels/adapters/imessage/media_ai.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+
+async def describe_image(image_data: bytes, vision_api_url: str = "", vision_api_key: str = "") -> str:
+    """通过外部 Vision API 描述图片内容。
+
+    如果未配置 API URL，返回基本元数据描述。
+    """
+    if not vision_api_url:
+        return f"Image attachment ({len(image_data)} bytes)"
+
+    try:
+        import base64
+
+        import aiohttp
+
+        encoded = base64.b64encode(image_data).decode("utf-8")
+        payload = {
+            "model": "gpt-4o-mini",
+            "messages": [
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": "Describe this image briefly in one sentence."},
+                        {
+                            "type": "image_url",
+                            "image_url": {"url": f"data:image/jpeg;base64,{encoded}"},
+                        },
+                    ],
+                }
+            ],
+            "max_tokens": 100,
+        }
+        headers = {"Authorization": f"Bearer {vision_api_key}"}
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                vision_api_url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=15)
+            ) as resp:
+                if resp.status == 200:
+                    data = await resp.json()
+                    return data["choices"][0]["message"]["content"]
+                return f"Image attachment ({len(image_data)} bytes)"
+    except Exception as e:
+        logger.warning(f"[iMessage/AI] Vision API failed: {e}")
+        return f"Image attachment ({len(image_data)} bytes)"
+
+
+async def transcribe_audio(audio_data: bytes, transcription_api_url: str = "", transcription_api_key: str = "") -> str:
+    """通过外部 API 转录音频内容。"""
+    if not transcription_api_url:
+        return f"Audio attachment ({len(audio_data)} bytes)"
+
+    try:
+        import aiohttp
+        from aiohttp import FormData
+
+        form = FormData()
+        form.add_field("file", audio_data, filename="audio.caf", content_type="audio/x-caf")
+        form.add_field("model", "whisper-1")
+
+        headers = {"Authorization": f"Bearer {transcription_api_key}"}
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                transcription_api_url, data=form, headers=headers, timeout=aiohttp.ClientTimeout(total=30)
+            ) as resp:
+                if resp.status == 200:
+                    data = await resp.json()
+                    return data.get("text", "")
+                return f"Audio attachment ({len(audio_data)} bytes)"
+    except Exception as e:
+        logger.warning(f"[iMessage/AI] Transcription API failed: {e}")
+        return f"Audio attachment ({len(audio_data)} bytes)"
+
+
+def is_vision_configured(config: dict[str, Any]) -> bool:
+    return bool(config.get("vision_api_url") and config.get("vision_api_key"))
+
+
+def is_transcription_configured(config: dict[str, Any]) -> bool:
+    return bool(config.get("transcription_api_url") and config.get("transcription_api_key"))

backend/package/yuxi/channels/adapters/imessage/media_security.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from yuxi.utils.logging_config import logger
+
+DEFAULT_IMESSAGE_ATTACHMENT_ROOTS = [
+    "/Users/*/Library/Messages/Attachments",
+]
+
+
+def resolve_attachment_roots(
+    configured_roots: list[str] | None = None,
+) -> list[str]:
+    if configured_roots:
+        return configured_roots
+    expanded: list[str] = []
+    for pattern in DEFAULT_IMESSAGE_ATTACHMENT_ROOTS:
+        import glob as _glob_module
+
+        matches = _glob_module.glob(pattern)
+        if matches:
+            expanded.extend(matches)
+    return expanded or DEFAULT_IMESSAGE_ATTACHMENT_ROOTS
+
+
+def validate_attachment_path(
+    file_path: str,
+    allowed_roots: list[str] | None = None,
+) -> bool:
+    """校验附件路径是否在允许的根路径内。
+
+    阻止路径遍历攻击（如 ../../etc/passwd）。
+    """
+    if not file_path:
+        return False
+
+    real_path = os.path.realpath(file_path) if os.path.exists(file_path) else os.path.abspath(file_path)
+
+    roots = allowed_roots or []
+    if not roots:
+        return True
+
+    for root in roots:
+        root_real = os.path.realpath(root) if os.path.exists(root) else os.path.abspath(root)
+        try:
+            Path(real_path).relative_to(root_real)
+            return True
+        except ValueError:
+            continue
+
+    logger.warning(f"[iMessage/Security] Attachment path rejected: {file_path} not in allowed roots")
+    return False
+
+
+def validate_remote_attachment_url(
+    url: str,
+    allowed_roots: list[str] | None = None,
+    server_url: str = "",
+) -> bool:
+    """校验远程附件 URL 是否安全。"""
+    if not url:
+        return False
+
+    if allowed_roots:
+        for root in allowed_roots:
+            if url.startswith(root):
+                return True
+        return False
+
+    if server_url and url.startswith(server_url):
+        return True
+
+    return False
+
+
+def sanitize_attachment_filename(filename: str) -> str:
+    """净化附件文件名，移除路径分隔符。"""
+    if not filename:
+        return "attachment"
+    basename = os.path.basename(filename)
+    safe = "".join(c for c in basename if c.isalnum() or c in "._-() ")
+    return safe or "attachment"

backend/package/yuxi/channels/adapters/imessage/monitor.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import random
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+import aiohttp
+
+from yuxi.utils.logging_config import logger
+
+
+_MAX_ERROR_MSG_LEN = 200
+
+
+class IMessageMonitor:
+    def __init__(self, config: dict[str, Any], bridge):
+        self._config = config
+        self._bridge = bridge
+        self._message_handler: Callable[[dict[str, Any]], Awaitable[None]] | None = None
+        self._task: asyncio.Task | None = None
+        self._reconnect_delay = 5.0
+        self._initial_reconnect_delay = config.get("ws_reconnect_initial_delay", 5.0)
+        self._max_reconnect_delay = config.get("ws_reconnect_max_delay", 60.0)
+        self._max_subscribe_attempts = config.get(
+            "watchSubscribeMaxAttempts", config.get("watch_subscribe_max_attempts", 3)
+        )
+        self._subscribe_retry_delay_ms = config.get(
+            "watchSubscribeRetryDelayMs", config.get("watch_subscribe_retry_delay_ms", 1000)
+        )
+        self._subscribe_attempts = 0
+
+    def on_message(self, handler: Callable[[dict[str, Any]], Awaitable[None]]) -> None:
+        self._message_handler = handler
+
+    async def start(self) -> None:
+        if self._task and not self._task.done():
+            return
+        self._task = asyncio.create_task(self._ws_loop())
+
+    async def stop(self) -> None:
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            self._task = None
+
+    async def _ws_loop(self) -> None:
+        while True:
+            try:
+                self._subscribe_attempts = 0
+                await self._connect_ws()
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                self._subscribe_attempts += 1
+                if self._subscribe_attempts >= self._max_subscribe_attempts:
+                    logger.error(
+                        f"[iMessage] Max subscribe attempts ({self._max_subscribe_attempts}) reached, stopping monitor"
+                    )
+                    break
+                delay = self._reconnect_delay * (0.5 + random.random())
+                err_msg = str(e)[:_MAX_ERROR_MSG_LEN]
+                logger.error(
+                    f"[iMessage] WebSocket error (attempt {self._subscribe_attempts}), "
+                    f"reconnecting in {delay:.1f}s: {err_msg}"
+                )
+                await asyncio.sleep(delay)
+                self._reconnect_delay = min(self._reconnect_delay * 1.5, self._max_reconnect_delay)
+
+    async def _connect_ws(self) -> None:
+        url = self._bridge.ws_url
+        logger.info(f"[iMessage] Connecting to BlueBubbles WebSocket: {url}")
+
+        async with aiohttp.ClientSession() as session:
+            headers = {"X-BlueBubbles-Password": self._bridge.password}
+            async with session.ws_connect(url, headers=headers) as ws:
+                self._reconnect_delay = self._initial_reconnect_delay
+                logger.info("[iMessage] BlueBubbles WebSocket connected")
+
+                async for msg in ws:
+                    if msg.type == aiohttp.WSMsgType.TEXT:
+                        try:
+                            data = json.loads(msg.data)
+                            event_type = data.get("type", "")
+                            logger.debug(f"[iMessage] WS event: {event_type}")
+
+                            if event_type == "error":
+                                err_data = data.get("data", {})
+                                err_str = str(err_data)
+                                if isinstance(err_data, dict):
+                                    err_str = err_data.get("message", err_data.get("error", str(err_data)))
+                                logger.error(f"[iMessage] BlueBubbles error: {err_str[:_MAX_ERROR_MSG_LEN]}")
+                            elif self._message_handler:
+                                await self._message_handler(data)
+                        except json.JSONDecodeError:
+                            logger.warning(f"[iMessage] WS non-JSON message: {msg.data[:200]}")
+                    elif msg.type == aiohttp.WSMsgType.ERROR:
+                        logger.error(f"[iMessage] WS error: {ws.exception()}")
+                        break
+                    elif msg.type == aiohttp.WSMsgType.CLOSED:
+                        logger.info("[iMessage] BlueBubbles WebSocket closed")
+                        break

backend/package/yuxi/channels/adapters/imessage/normalize.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_E164_RE = re.compile(r"^\+[1-9]\d{1,14}$")
+_NON_DIGIT_RE = re.compile(r"[^\d+]")
+_PHONE_CLEAN_RE = re.compile(r"[\s\-\(\)\.]")
+
+
+def normalize_e164(handle: str, default_country_code: str = "") -> str:
+    """将电话号码标准化为 E.164 格式。
+
+    去除空格、括号、连字符，补全国际区号。
+    对已经是国际格式的号码保留 '+' 前缀。
+    """
+    clean = _PHONE_CLEAN_RE.sub("", handle.strip())
+
+    clean = _strip_service_prefixes(clean)
+
+    if not clean:
+        return handle
+
+    if clean.startswith("+"):
+        digits = clean[1:]
+        if digits.isdigit() and len(digits) <= 15:
+            return clean
+        return handle
+
+    if clean.startswith("00"):
+        digits = clean[2:]
+        if digits.isdigit() and len(digits) <= 15:
+            return f"+{digits}"
+
+    if clean.startswith("011"):
+        digits = clean[3:]
+        if digits.isdigit() and len(digits) <= 15:
+            return f"+{digits}"
+
+    if clean.isdigit() and default_country_code:
+        return f"+{default_country_code}{clean}"
+
+    if clean.isdigit() and len(clean) <= 15:
+        return f"+{clean}"
+
+    return handle
+
+
+def _strip_service_prefixes(handle: str) -> str:
+    for prefix in ("imessage:", "sms:", "auto:"):
+        if handle.lower().startswith(prefix):
+            return handle[len(prefix) :]
+    return handle
+
+
+def is_e164(handle: str) -> bool:
+    return bool(_E164_RE.match(handle))
+
+
+def extract_digits(handle: str) -> str:
+    return _NON_DIGIT_RE.sub("", handle)
+
+
+def parse_forwarded(data: dict) -> dict[str, Any] | None:
+    forwarded = data.get("forwarded")
+    if not forwarded:
+        return None
+
+    result: dict[str, Any] = {"is_forwarded": True}
+    if isinstance(forwarded, dict):
+        result["forwarded_from"] = forwarded.get("handle") or forwarded.get("sender", "")
+        result["forwarded_date"] = forwarded.get("date") or forwarded.get("originalDate", "")
+    elif isinstance(forwarded, bool) and forwarded:
+        result["forwarded_from"] = data.get("forwardedFrom", "")
+    return result

backend/package/yuxi/channels/adapters/imessage/pairing.py

@@ -0,0 +1,203 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import secrets
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+PAIRING_CODE_CHARS = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
+PAIRING_CODE_LENGTH = 8
+PAIRING_VALIDITY_S = 3600
+PAIRING_RATE_LIMIT_S = 3600
+PAIRING_MAX_PENDING = 3
+
+
+@dataclass
+class PairingRequest:
+    code: str
+    sender_handle: str
+    channel_user_id: str
+    chat_guid: str
+    requested_at: float
+    expires_at: float
+    metadata: dict[str, Any] | None = None
+
+
+class IMessagePairingManager:
+    def __init__(self, storage_dir: str | None = None, account_id: str = "default", config_writes: bool = True):
+        self._pending: dict[str, PairingRequest] = {}
+        self._rate_limit: dict[str, float] = {}
+        self._approved: set[str] = set()
+        self._storage_dir = storage_dir or str(Path.home() / ".yuxi" / "imessage")
+        self._account_id = account_id
+        self._config_writes = config_writes
+
+    def _schedule_save(self) -> None:
+        if not self._config_writes:
+            return
+        try:
+            loop = asyncio.get_running_loop()
+            loop.create_task(self._save_all())
+        except RuntimeError:
+            pass
+
+    async def _save_all(self) -> None:
+        await self.save_allowlist()
+        await self.save_pending()
+
+    def generate_code(self) -> str:
+        return "".join(secrets.choice(PAIRING_CODE_CHARS) for _ in range(PAIRING_CODE_LENGTH))
+
+    def request_pairing(
+        self,
+        sender_handle: str,
+        channel_user_id: str,
+        chat_guid: str,
+        metadata: dict[str, Any] | None = None,
+    ) -> str | None:
+        if self.is_approved(channel_user_id):
+            return None
+
+        now = time.monotonic()
+        last_request = self._rate_limit.get(channel_user_id, 0)
+        if now - last_request < PAIRING_RATE_LIMIT_S:
+            logger.info(f"[iMessage/Pairing] Rate limited for {channel_user_id}")
+            return None
+
+        if len(self._pending) >= PAIRING_MAX_PENDING:
+            self._cleanup_expired()
+            if len(self._pending) >= PAIRING_MAX_PENDING:
+                logger.warning("[iMessage/Pairing] Max pending requests reached")
+                return None
+
+        code = self.generate_code()
+        self._pending[code] = PairingRequest(
+            code=code,
+            sender_handle=sender_handle,
+            channel_user_id=channel_user_id,
+            chat_guid=chat_guid,
+            requested_at=time.time(),
+            expires_at=time.time() + PAIRING_VALIDITY_S,
+            metadata=metadata,
+        )
+        self._rate_limit[channel_user_id] = now
+        logger.info(f"[iMessage/Pairing] New pairing request: code={code}, sender={sender_handle}")
+        self._schedule_save()
+        return code
+
+    def approve(self, code: str) -> bool:
+        self._cleanup_expired()
+        req = self._pending.pop(code, None)
+        if req is None:
+            return False
+        self._approved.add(req.channel_user_id)
+        logger.info(f"[iMessage/Pairing] Approved: {req.channel_user_id}")
+        self._schedule_save()
+        return True
+
+    def reject(self, code: str) -> bool:
+        req = self._pending.pop(code, None)
+        if req is None:
+            return False
+        logger.info(f"[iMessage/Pairing] Rejected: {req.channel_user_id}")
+        self._schedule_save()
+        return True
+
+    def list_pending(self) -> list[dict[str, Any]]:
+        self._cleanup_expired()
+        return [
+            {
+                "code": r.code,
+                "sender_handle": r.sender_handle,
+                "channel_user_id": r.channel_user_id,
+                "chat_guid": r.chat_guid,
+                "requested_at": r.requested_at,
+                "expires_at": r.expires_at,
+            }
+            for r in self._pending.values()
+        ]
+
+    def is_approved(self, channel_user_id: str) -> bool:
+        return channel_user_id in self._approved
+
+    def get_paired_handles(self) -> list[str]:
+        return sorted(self._approved)
+
+    def is_paired(self, channel_user_id: str) -> bool:
+        return channel_user_id in self._approved
+
+    def revoke_approval(self, channel_user_id: str) -> bool:
+        if channel_user_id in self._approved:
+            self._approved.discard(channel_user_id)
+            self._schedule_save()
+            return True
+        return False
+
+    def _cleanup_expired(self) -> None:
+        now = time.time()
+        expired = [code for code, req in self._pending.items() if req.expires_at < now]
+        for code in expired:
+            self._pending.pop(code, None)
+
+    async def save_allowlist(self) -> None:
+        dir_path = Path(self._storage_dir) / self._account_id
+        dir_path.mkdir(parents=True, exist_ok=True)
+        file_path = dir_path / "paired_handles.json"
+        file_path.write_text(
+            json.dumps(sorted(self._approved), ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+
+    async def load_allowlist(self) -> list[str]:
+        file_path = Path(self._storage_dir) / self._account_id / "paired_handles.json"
+        if not file_path.exists():
+            return []
+        try:
+            data = json.loads(file_path.read_text(encoding="utf-8"))
+            if isinstance(data, list):
+                self._approved.update(data)
+                return data
+        except (json.JSONDecodeError, OSError) as e:
+            logger.warning(f"[iMessage/Pairing] Failed to load paired handles: {e}")
+        return []
+
+    async def save_pending(self) -> None:
+        dir_path = Path(self._storage_dir) / self._account_id
+        dir_path.mkdir(parents=True, exist_ok=True)
+        pending_data = {
+            code: {
+                "sender_handle": r.sender_handle,
+                "channel_user_id": r.channel_user_id,
+                "chat_guid": r.chat_guid,
+                "requested_at": r.requested_at,
+                "expires_at": r.expires_at,
+            }
+            for code, r in self._pending.items()
+        }
+        file_path = dir_path / "pairing-pending.json"
+        file_path.write_text(json.dumps(pending_data, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    async def load_pending(self) -> None:
+        file_path = Path(self._storage_dir) / self._account_id / "pairing-pending.json"
+        if not file_path.exists():
+            return
+        try:
+            data = json.loads(file_path.read_text(encoding="utf-8"))
+            for code, info in data.items():
+                if info.get("expires_at", 0) < time.time():
+                    continue
+                self._pending[code] = PairingRequest(
+                    code=code,
+                    sender_handle=info.get("sender_handle", ""),
+                    channel_user_id=info.get("channel_user_id", ""),
+                    chat_guid=info.get("chat_guid", ""),
+                    requested_at=info.get("requested_at", time.time()),
+                    expires_at=info.get("expires_at", time.time() + PAIRING_VALIDITY_S),
+                )
+        except (json.JSONDecodeError, OSError) as e:
+            logger.warning(f"[iMessage/Pairing] Failed to load pending: {e}")

backend/package/yuxi/channels/adapters/imessage/polls.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+
+@dataclass
+class PollOption:
+    index: int
+    label: str
+    vote_count: int = 0
+
+
+@dataclass
+class Poll:
+    poll_id: str
+    chat_guid: str
+    question: str
+    options: list[PollOption]
+    created_at: float = field(default_factory=time.time)
+    closed: bool = False
+    expires_at: float = 0.0
+    voter_map: dict[str, int] = field(default_factory=dict)
+
+    def vote(self, user_id: str, option_index: int) -> bool:
+        if self.closed:
+            return False
+        if option_index < 0 or option_index >= len(self.options):
+            return False
+
+        prev_vote = self.voter_map.get(user_id)
+        if prev_vote is not None and 0 <= prev_vote < len(self.options):
+            self.options[prev_vote].vote_count -= 1
+
+        self.options[option_index].vote_count += 1
+        self.voter_map[user_id] = option_index
+        return True
+
+    def results_text(self) -> str:
+        lines = [f"Poll: {self.question}"]
+        total = sum(o.vote_count for o in self.options)
+        for opt in self.options:
+            bar = _bar(opt.vote_count, total) if total > 0 else ""
+            lines.append(f"  {opt.index + 1}. {opt.label} — {opt.vote_count} vote(s) {bar}")
+        lines.append(f"Total: {total} vote(s)")
+        if self.closed:
+            lines.append("(Poll closed)")
+        return "\n".join(lines)
+
+
+def _bar(count: int, total: int, width: int = 10) -> str:
+    if total == 0:
+        return ""
+    filled = round(count / total * width)
+    return "[" + "■" * filled + "□" * (width - filled) + "]"
+
+
+class IMessagePollManager:
+    def __init__(self, adapter: Any):
+        self._adapter = adapter
+        self._polls: dict[str, Poll] = {}
+        self._counter = 0
+
+    async def create_poll(
+        self,
+        chat_guid: str,
+        question: str,
+        option_labels: list[str],
+        expires_in_s: float = 300.0,
+    ) -> Poll:
+        self._counter += 1
+        poll_id = f"poll_{self._counter}_{int(time.time())}"
+
+        options = [PollOption(index=i, label=label) for i, label in enumerate(option_labels)]
+        poll = Poll(
+            poll_id=poll_id,
+            chat_guid=chat_guid,
+            question=question,
+            options=options,
+            expires_at=time.time() + expires_in_s,
+        )
+        self._polls[poll_id] = poll
+
+        poll_text = _format_poll_text(poll)
+        result = await self._adapter._get_client().send_text_message(
+            chat_guid=chat_guid,
+            content=poll_text,
+        )
+        if result.success and result.message_id:
+            logger.info(f"[iMessage/Poll] Created poll {poll_id} in {chat_guid}")
+        else:
+            logger.error(f"[iMessage/Poll] Failed to send poll {poll_id}: {result.error}")
+
+        if expires_in_s > 0:
+            asyncio.create_task(self._auto_close(poll_id, expires_in_s))
+
+        return poll
+
+    async def vote(self, poll_id: str, user_id: str, option_index: int) -> dict[str, Any] | None:
+        poll = self._polls.get(poll_id)
+        if poll is None:
+            return None
+
+        if not poll.vote(user_id, option_index):
+            return {"error": "Invalid vote", "poll_closed": poll.closed}
+
+        return {
+            "success": True,
+            "results": poll.results_text(),
+        }
+
+    async def close_poll(self, poll_id: str) -> dict[str, Any] | None:
+        poll = self._polls.get(poll_id)
+        if poll is None:
+            return None
+
+        poll.closed = True
+        results = poll.results_text()
+
+        await self._adapter._get_client().send_text_message(
+            chat_guid=poll.chat_guid,
+            content=results,
+        )
+        return {"closed": True, "results": results}
+
+    def get_poll(self, poll_id: str) -> Poll | None:
+        return self._polls.get(poll_id)
+
+    async def _auto_close(self, poll_id: str, delay: float) -> None:
+        await asyncio.sleep(delay)
+        await self.close_poll(poll_id)
+
+
+def _format_poll_text(poll: Poll) -> str:
+    lines = [f"Poll: {poll.question}", ""]
+    for opt in poll.options:
+        lines.append(f"  {opt.index + 1}. {opt.label}")
+    lines.append("")
+    lines.append("Reply with the number to vote (e.g. '1')")
+    return "\n".join(lines)

backend/package/yuxi/channels/adapters/imessage/probe.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from yuxi.channels.adapters.imessage.bridge_client import BlueBubblesClient
+
+_rpc_support_cache: dict[str, bool] = {}
+
+
+def get_rpc_support_cached(server_url: str) -> bool | None:
+    return _rpc_support_cache.get(server_url)
+
+
+def set_rpc_support_cached(server_url: str, supported: bool) -> None:
+    _rpc_support_cache[server_url] = supported
+
+
+def clear_rpc_support_cache() -> None:
+    _rpc_support_cache.clear()
+
+
+async def probe_bridge(bridge_client: BlueBubblesClient) -> dict[str, Any]:
+    """探测 BlueBubbles Server 状态和 iMessage 登录状态"""
+    server_info = await bridge_client.probe_server()
+    handle_info = await bridge_client.get_handle()
+
+    return {
+        "bridge_version": server_info.get("version"),
+        "bridge_os_version": server_info.get("os_version"),
+        "imessage_connected": handle_info.get("connected", False),
+        "imessage_handle": handle_info.get("handle"),
+        "server_info": server_info,
+        "handle_info": handle_info,
+    }
+
+
+async def run_doctor_diagnosis(
+    bridge_client: BlueBubblesClient,
+    security_warnings: list[str],
+    adapter_status: str,
+    self_handle: str | None,
+    pairing_info: dict[str, Any],
+) -> dict[str, Any]:
+    """全量 health + warning 报告。
+
+    Returns:
+        dict with status, connectivity, security, pairing sections
+    """
+    diagnosis: dict[str, Any] = {
+        "status": adapter_status,
+        "timestamp": None,
+        "connectivity": {},
+        "security": {
+            "warnings": security_warnings,
+            "assessment": _assess_security(security_warnings),
+        },
+        "pairing": pairing_info,
+    }
+
+    try:
+        server_info = await bridge_client.probe_server()
+        handle_info = await bridge_client.get_handle()
+
+        diagnosis["connectivity"] = {
+            "server_reachable": True,
+            "bridge_version": server_info.get("version", "unknown"),
+            "bridge_os_version": server_info.get("os_version", "unknown"),
+            "imessage_connected": handle_info.get("connected", False),
+            "imessage_handle": handle_info.get("handle", ""),
+            "self_handle": self_handle,
+        }
+        diagnosis["timestamp"] = None
+    except Exception as e:
+        diagnosis["connectivity"] = {
+            "server_reachable": False,
+            "error": str(e)[:200],
+        }
+
+    return diagnosis
+
+
+def _assess_security(warnings: list[str]) -> str:
+    if not warnings:
+        return "secure"
+    critical_keywords = ["disabled", "empty", "open"]
+    for w in warnings:
+        for kw in critical_keywords:
+            if kw in w.lower():
+                return "at_risk"
+    return "needs_attention"

backend/package/yuxi/channels/adapters/imessage/rate_limiter.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import time
+
+
+class LoopRateLimiter:
+    """循环速率限制：检测持续回显循环并抑制对话。
+
+    当同一会话在短时间内反复收到相同内容的消息时，
+    判定为回显循环，对 conversation 进行抑制。
+    """
+
+    def __init__(self, max_loops: int = 5, window_s: float = 30.0, cooldown_s: float = 60.0):
+        self._max_loops = max_loops
+        self._window_s = window_s
+        self._cooldown_s = cooldown_s
+        self._counters: dict[str, list[float]] = {}
+        self._suppressed: dict[str, float] = {}
+
+    def check(self, conversation_key: str, content: str) -> bool:
+        """返回 True 表示应该抑制此消息。"""
+
+        if conversation_key in self._suppressed:
+            suppressed_at = self._suppressed[conversation_key]
+            if time.monotonic() - suppressed_at < self._cooldown_s:
+                return True
+            del self._suppressed[conversation_key]
+
+        now = time.monotonic()
+        if conversation_key not in self._counters:
+            self._counters[conversation_key] = []
+        timestamps = self._counters[conversation_key]
+
+        timestamps.append(now)
+        timestamps[:] = [t for t in timestamps if now - t <= self._window_s]
+
+        if len(timestamps) >= self._max_loops:
+            self._suppressed[conversation_key] = now
+            self._counters.pop(conversation_key, None)
+            return True
+
+        return False
+
+    def is_suppressed(self, conversation_key: str) -> bool:
+        if conversation_key in self._suppressed:
+            if time.monotonic() - self._suppressed[conversation_key] < self._cooldown_s:
+                return True
+            del self._suppressed[conversation_key]
+        return False
+
+    def reset(self, conversation_key: str) -> None:
+        self._counters.pop(conversation_key, None)
+        self._suppressed.pop(conversation_key, None)

backend/package/yuxi/channels/adapters/imessage/reflection_guard.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import re
+
+_TOOL_CALL_RE = re.compile(r"\[TOOL_CALL:", re.IGNORECASE)
+_ASSISTANT_MARKER_RE = re.compile(r"\[ASSISTANT_SYSTEM\]", re.IGNORECASE)
+
+
+class ReflectionGuard:
+    """反射防护：检测入站消息中是否包含 Assistant 内部标记。
+
+    当出站消息包含 [TOOL_CALL:...] 等内部标记，且被 Bridge 反射回
+    入站事件时，应丢弃此类消息，避免 Assistant 将其解析为指令。
+    """
+
+    def __init__(self):
+        self._blocked_count = 0
+
+    def is_reflection(self, text: str) -> bool:
+        if not text:
+            return False
+        if _TOOL_CALL_RE.search(text):
+            return True
+        if _ASSISTANT_MARKER_RE.search(text):
+            return True
+        return False
+
+    def mark_blocked(self) -> None:
+        self._blocked_count += 1
+
+    @property
+    def blocked_count(self) -> int:
+        return self._blocked_count

backend/package/yuxi/channels/adapters/imessage/runtime_store.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from typing import Any
+
+_runtime_store: dict[str, Any] = {}
+
+
+def set_runtime(key: str, value: Any) -> None:
+    _runtime_store[key] = value
+
+
+def get_runtime(key: str, default: Any = None) -> Any:
+    return _runtime_store.get(key, default)
+
+
+def has_runtime(key: str) -> bool:
+    return key in _runtime_store
+
+
+def delete_runtime(key: str) -> None:
+    _runtime_store.pop(key, None)
+
+
+def clear_runtime() -> None:
+    _runtime_store.clear()

backend/package/yuxi/channels/adapters/imessage/sanitize.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import re
+
+_CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")
+_MULTI_NEWLINE_RE = re.compile(r"\n{3,}")
+_LENGTH_PREFIX_RE = re.compile(r"^(\d+)\n")
+_TERMINAL_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]")
+_MAX_TERMINAL_MSG_LEN = 200
+
+
+def sanitize_terminal_text(text: str, max_len: int = _MAX_TERMINAL_MSG_LEN) -> str:
+    """净化终端输出文本，移除 ANSI 转义序列和控制字符。
+
+    用于日志记录和错误消息净化。
+    """
+    if not text:
+        return ""
+    text = _TERMINAL_ESCAPE_RE.sub("", text)
+    text = _CONTROL_CHAR_RE.sub("", text)
+    if len(text) > max_len:
+        text = text[:max_len] + "..."
+    return text
+
+
+def sanitize_outbound_text(text: str) -> str:
+    """净化出站文本，移除 iMessage 中不支持的格式。
+
+    - 去除 \\r 字符（换行标准化）
+    - 去除长度前缀损坏（如 "42\\nThis is message"）
+    - 过滤控制字符
+    - 合并多余空白行
+    """
+    text = text.replace("\r", "")
+
+    text = _strip_length_prefix(text)
+
+    text = _CONTROL_CHAR_RE.sub("", text)
+
+    text = _MULTI_NEWLINE_RE.sub("\n\n", text)
+    text = text.strip()
+
+    return text
+
+
+def _strip_length_prefix(text: str) -> str:
+    match = _LENGTH_PREFIX_RE.match(text)
+    if match:
+        prefix_len = int(match.group(1))
+        remaining = text[match.end() :]
+        if abs(len(remaining) - prefix_len) <= 5:
+            return remaining
+    return text
+
+
+def media_placeholder(media_type: str) -> str:
+    """为纯媒体消息生成占位文本。
+
+    OpenClaw 在无文本媒体消息中生成 <media:image> 等占位符。
+    """
+    placeholders = {
+        "image": "<media:image>",
+        "video": "<media:video>",
+        "audio": "<media:audio>",
+        "file": "<media:file>",
+    }
+    return placeholders.get(media_type, "<media:file>")

backend/package/yuxi/channels/adapters/imessage/security.py

@@ -0,0 +1,371 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from dataclasses import dataclass, field
+from enum import StrEnum
+from pathlib import Path
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+
+class DmPolicy(StrEnum):
+    PAIRING = "pairing"
+    ALLOWLIST = "allowlist"
+    OPEN = "open"
+    DISABLED = "disabled"
+
+
+class GroupPolicy(StrEnum):
+    OPEN = "open"
+    ALLOWLIST = "allowlist"
+    DISABLED = "disabled"
+
+
+@dataclass
+class SecurityCheckResult:
+    allowed: bool
+    reject_reason: str | None = None
+    reply: str | None = None
+
+
+@dataclass
+class IMessageSecurityConfig:
+    dm_policy: DmPolicy = DmPolicy.PAIRING
+    group_policy: GroupPolicy = GroupPolicy.ALLOWLIST
+    allow_from: list[str] = field(default_factory=list)
+    group_allow_from: list[str] = field(default_factory=list)
+    require_mention: bool = False
+
+
+class IMessageSecurityPolicy:
+    def __init__(self, config: dict[str, Any], storage_dir: str | None = None, config_writes: bool = True):
+        self._config = config
+        self._dm_policy = resolve_dm_policy(config)
+        self._group_policy = resolve_group_policy(config)
+        self._allow_list = _normalize_entries(config.get("allowFrom", config.get("allow_from", [])))
+        self._group_allow_list = _normalize_entries(config.get("groupAllowFrom", config.get("group_allow_from", [])))
+        self._require_mention = config.get("requireMention", config.get("require_mention", False))
+        self._groups_config: dict[str, dict] = config.get("groups", {})
+        self._group_allow_from_fallback = config.get(
+            "groupAllowFromFallback", config.get("group_allow_from_fallback", False)
+        )
+        self._storage_dir = storage_dir or str(Path.home() / ".yuxi" / "imessage")
+        self._config_writes = config_writes
+        self._account_id = config.get("accountId", config.get("account_id", "default"))
+        self._context_visibility_mode = config.get(
+            "contextVisibilityMode", config.get("context_visibility_mode", "allowlist")
+        )
+        self._warned_policies: set[str] = set()
+        self._pinned_dm_owner: str | None = None
+
+    def _schedule_save(self) -> None:
+        if not self._config_writes:
+            return
+        try:
+            loop = asyncio.get_running_loop()
+            loop.create_task(self.save_allowlist())
+        except RuntimeError:
+            pass
+
+    async def save_allowlist(self) -> None:
+        dir_path = Path(self._storage_dir) / self._account_id
+        dir_path.mkdir(parents=True, exist_ok=True)
+        data = {
+            "allow_from": list(self._allow_list),
+            "group_allow_from": list(self._group_allow_list),
+        }
+        file_path = dir_path / "allowlist.json"
+        file_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    async def load_allowlist(self) -> None:
+        file_path = Path(self._storage_dir) / self._account_id / "allowlist.json"
+        if not file_path.exists():
+            return
+        try:
+            data = json.loads(file_path.read_text(encoding="utf-8"))
+            if isinstance(data, dict):
+                loaded_allow = _normalize_entries(data.get("allow_from", []))
+                loaded_group = _normalize_entries(data.get("group_allow_from", []))
+                for entry in loaded_allow:
+                    if entry not in self._allow_list:
+                        self._allow_list.append(entry)
+                for entry in loaded_group:
+                    if entry not in self._group_allow_list:
+                        self._group_allow_list.append(entry)
+        except (json.JSONDecodeError, OSError) as e:
+            logger.warning(f"[iMessage/Security] Failed to load allowlist: {e}")
+
+    def check_dm_access(self, sender_handle: str) -> SecurityCheckResult:
+        if self._dm_policy == DmPolicy.OPEN:
+            return SecurityCheckResult(allowed=True)
+        if self._dm_policy == DmPolicy.DISABLED:
+            return SecurityCheckResult(
+                allowed=False,
+                reject_reason="dm_disabled",
+                reply="iMessage DM is disabled for this bot.",
+            )
+        if self._dm_policy == DmPolicy.ALLOWLIST:
+            if _match_entry(sender_handle, self._allow_list):
+                return SecurityCheckResult(allowed=True)
+            return SecurityCheckResult(
+                allowed=False,
+                reject_reason="not_in_dm_allowlist",
+                reply="You are not authorized to message this bot.",
+            )
+        if self._dm_policy == DmPolicy.PAIRING:
+            return SecurityCheckResult(allowed=True)
+        return SecurityCheckResult(allowed=False, reject_reason="unknown_policy")
+
+    def check_group_access(self, chat_guid: str) -> SecurityCheckResult:
+        group_override = self._groups_config.get(chat_guid, {})
+        if group_override.get("enabled") is False:
+            return SecurityCheckResult(
+                allowed=False,
+                reject_reason="group_disabled_by_override",
+                reply="This bot is not enabled in this group chat.",
+            )
+
+        if self._group_policy == GroupPolicy.OPEN:
+            return SecurityCheckResult(allowed=True)
+        if self._group_policy == GroupPolicy.DISABLED:
+            return SecurityCheckResult(
+                allowed=False,
+                reject_reason="group_disabled",
+                reply="This bot does not participate in group chats.",
+            )
+        if self._group_policy == GroupPolicy.ALLOWLIST:
+            if _match_entry(chat_guid, self._group_allow_list):
+                return SecurityCheckResult(allowed=True)
+            if self._group_allow_from_fallback and _match_entry(chat_guid, self._allow_list):
+                return SecurityCheckResult(allowed=True)
+            return SecurityCheckResult(
+                allowed=False,
+                reject_reason="group_not_in_allowlist",
+                reply="This bot is not authorized in this group chat.",
+            )
+        return SecurityCheckResult(allowed=False, reject_reason="unknown_policy")
+
+    def check_mention_required(self, chat_guid: str, mentions, bot_handle: str) -> SecurityCheckResult:
+        group_override = self._groups_config.get(chat_guid, {})
+        require_mention = group_override.get("require_mention", self._require_mention)
+        if not require_mention:
+            return SecurityCheckResult(allowed=True)
+
+        mentioned_ids: list[str] = []
+        if mentions:
+            mentioned_ids = mentions.mentioned_user_ids or []
+        elif isinstance(mentions, list):
+            mentioned_ids = mentions
+        elif isinstance(mentions, dict):
+            mentioned_ids = mentions.get("mentioned_user_ids", [])
+
+        bot_clean = bot_handle.strip().replace("+", "").replace(" ", "")
+        for uid in mentioned_ids:
+            uid_clean = uid.strip().replace("+", "").replace(" ", "")
+            if uid_clean == bot_clean:
+                return SecurityCheckResult(allowed=True)
+
+        return SecurityCheckResult(
+            allowed=False,
+            reject_reason="mention_required",
+            reply=None,
+        )
+
+    def collect_warnings(self) -> list[str]:
+        warnings: list[str] = []
+        if self._dm_policy == DmPolicy.OPEN:
+            warnings.append(
+                "dmPolicy='open' — anyone can message this bot; consider using 'pairing' or 'allowlist' for security"
+            )
+        if self._dm_policy == DmPolicy.ALLOWLIST and not self._allow_list:
+            warnings.append("dmPolicy='allowlist' but allowFrom is empty — no one can message the bot")
+        if self._group_policy == GroupPolicy.OPEN and not self._group_allow_list:
+            warnings.append("groupPolicy='open' but no groupAllowFrom configured — consider adding a group allowlist")
+        if self._group_policy == GroupPolicy.ALLOWLIST and not self._group_allow_list:
+            warnings.append("groupPolicy='allowlist' but groupAllowFrom is empty — no group can interact")
+        if self._group_allow_from_fallback:
+            warnings.append(
+                "groupAllowFromFallback=true — group allowlist falls back to DM allowlist; "
+                "consider disabling this for stricter group access control"
+            )
+        return warnings
+
+    def add_to_allow_list(self, handle: str) -> None:
+        handle_clean = _normalize_handle(handle)
+        if handle_clean not in self._allow_list:
+            self._allow_list.append(handle_clean)
+            self._schedule_save()
+
+    def remove_from_allow_list(self, handle: str) -> bool:
+        handle_clean = _normalize_handle(handle)
+        if handle_clean in self._allow_list:
+            self._allow_list.remove(handle_clean)
+            self._schedule_save()
+            return True
+        return False
+
+    def add_to_group_allow_list(self, chat_guid: str) -> None:
+        if chat_guid not in self._group_allow_list:
+            self._group_allow_list.append(chat_guid)
+            self._schedule_save()
+
+    def remove_from_group_allow_list(self, chat_guid: str) -> bool:
+        if chat_guid in self._group_allow_list:
+            self._group_allow_list.remove(chat_guid)
+            self._schedule_save()
+            return True
+        return False
+
+    def resolve_runtime_group_policy(self, chat_guid: str) -> GroupPolicy:
+        group_override = self._groups_config.get(chat_guid, {})
+        if "group_policy" in group_override:
+            return GroupPolicy(group_override["group_policy"])
+
+        warning_key = f"group_policy_fallback:{chat_guid}"
+        if warning_key not in self._warned_policies:
+            logger.warning(
+                f"[iMessage/Security] No per-group policy for {chat_guid}, "
+                f"falling back to default groupPolicy='{self._group_policy.value}'. "
+                f"Consider adding a group override config."
+            )
+            self._warned_policies.add(warning_key)
+
+        return self._group_policy
+
+    def evaluate_context_visibility(
+        self,
+        chat_guid: str,
+        reply_sender_handle: str,
+    ) -> bool:
+        if self._context_visibility_mode == "open":
+            return True
+
+        if self._context_visibility_mode == "disabled":
+            return False
+
+        if self._context_visibility_mode == "allowlist":
+            chat_type = _resolve_chat_type_from_guid(chat_guid)
+            if chat_type == "group":
+                if _match_entry(chat_guid, self._group_allow_list):
+                    return True
+                if self._group_allow_from_fallback and _match_entry(chat_guid, self._allow_list):
+                    return True
+                if _match_entry(reply_sender_handle, self._allow_list):
+                    return True
+                return False
+            return _match_entry(reply_sender_handle, self._allow_list)
+
+        return False
+
+    def resolve_pinned_dm_owner(self) -> str | None:
+        if self._pinned_dm_owner:
+            return self._pinned_dm_owner
+        for entry in self._allow_list:
+            entry_stripped, prefix = _strip_prefix(entry)
+            if prefix in ("imessage", "sms", None) and entry_stripped:
+                clean = _normalize_handle(entry_stripped)
+                if clean and "@" not in clean:
+                    self._pinned_dm_owner = clean
+                    return clean
+        return None
+
+    def should_update_last_route(self, sender_handle: str) -> bool:
+        owner = self.resolve_pinned_dm_owner()
+        if owner:
+            sender_clean = _normalize_handle(sender_handle)
+            return sender_clean == owner
+        return True
+
+    @property
+    def dm_policy(self) -> DmPolicy:
+        return self._dm_policy
+
+    @property
+    def group_policy(self) -> GroupPolicy:
+        return self._group_policy
+
+    @property
+    def allow_list(self) -> list[str]:
+        return list(self._allow_list)
+
+    @property
+    def group_allow_list(self) -> list[str]:
+        return list(self._group_allow_list)
+
+    @property
+    def require_mention(self) -> bool:
+        return self._require_mention
+
+
+def resolve_dm_policy(config: dict[str, Any]) -> DmPolicy:
+    explicit = config.get("dmPolicy", config.get("dm_policy", ""))
+    if explicit:
+        try:
+            return DmPolicy(explicit)
+        except ValueError:
+            logger.warning(f"[iMessage] Invalid dmPolicy '{explicit}', falling back to auto-detect")
+    allow_from = config.get("allowFrom", config.get("allow_from", []))
+    if not allow_from or "*" in allow_from:
+        return DmPolicy.OPEN
+    return DmPolicy.ALLOWLIST
+
+
+def resolve_group_policy(config: dict[str, Any]) -> GroupPolicy:
+    explicit = config.get("groupPolicy", config.get("group_policy", ""))
+    if explicit:
+        try:
+            return GroupPolicy(explicit)
+        except ValueError:
+            logger.warning(f"[iMessage] Invalid groupPolicy '{explicit}', using ALLOWLIST")
+    return GroupPolicy.ALLOWLIST
+
+
+def _normalize_handle(handle: str) -> str:
+    return handle.strip().replace(" ", "").lstrip("+")
+
+
+def _normalize_entries(entries: list[str]) -> list[str]:
+    return [_normalize_handle(e) for e in entries if e]
+
+
+PREFIX_MAP: dict[str, str] = {
+    "chat_id:": "chat_id",
+    "chat_guid:": "chat_guid",
+    "chat_identifier:": "chat_identifier",
+    "imessage:": "imessage",
+    "sms:": "sms",
+    "auto:": "auto",
+}
+
+
+def _strip_prefix(entry: str) -> tuple[str, str | None]:
+    entry_lower = entry.lower()
+    for prefix, name in PREFIX_MAP.items():
+        if entry_lower.startswith(prefix):
+            return entry[len(prefix) :], name
+    return entry, None
+
+
+def _match_entry(target: str, allow_list: list[str]) -> bool:
+    if "*" in allow_list:
+        return True
+    target_clean = _normalize_handle(target)
+    for entry in allow_list:
+        entry_stripped, prefix = _strip_prefix(entry)
+        if prefix == "chat_id" and target_clean in [_normalize_handle(entry_stripped), entry_stripped]:
+            return True
+        if prefix in ("chat_guid", "chat_identifier") and entry_stripped == target_clean:
+            return True
+        if prefix in ("imessage", "sms", "auto", None):
+            entry_clean = _normalize_handle(entry_stripped)
+            if entry_clean == target_clean:
+                return True
+    return False
+
+
+def _resolve_chat_type_from_guid(chat_guid: str) -> str:
+    if ";-;" in chat_guid or "@chat" in chat_guid:
+        return "group"
+    return "direct"

backend/package/yuxi/channels/adapters/imessage/sent_cache.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import Any
+
+from yuxi.utils.logging_config import logger
+
+
+class SentMessageCache:
+    def __init__(self, storage_dir: str | None = None, max_entries: int = 1000):
+        self._storage_dir = Path(storage_dir or str(Path.home() / ".yuxi" / "imessage"))
+        self._max_entries = max_entries
+        self._entries: dict[str, dict[str, Any]] = {}
+        self._storage_dir.mkdir(parents=True, exist_ok=True)
+        self._load()
+
+    def record(self, message_id: str, content: str, chat_guid: str, metadata: dict[str, Any] | None = None) -> None:
+        entry = {
+            "message_id": message_id,
+            "content": content[:500],
+            "chat_guid": chat_guid,
+            "sent_at": time.time(),
+            "metadata": metadata or {},
+        }
+        self._entries[message_id] = entry
+        if len(self._entries) > self._max_entries:
+            oldest = min(self._entries.values(), key=lambda e: e["sent_at"])
+            self._entries.pop(oldest["message_id"], None)
+
+    def get(self, message_id: str) -> dict[str, Any] | None:
+        return self._entries.get(message_id)
+
+    def get_content(self, message_id: str) -> str | None:
+        entry = self._entries.get(message_id)
+        return entry["content"] if entry else None
+
+    def list_recent(self, limit: int = 50) -> list[dict[str, Any]]:
+        entries = sorted(self._entries.values(), key=lambda e: e["sent_at"], reverse=True)
+        return entries[:limit]
+
+    async def persist(self) -> None:
+        file_path = self._storage_dir / "sent_messages.json"
+        file_path.write_text(json.dumps(self._entries, ensure_ascii=False, indent=2), encoding="utf-8")
+        logger.debug(f"[iMessage/SentCache] Persisted {len(self._entries)} entries")
+
+    def _load(self) -> None:
+        file_path = self._storage_dir / "sent_messages.json"
+        if not file_path.exists():
+            return
+        try:
+            self._entries = json.loads(file_path.read_text(encoding="utf-8"))
+            now = time.time()
+            self._entries = {
+                mid: entry for mid, entry in self._entries.items() if now - entry.get("sent_at", 0) < 86400
+            }
+        except (json.JSONDecodeError, OSError):
+            self._entries = {}

backend/package/yuxi/channels/adapters/imessage/session.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from yuxi.channels.models import ChannelIdentity, ChatType
+
+
+def resolve_thread_key(
+    identity: ChannelIdentity,
+    account_id: str = "default",
+    pinned_dm_owner: str | None = None,
+) -> str:
+    channel_user_id = identity.channel_user_id
+    chat_guid = identity.channel_chat_id
+    chat_type = resolve_chat_type(chat_guid)
+
+    if chat_type == ChatType.GROUP:
+        return f"agent:main:imessage:group:{chat_guid}:{account_id}"
+
+    if pinned_dm_owner:
+        clean_owner = pinned_dm_owner.replace("+", "").replace("-", "").replace(" ", "")
+        return f"agent:main:imessage:dm:{clean_owner}:{account_id}"
+
+    clean_phone = channel_user_id.replace("+", "").replace("-", "").replace(" ", "")
+    return f"agent:main:imessage:dm:{clean_phone}:{account_id}"
+
+
+def resolve_chat_type(chat_guid: str) -> ChatType:
+    if ";-;" in chat_guid or "@chat" in chat_guid:
+        return ChatType.GROUP
+    return ChatType.DIRECT

backend/package/yuxi/channels/adapters/imessage/setup_wizard.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+_registered_plugin: bool = False
+
+
+def register_setup_plugin() -> bool:
+    global _registered_plugin
+    if _registered_plugin:
+        return True
+    _registered_plugin = True
+    return True
+
+
+def is_setup_plugin_registered() -> bool:
+    return _registered_plugin
+
+
+def unregister_setup_plugin() -> None:
+    global _registered_plugin
+    _registered_plugin = False
+
+
+@dataclass
+class SetupWizardStep:
+    step_id: str
+    label: str
+    description: str
+    field: str
+    field_type: str
+    required: bool = True
+    default_value: Any = None
+    options: list[dict[str, str]] = field(default_factory=list)
+    hint: str = ""
+
+
+SETUP_STEPS: list[SetupWizardStep] = [
+    SetupWizardStep(
+        step_id="server_url",
+        label="BlueBubbles Server URL",
+        description="The URL of your BlueBubbles server (e.g. http://192.168.1.100:1234)",
+        field="server_url",
+        field_type="text",
+        default_value="http://localhost:1234",
+    ),
+    SetupWizardStep(
+        step_id="password",
+        label="Server Password",
+        description="The password configured in your BlueBubbles server settings",
+        field="password",
+        field_type="password",
+    ),
+    SetupWizardStep(
+        step_id="dm_policy",
+        label="DM Access Policy",
+        description="How should the bot handle direct messages?",
+        field="dm_policy",
+        field_type="select",
+        default_value="pairing",
+        options=[
+            {"value": "pairing", "label": "Pairing (users must request approval)"},
+            {"value": "allowlist", "label": "Allowlist (only approved contacts)"},
+            {"value": "open", "label": "Open (anyone can DM)"},
+            {"value": "disabled", "label": "Disabled (no DMs)"},
+        ],
+        hint="Pairing is recommended for controlled access. Users get a code to share with admins.",
+    ),
+    SetupWizardStep(
+        step_id="group_policy",
+        label="Group Chat Policy",
+        description="How should the bot handle group chats?",
+        field="group_policy",
+        field_type="select",
+        default_value="allowlist",
+        options=[
+            {"value": "open", "label": "Open (any group can add the bot)"},
+            {"value": "allowlist", "label": "Allowlist (only approved groups)"},
+            {"value": "disabled", "label": "Disabled (no group chats)"},
+        ],
+    ),
+    SetupWizardStep(
+        step_id="allow_from",
+        label="Allowed Contacts",
+        description="List of contacts allowed to interact with the bot (one per line). Use * for everyone.",
+        field="allow_from",
+        field_type="textarea",
+        required=False,
+        hint="Phone numbers (E.164), email addresses, or * for all. Prefix with chat_id: or chat_guid: for specific chats.",
+    ),
+    SetupWizardStep(
+        step_id="completed",
+        label="Setup Complete",
+        description="Your iMessage channel is now configured! The bot will restart with these settings.",
+        field="completed",
+        field_type="info",
+    ),
+]
+
+
+def get_setup_steps() -> list[dict[str, Any]]:
+    return [
+        {
+            "step_id": s.step_id,
+            "label": s.label,
+            "description": s.description,
+            "field": s.field,
+            "field_type": s.field_type,
+            "required": s.required,
+            "default_value": s.default_value,
+            "options": s.options,
+            "hint": s.hint,
+        }
+        for s in SETUP_STEPS
+    ]
+
+
+def generate_permissions_summary(config: dict[str, Any]) -> str:
+    """生成权限提醒摘要文本。"""
+    dm_policy = config.get("dmPolicy", config.get("dm_policy", "pairing"))
+    lines = [
+        "iMessage channel configured successfully!",
+        "",
+        f"DM Policy: {dm_policy}",
+        f"Group Policy: {config.get('groupPolicy', config.get('group_policy', 'allowlist'))}",
+        "",
+        "Notes:",
+        "- Ensure BlueBubbles server has 'Private API' enabled",
+        "- The bot must be signed into iMessage on the Mac running BlueBubbles",
+        "- Group chats: add the bot to a group, then approve the group GUID via allowlist",
+    ]
+    return "\n".join(lines)

backend/package/yuxi/channels/adapters/imessage/sticker_cache.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import Any
+
+
+class StickerCache:
+    def __init__(self, cache_dir: str | None = None, max_entries: int = 500):
+        self._cache_dir = Path(cache_dir or str(Path.home() / ".yuxi" / "imessage" / "stickers"))
+        self._max_entries = max_entries
+        self._index: dict[str, dict[str, Any]] = {}
+        self._cache_dir.mkdir(parents=True, exist_ok=True)
+        self._load_index()
+
+    def add(self, sticker_id: str, url: str, metadata: dict[str, Any] | None = None) -> None:
+        entry = {
+            "sticker_id": sticker_id,
+            "url": url,
+            "metadata": metadata or {},
+            "cached_at": time.time(),
+        }
+        self._index[sticker_id] = entry
+        if len(self._index) > self._max_entries:
+            oldest = min(self._index.values(), key=lambda e: e["cached_at"])
+            self._index.pop(oldest["sticker_id"], None)
+        self._save_index()
+
+    def get(self, sticker_id: str) -> dict[str, Any] | None:
+        return self._index.get(sticker_id)
+
+    def get_url(self, sticker_id: str) -> str | None:
+        entry = self._index.get(sticker_id)
+        return entry["url"] if entry else None
+
+    def clear(self) -> None:
+        self._index.clear()
+        self._save_index()
+
+    def _load_index(self) -> None:
+        index_path = self._cache_dir / "sticker_index.json"
+        if index_path.exists():
+            try:
+                self._index = json.loads(index_path.read_text(encoding="utf-8"))
+            except (json.JSONDecodeError, OSError):
+                self._index = {}
+
+    def _save_index(self) -> None:
+        index_path = self._cache_dir / "sticker_index.json"
+        index_path.write_text(json.dumps(self._index, ensure_ascii=False, indent=2), encoding="utf-8")

backend/package/yuxi/channels/adapters/imessage/tapback.py

@@ -0,0 +1,58 @@
+TAPBACK_VALUE_MAP: dict[str, int] = {
+    "love": 0,
+    "heart": 0,
+    "like": 1,
+    "thumbs_up": 1,
+    "dislike": 2,
+    "thumbs_down": 2,
+    "laugh": 3,
+    "exclaim": 4,
+    "question": 5,
+}
+
+TAPBACK_EMOJI_MAP: dict[str, str] = {
+    "❤️": "love",
+    "👍": "like",
+    "👎": "dislike",
+    "😂": "laugh",
+    "❗": "exclaim",
+    "❓": "question",
+}
+
+TAPBACK_REVERSE_MAP: dict[int, str] = {
+    0: "❤️",
+    1: "👍",
+    2: "👎",
+    3: "😂",
+    4: "❗",
+    5: "❓",
+}
+
+TAPBACK_NAME_MAP: dict[int, str] = {
+    0: "❤️ Loved",
+    1: "👍 Liked",
+    2: "👎 Disliked",
+    3: "😂 Laughed",
+    4: "❗ Emphasized",
+    5: "❓ Questioned",
+}
+
+
+def emoji_to_tapback(emoji: str) -> str | None:
+    return TAPBACK_EMOJI_MAP.get(emoji)
+
+
+def resolve_tapback_value(tapback_name: str) -> int | None:
+    return TAPBACK_VALUE_MAP.get(tapback_name.lower().strip().lstrip(":"))
+
+
+def tapback_to_emoji(tapback_value: int) -> str | None:
+    return TAPBACK_REVERSE_MAP.get(tapback_value)
+
+
+def describe_tapback(tapback_value: int) -> str:
+    return TAPBACK_NAME_MAP.get(tapback_value, "Reacted")
+
+
+def is_valid_tapback(emoji: str) -> bool:
+    return emoji in TAPBACK_EMOJI_MAP

backend/package/yuxi/channels/adapters/imessage/target_parsing_helpers.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Literal
+
+TargetType = Literal["chat_id", "chat_guid", "chat_identifier", "handle", "phone", "email", "unknown"]
+
+
+@dataclass
+class ParsedTarget:
+    type: TargetType
+    value: str
+
+
+def parse_imessage_target(raw_target: str) -> ParsedTarget:
+    from yuxi.channels.adapters.imessage.targets import parse_target
+
+    result = parse_target(raw_target)
+    return ParsedTarget(type=result["type"], value=result["value"])
+
+
+def is_explicit_target_id(target: str) -> bool:
+    from yuxi.channels.adapters.imessage.targets import looks_like_explicit_target_id
+
+    return looks_like_explicit_target_id(target)
+
+
+def is_imessage_target_id(target: str) -> bool:
+    from yuxi.channels.adapters.imessage.targets import looks_like_imessage_target_id
+
+    return looks_like_imessage_target_id(target)
+
+
+def resolve_target_prefix(target: str) -> str | None:
+    from yuxi.channels.adapters.imessage.targets import TARGET_PREFIX_MAP
+
+    target_lower = target.lower()
+    for prefix, type_name in TARGET_PREFIX_MAP.items():
+        if target_lower.startswith(prefix):
+            return type_name
+    return None
+
+
+def strip_target_prefix(target: str) -> tuple[str, str | None]:
+    from yuxi.channels.adapters.imessage.targets import TARGET_PREFIX_MAP
+
+    target_lower = target.lower()
+    for prefix, type_name in TARGET_PREFIX_MAP.items():
+        if target_lower.startswith(prefix):
+            return target[len(prefix) :], type_name
+    return target, None
+
+
+def normalize_imessage_handle(handle: str) -> str:
+    from yuxi.channels.adapters.imessage.normalize import normalize_e164
+
+    return normalize_e164(handle)
+
+
+def normalize_imessage_handle_preserve_alias(handle: str) -> str:
+    from yuxi.channels.adapters.imessage.normalize import _strip_service_prefixes
+
+    return _strip_service_prefixes(handle)
+
+
+def looks_like_phone(handle: str) -> bool:
+    cleaned = handle.strip().replace(" ", "").lstrip("+")
+    return cleaned.isdigit() and len(cleaned) >= 7
+
+
+def looks_like_email(handle: str) -> bool:
+    return "@" in handle and "." in handle.split("@")[-1]
+
+
+__all__ = [
+    "ParsedTarget",
+    "TargetType",
+    "parse_imessage_target",
+    "is_explicit_target_id",
+    "is_imessage_target_id",
+    "resolve_target_prefix",
+    "strip_target_prefix",
+    "normalize_imessage_handle",
+    "normalize_imessage_handle_preserve_alias",
+    "looks_like_phone",
+    "looks_like_email",
+]

backend/package/yuxi/channels/adapters/imessage/targets.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from yuxi.channels.models import ChatType
+
+TARGET_PREFIX_MAP: dict[str, str] = {
+    "chat_id:": "chat_id",
+    "chat_guid:": "chat_guid",
+    "chat_identifier:": "chat_identifier",
+    "handle:": "handle",
+    "phone:": "phone",
+    "email:": "email",
+}
+
+EXPLICIT_TARGET_RE_PREFIXES = (
+    "chat_id:",
+    "chat_guid:",
+    "chat_identifier:",
+    "iMessage;",
+    "iMessage",
+    "handle:",
+    "phone:",
+    "email:",
+)
+
+IMESSAGE_TARGET_PREFIXES = (
+    "chat_id:",
+    "chat_guid:",
+    "chat_identifier:",
+    "iMessage;",
+    "iMessage",
+    "handle:",
+    "phone:",
+    "email:",
+    "tel:",
+    "mailto:",
+    "+",
+)
+
+
+def parse_target(raw_target: str) -> dict[str, str]:
+    """解析目标字符串，返回 {type, value}。
+
+    支持的类型：
+    - chat_id:<value>   → chat_id
+    - chat_guid:<value> → chat_guid
+    - chat_identifier:<value> → chat_identifier (chat_id 或 chat_guid)
+    - handle:<value>    → handle
+    - phone:<value>     → phone (E.164)
+    - email:<value>     → email
+    - 其他：自动推断为 handle
+    """
+    if not raw_target:
+        return {"type": "unknown", "value": ""}
+
+    raw_lower = raw_target.lower()
+    for prefix, type_name in TARGET_PREFIX_MAP.items():
+        if raw_lower.startswith(prefix):
+            return {"type": type_name, "value": raw_target[len(prefix) :]}
+
+    if looks_like_explicit_target_id(raw_target):
+        if ";-;" in raw_target or "@chat" in raw_target:
+            return {"type": "chat_guid", "value": raw_target}
+        if raw_target.startswith("iMessage;"):
+            return {"type": "chat_id", "value": raw_target}
+
+    if "@" in raw_target and "." in raw_target.split("@")[-1]:
+        return {"type": "email", "value": raw_target}
+
+    return {"type": "handle", "value": raw_target}
+
+
+def looks_like_explicit_target_id(target: str) -> bool:
+    return any(target.lower().startswith(p.lower()) for p in EXPLICIT_TARGET_RE_PREFIXES)
+
+
+def looks_like_imessage_target_id(target: str) -> bool:
+    return any(target.lower().startswith(p.lower()) for p in IMESSAGE_TARGET_PREFIXES)
+
+
+def resolve_chat_type_from_target(target: str) -> ChatType:
+    parsed = parse_target(target)
+    value = parsed["value"]
+    if ";-;" in value or "@chat" in value:
+        return ChatType.GROUP
+    return ChatType.DIRECT

backend/package/yuxi/channels/adapters/imessage/threading.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+
+def format_reply_context(
+    reply_to_message_id: str | None = None,
+    reply_to_text: str | None = None,
+    reply_to_sender: str | None = None,
+) -> str:
+    """格式化引用回复上下文。
+
+    生成如 "[Replying to +8613800138000 id:msg_001]" 格式的上下文行，
+    附加在出站消息文本前。
+    """
+    if not reply_to_message_id and not reply_to_text:
+        return ""
+
+    parts: list[str] = ["[Replying to"]
+
+    if reply_to_sender:
+        parts.append(f" {reply_to_sender}")
+
+    if reply_to_message_id:
+        parts.append(f" id:{reply_to_message_id}")
+
+    if reply_to_text:
+        preview = reply_to_text[:100].replace("\n", " ")
+        parts.append(f' "{preview}"')
+
+    parts.append("]\n")
+    return " ".join(parts) if len(parts) > 2 else ""
+
+
+def describe_reply_context(data: dict) -> str:
+    """从消息数据中提取并格式化回复上下文。"""
+    reply_to_guid = data.get("replyToGuid")
+    reply_to_text = data.get("threadOriginatorText")
+    reply_to_sender = data.get("threadOriginator")
+
+    return format_reply_context(
+        reply_to_message_id=reply_to_guid,
+        reply_to_text=reply_to_text,
+        reply_to_sender=reply_to_sender,
+    )

backend/package/yuxi/channels/adapters/imessage/transport.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+
+from yuxi.utils.logging_config import logger
+
+DEFAULT_TRANSPORT_TIMEOUT_S = 30.0
+DEFAULT_TRANSPORT_POLL_INTERVAL_S = 0.5
+
+
+async def wait_for_transport_ready(
+    probe_fn: Callable[[], Awaitable[bool]],
+    timeout_s: float = DEFAULT_TRANSPORT_TIMEOUT_S,
+    poll_interval_s: float = DEFAULT_TRANSPORT_POLL_INTERVAL_S,
+) -> bool:
+    deadline = asyncio.get_event_loop().time() + timeout_s
+
+    while asyncio.get_event_loop().time() < deadline:
+        try:
+            ready = await probe_fn()
+            if ready:
+                logger.info("[iMessage/Transport] Transport ready")
+                return True
+        except Exception:
+            pass
+
+        await asyncio.sleep(poll_interval_s)
+
+    logger.error(f"[iMessage/Transport] Transport not ready after {timeout_s}s")
+    return False