diff --git a/sdk/app/routers/fleet.py b/sdk/app/routers/fleet.py
new file mode 100644
index 0000000000..3de9c406d7
--- /dev/null
+++ b/sdk/app/routers/fleet.py
@@ -0,0 +1,409 @@
+"""
+多手机设备管理接口。
+
+面向存客宝/触客宝/超管等外部系统：把单机 SDK 能力包装成可筛选、
+可批量执行、可审计的 Fleet API。单机能力仍由 devices/unified 路由负责。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+
+from services.device_id_util import device_id_md5
+from services.ws_hub import ws_hub
+
+
+router = APIRouter()
+
+
+WRITE_ACTIONS = {
+    "send_message",
+    "batch_send",
+    "mass_send",
+    "add_friend",
+    "batch_add_friend",
+    "post_moments",
+    "like_moments",
+    "comment_moments",
+    "create_group",
+    "invite_to_group",
+    "remove_from_group",
+    "set_group_notice",
+    "set_group_name",
+    "set_group_welcome",
+    "delete_tag",
+    "create_tag",
+    "tag_add",
+    "tag_remove",
+}
+
+
+def _result_success(result: Dict[str, Any]) -> bool:
+    """只接受执行端明确 success=true，禁止以 HTTP/code 单独判定成功。"""
+    if not isinstance(result, dict):
+        return False
+    payload = result.get("data") if isinstance(result.get("data"), dict) else result
+    return payload.get("success") is True
+
+
+def _normalize_agent_result(action: str, result: Dict[str, Any]) -> Dict[str, Any]:
+    """将 Agent 原始动作回执归一化为 Fleet 的标准可验收结构。"""
+    normalized = dict(result or {})
+    data = dict(normalized.get("data") or {})
+    raw = normalized.get("raw_rpc_receipt") or dict(normalized)
+    trace_id = (normalized.get("trace_id") or normalized.get("command_id")
+                or raw.get("trace_id") or raw.get("command_id"))
+
+    # Android Agent 的截图字段是 image_base64；Fleet 标准字段为 base64。
+    # 有有效图片即是该只读采集动作的业务成功，不依赖 HTTP 200 单独判定。
+    if action == "screenshot" and int(normalized.get("code", 500)) == 200 and data.get("image_base64"):
+        data.setdefault("base64", data["image_base64"])
+        data["success"] = True
+        normalized["readback"] = {
+            "kind": "screenshot",
+            "image_bytes_base64": len(str(data["image_base64"])),
+            "channel": data.get("channel") or "agent_internal_screenshot",
+        }
+
+    normalized["data"] = data
+    normalized["trace_id"] = trace_id
+    normalized["raw_rpc_receipt"] = raw
+    return normalized
+
+
+def _timeout_result(device_id: str, timeout: int) -> Dict[str, Any]:
+    """统一 Fleet 超时回执，保留可重试元数据。"""
+    trace_id = uuid.uuid4().hex
+    receipt = {
+        "operation": "fleet_execute",
+        "device_id": device_id,
+        "channel": "websocket/timeout",
+        "trace_id": trace_id,
+    }
+    return {
+        "device_id": device_id,
+        "trace_id": trace_id,
+        "channel_used": "websocket/timeout",
+        "raw_rpc_receipt": receipt,
+        "readback": None,
+        "success": False,
+        "result": {
+            "code": 504,
+            "success": False,
+            "error_code": "timeout",
+            "error_message": f"设备执行超过 {timeout}s 超时",
+            "retryable": True,
+            "retry_after_seconds": 1,
+            "channel_used": "websocket/timeout",
+            "trace_id": trace_id,
+            "raw_rpc_receipt": receipt,
+            "readback": None,
+        },
+    }
+
+
+class FleetOperationReceipt(BaseModel):
+    """Fleet 统一离线/失败回执。"""
+    code: int
+    success: bool
+    data: dict = Field(default_factory=dict)
+    error_code: Optional[str] = None
+    error_message: Optional[str] = None
+    retryable: bool = False
+    trace_id: Optional[str] = None
+    channel_used: str
+    raw_rpc_receipt: Optional[dict] = None
+    readback: Optional[dict] = None
+
+
+def _fleet_offline_receipt(req: "FleetExecuteRequest") -> JSONResponse:
+    """Fleet 无 WSS 目标时返回结构化 503，不转 ADB 或本机通道。"""
+    trace_id = req.trace_id or uuid.uuid4().hex
+    receipt = {
+        "operation": "fleet_execute",
+        "device_ids": req.device_ids or [],
+        "project_id": req.project_id,
+        "channel": "websocket/offline",
+        "trace_id": trace_id,
+    }
+    return JSONResponse(
+        status_code=503,
+        content={
+            "code": 503,
+            "success": False,
+            "data": {"action": req.action, "target_count": 0, "results": []},
+            "error_code": "device_offline",
+            "error_message": "没有匹配的 WSS 在线设备",
+            "retryable": True,
+            "trace_id": trace_id,
+            "channel_used": "websocket/offline",
+            "raw_rpc_receipt": receipt,
+            "readback": None,
+        },
+    )
+
+
+class FleetExecuteRequest(BaseModel):
+    """批量执行请求。"""
+
+    device_ids: Optional[List[str]] = None
+    project_id: Optional[str] = None
+    all_online: bool = False
+    platform: str = "wechat"
+    action: str
+    params: Dict[str, Any] = Field(default_factory=dict)
+    hook_only: bool = False
+    timeout: int = 60
+    max_concurrency: int = 3
+    dry_run: bool = False
+    confirm: bool = False
+    trace_id: Optional[str] = None
+
+
+def _normalize_status(device: dict) -> str:
+    """在线态只信任当前进程中的 WSS 连接，历史登记不能冒充在线。"""
+    return "online" if ws_hub.is_online(device.get("device_id", "")) else "offline"
+
+
+async def _select_devices(
+    *,
+    device_ids: Optional[List[str]] = None,
+    project_id: Optional[str] = None,
+    all_online: bool = False,
+    status: str = "",
+    capability: str = "",
+) -> List[dict]:
+    from routers.devices import list_ws_managed_devices
+
+    devices = await list_ws_managed_devices()
+    selected = []
+    wanted = set(device_ids or [])
+    for device in devices:
+        did = device.get("device_id", "")
+        if wanted and did not in wanted:
+            continue
+        if project_id and str(device.get("project_id") or "") != str(project_id):
+            continue
+        if all_online and not ws_hub.is_online(did):
+            continue
+        if status and _normalize_status(device) != status:
+            continue
+        if capability and capability not in (device.get("capabilities") or []):
+            continue
+        normalized = dict(device)
+        normalized["status"] = _normalize_status(normalized)
+        normalized["online"] = ws_hub.is_online(did)
+        normalized["device_id_md5"] = normalized.get("device_id_md5") or device_id_md5(did)
+        selected.append(normalized)
+    return selected
+
+
+@router.get("/fleet/summary", response_model=dict, responses={503: {"model": FleetOperationReceipt}})
+async def fleet_summary(project_id: str = ""):
+    """所有手机/项目手机总览。"""
+    devices = await _select_devices(project_id=project_id or None)
+    online = [d for d in devices if d.get("online")]
+    offline = [d for d in devices if not d.get("online")]
+    by_project: Dict[str, int] = {}
+    for device in devices:
+        pid = str(device.get("project_id") or "")
+        by_project[pid] = by_project.get(pid, 0) + 1
+    return {
+        "code": 200,
+        "data": {
+            "total": len(devices),
+            "online": len(online),
+            "adb": 0,
+            "offline": len(offline),
+            "by_project": by_project,
+            "device_ids": [d.get("device_id") for d in devices],
+            "online_device_ids": [d.get("device_id") for d in online],
+        },
+    }
+
+
+@router.get("/fleet/devices", response_model=dict, responses={503: {"model": FleetOperationReceipt}})
+async def fleet_devices(
+    project_id: str = "",
+    status: str = "",
+    capability: str = "",
+    online_only: bool = False,
+):
+    """按项目、状态、能力筛选设备列表。"""
+    devices = await _select_devices(
+        project_id=project_id or None,
+        status=status,
+        capability=capability,
+        all_online=online_only,
+    )
+    return {"code": 200, "data": {"devices": devices, "count": len(devices)}}
+
+
+@router.post("/fleet/execute", response_model=dict, responses={503: {"model": FleetOperationReceipt}})
+async def fleet_execute(req: FleetExecuteRequest):
+    """在多台在线手机上批量执行同一个设备动作。"""
+    batch_trace_id = req.trace_id or uuid.uuid4().hex
+    devices = await _select_devices(
+        device_ids=req.device_ids,
+        project_id=req.project_id,
+        all_online=req.all_online or not req.device_ids,
+        status="online",
+    )
+    if not devices:
+        return _fleet_offline_receipt(req)
+
+    is_write = req.action in WRITE_ACTIONS
+    if is_write and not (req.dry_run or req.confirm):
+        return {
+            "code": 200,
+            "success": False,
+            "trace_id": batch_trace_id,
+            "channel_used": "fleet/confirm_required",
+            "raw_rpc_receipt": {
+                "operation": "fleet_execute",
+                "action": req.action,
+                "confirm_required": True,
+                "trace_id": batch_trace_id,
+            },
+            "readback": None,
+            "data": {
+                "success": False,
+                "confirm_required": True,
+                "reason": "批量写类动作需要 confirm=true；可先 dry_run=true 查看目标设备",
+                "action": req.action,
+                "target_count": len(devices),
+                "targets": [d.get("device_id") for d in devices],
+            },
+        }
+    if req.dry_run:
+        return {
+            "code": 200,
+            "success": True,
+            "trace_id": batch_trace_id,
+            "channel_used": "fleet/dry_run",
+            "raw_rpc_receipt": {
+                "operation": "fleet_execute",
+                "action": req.action,
+                "dry_run": True,
+                "executed": False,
+                "trace_id": batch_trace_id,
+            },
+            "readback": None,
+            "data": {
+                "success": True,
+                "dry_run": True,
+                "executed": False,
+                "action": req.action,
+                "target_count": len(devices),
+                "targets": [d.get("device_id") for d in devices],
+            },
+        }
+
+    sem = asyncio.Semaphore(max(1, min(int(req.max_concurrency or 1), 10)))
+    timeout = max(5, min(int(req.timeout or 60), 300))
+
+    async def run_one(device: dict) -> dict:
+        did = device.get("device_id", "")
+        async with sem:
+            try:
+                # Fleet 与单设备微信路由共用 unified + device_transport，
+                # 确保 WS/Hook 选路、hook_only、错误码和真实回执口径一致。
+                from routers.unified import _execute_skill
+
+                result = await _execute_skill(
+                    did,
+                    req.platform,
+                    req.action,
+                    req.params or {},
+                    timeout=timeout,
+                    # 微信默认固定走 WSS Agent → 手机本机 Frida RPC；调用方
+                    # 仅在非微信平台时可维持原有 generic Agent 语义。
+                    hook_only=req.hook_only or req.platform == "wechat",
+                )
+                result = _normalize_agent_result(req.action, result)
+                return {
+                    "device_id": did,
+                    "device_id_md5": device_id_md5(did),
+                    "success": _result_success(result),
+                    "trace_id": result.get("trace_id"),
+                    "channel_used": result.get("channel_used") or result.get("_channel_used") or "websocket/agent",
+                    "raw_rpc_receipt": result.get("raw_rpc_receipt"),
+                    "readback": result.get("readback") or result.get("db_readback"),
+                    "result": result,
+                }
+            except asyncio.TimeoutError:
+                return _timeout_result(did, timeout)
+            except Exception as exc:
+                trace_id = uuid.uuid4().hex
+                receipt = {
+                    "operation": "fleet_execute",
+                    "device_id": did,
+                    "channel": "websocket/error",
+                    "exception": type(exc).__name__,
+                    "trace_id": trace_id,
+                }
+                return {
+                    "device_id": did,
+                    "device_id_md5": device_id_md5(did),
+                    "success": False,
+                    "trace_id": trace_id,
+                    "channel_used": "websocket/error",
+                    "raw_rpc_receipt": receipt,
+                    "readback": None,
+                    "result": {
+                        "code": 503,
+                        "success": False,
+                        "error_code": "execution_failed",
+                        "error_message": str(exc),
+                        "retryable": True,
+                        "retry_after_seconds": 1,
+                        "channel_used": "websocket/error",
+                        "trace_id": trace_id,
+                        "raw_rpc_receipt": receipt,
+                        "readback": None,
+                    },
+                }
+
+    results = await asyncio.gather(*(run_one(device) for device in devices))
+    ok = sum(1 for item in results if item.get("success"))
+    body = {
+        "code": 200,
+        "success": ok == len(results),
+        "trace_id": batch_trace_id,
+        "channel_used": "frida_rpc" if req.platform == "wechat" else "websocket/agent",
+        "raw_rpc_receipt": {
+            "operation": "fleet_execute",
+            "action": req.action,
+            "trace_id": batch_trace_id,
+            "results": [item.get("raw_rpc_receipt") for item in results],
+        },
+        "readback": [item.get("readback") for item in results],
+        "data": {
+            "success": ok == len(results),
+            "action": req.action,
+            "target_count": len(results),
+            "success_count": ok,
+            "failed_count": len(results) - ok,
+            "results": results,
+        },
+    }
+    if ok == 0 and results and all(int(item.get("result", {}).get("code", 200)) >= 500 for item in results):
+        trace_id = batch_trace_id
+        body.update({
+            "code": 503,
+            "error_code": "fleet_execution_failed",
+            "error_message": "所有 WSS Agent 执行目标失败",
+            "retryable": True,
+            "trace_id": trace_id,
+            "channel_used": "websocket/agent",
+            "raw_rpc_receipt": {"operation": "fleet_execute", "trace_id": trace_id, "results": results},
+            "readback": None,
+        })
+        return JSONResponse(status_code=503, content=body)
+    return body
diff --git a/sdk/app/services/device_console.py b/sdk/app/services/device_console.py
new file mode 100644
index 0000000000..bbeb2a86b0
--- /dev/null
+++ b/sdk/app/services/device_console.py
@@ -0,0 +1,260 @@
+"""设备详情、微信摘要、删除和单机命令的专用服务。"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+import uuid
+
+from services.device_manager import device_manager
+from services.device_transport import device_transport
+from services.device_id_util import sanitize_sensitive_fields
+from services.ws_hub import ws_hub
+from services.workbench import (
+    ControlPlaneStorageUnavailable,
+    IdempotencyConflict,
+    control_plane_store,
+    now_iso,
+    record_audit,
+    reserve_idempotency,
+    save_idempotency_result,
+)
+
+
+class DeviceOnlineError(RuntimeError):
+    def __init__(self, result: dict):
+        self.result = result
+
+
+class DeviceDeleteConfirmationRequired(RuntimeError):
+    def __init__(self, result: dict):
+        self.result = result
+
+
+class DeviceOfflineError(RuntimeError):
+    def __init__(self, result: dict):
+        self.result = result
+
+
+class DeviceCommandVerificationError(RuntimeError):
+    def __init__(self, result: dict):
+        self.result = result
+
+
+class DeviceConsoleService:
+    def __init__(self, store: Any = control_plane_store) -> None:
+        self.store = store
+
+    async def _log_command_history(self, *, device_id: str, action: str, payload: dict, result: dict) -> None:
+        """保存命令历史；只留下脱敏后的请求和结果必要字段。"""
+        history_result = sanitize_sensitive_fields({
+            "command_id": result.get("command_id"),
+            "status": result.get("status"),
+            "trace_id": result.get("trace_id"),
+            "readback": result.get("readback"),
+            "raw_receipt": result.get("raw_receipt"),
+            "verified": result.get("status") == "succeeded",
+        })
+        await device_manager.log_command(
+            device_id,
+            action,
+            sanitize_sensitive_fields(payload),
+            history_result,
+        )
+
+    async def _lookup(self, device_id: str) -> Optional[dict]:
+        online = ws_hub.get_device_info(device_id)
+        stored = None
+        try:
+            stored = await device_manager.get_device(device_id)
+        except Exception:
+            stored = None
+        if not online and not stored:
+            return None
+        merged = {**(stored or {}), **(online or {})}
+        merged["device_id"] = device_id
+        # Mongo 仅保存历史登记；控制台“在线”只由本进程 WSS 会话判定。
+        is_online = ws_hub.is_online(device_id)
+        merged["status"] = "online" if is_online else "offline"
+        merged["connection_type"] = "websocket" if is_online else "offline"
+        return merged
+
+    async def detail(self, device_id: str) -> Optional[dict]:
+        item = await self._lookup(device_id)
+        if item is None:
+            return None
+        return sanitize_sensitive_fields({
+            "device_id": device_id,
+            "name": item.get("name") or item.get("model") or device_id,
+            "status": item.get("status", "offline"),
+            "project_id": item.get("project_id"),
+            "groups": await self._groups_for(device_id),
+            "capabilities": item.get("capabilities") or [],
+            "connection": {
+                "type": item.get("connection_type", "offline"),
+                "online": item.get("status") == "online",
+                "last_heartbeat": item.get("last_heartbeat"),
+                "source": "wss_heartbeat" if item.get("status") == "online" else "mongo.devices",
+            },
+            "version": {
+                "agent_version": item.get("agent_version"),
+                "wechat_version": self._wechat_value(item, "wechat_version"),
+                "android_version": item.get("android_version"),
+                "source_at": self._source_at(item),
+            },
+            "wechat": await self.wechat_summary(device_id, item=item),
+            "health": {
+                "battery_level": item.get("battery_level"),
+                "memory_usage_pct": item.get("memory_usage_pct"),
+                "storage_free_mb": item.get("storage_free_mb"),
+                "last_heartbeat": item.get("last_heartbeat"),
+                "source_at": self._source_at(item),
+            },
+            "source_at": self._source_at(item),
+        })
+
+    async def wechat_summary(self, device_id: str, *, item: Optional[dict] = None) -> dict:
+        item = item or await self._lookup(device_id) or {}
+        quick = item.get("quick_status") or item.get("last_status") or {}
+        logged_in = item.get("wechat_logged_in")
+        if logged_in is None:
+            logged_in = quick.get("logged_in", quick.get("wechat_logged_in"))
+        wechat_id = item.get("wechat_id") or item.get("wxid") or quick.get("wechat_id") or quick.get("wxid")
+        friend_count = item.get("friend_count")
+        if friend_count is None:
+            friend_count = quick.get("friend_count")
+        version = item.get("wechat_version") or quick.get("wechat_version")
+        source_at = item.get("wechat_source_at") or quick.get("source_at") or self._source_at(item)
+        return sanitize_sensitive_fields({
+            "device_id": device_id,
+            "logged_in": logged_in if isinstance(logged_in, bool) else None,
+            "wechat_id": wechat_id,
+            "friend_count": friend_count if isinstance(friend_count, int) else None,
+            "wechat_version": version,
+            "source_at": source_at,
+            "state": "ready" if logged_in is True and wechat_id else ("not_bound" if logged_in is False else "not_collected"),
+        })
+
+    async def delete_device(self, *, device_id: str, actor: str, key: str, trace_id: str, confirm: bool = False) -> dict:
+        payload = {"device_id": device_id, "confirm": bool(confirm)}
+        item = await self._lookup(device_id)
+
+        # 在线设备必须先拦截，不能因为控制面存储暂不可用而落入删除分支。
+        if ws_hub.is_online(device_id):
+            result = {
+                "deleted": False,
+                "device_id": device_id,
+                "trace_id": trace_id,
+                "protection": "online_device_delete_blocked",
+                "phone_command_increment": 0,
+            }
+            try:
+                idem = await reserve_idempotency(self.store, scope="device-delete", actor=actor, key=key, payload=payload)
+                if idem.replay:
+                    return idem.result or {}
+                audit = await record_audit(self.store, actor=actor, action="device.delete", resource={"device_id": device_id}, request_payload=payload, result_code="device_online", trace_id=trace_id)
+                result["audit_id"] = audit["audit_id"]
+                await save_idempotency_result(self.store, scope="device-delete", actor=actor, key=key, result=result)
+            except ControlPlaneStorageUnavailable:
+                result["audit_id"] = None
+                result["audit_persisted"] = False
+                result["audit_error"] = "control_plane_storage_unavailable"
+            raise DeviceOnlineError(result)
+
+        if item is not None and not confirm:
+            result = {
+                "deleted": False,
+                "device_id": device_id,
+                "trace_id": trace_id,
+                "precheck": True,
+                "confirmation_required": True,
+                "protection": "offline_device_delete_confirmation_required",
+                "phone_command_increment": 0,
+            }
+            audit = await record_audit(
+                self.store,
+                actor=actor,
+                action="device.delete.precheck",
+                resource={"device_id": device_id},
+                request_payload=payload,
+                result_code="confirmation_required",
+                trace_id=trace_id,
+                risk_level="high",
+            )
+            result["audit_id"] = audit["audit_id"]
+            raise DeviceDeleteConfirmationRequired(result)
+
+        idem = await reserve_idempotency(self.store, scope="device-delete", actor=actor, key=key, payload=payload)
+        if idem.replay:
+            return idem.result or {}
+        if item is None:
+            result = {"deleted": False, "device_id": device_id, "trace_id": trace_id}
+            await save_idempotency_result(self.store, scope="device-delete", actor=actor, key=key, result=result)
+            return result
+        deletion = await device_manager.delete_device(device_id)
+        if not deletion.get("deleted"):
+            raise ControlPlaneStorageUnavailable("设备目录删除未产生持久化变更")
+        audit = await record_audit(self.store, actor=actor, action="device.delete", resource={"device_id": device_id}, request_payload=payload, result_code="succeeded", trace_id=trace_id)
+        result = {
+            "deleted": True,
+            "device_id": device_id,
+            "audit_id": audit["audit_id"],
+            "trace_id": trace_id,
+            "confirmed": True,
+            "precheck": False,
+        }
+        await save_idempotency_result(self.store, scope="device-delete", actor=actor, key=key, result=result)
+        return result
+
+    async def command(self, *, device_id: str, action: str, parameters: dict, timeout_ms: int, confirm: bool, actor: str, key: str, trace_id: str) -> dict:
+        payload = {"device_id": device_id, "action": action, "parameters": parameters, "timeout_ms": timeout_ms, "confirm": confirm}
+        idem = await reserve_idempotency(self.store, scope="device-command", actor=actor, key=key, payload=payload)
+        if idem.replay:
+            return idem.result or {}
+        if not ws_hub.is_online(device_id):
+            raw = {"operation": action, "device_id": device_id, "channel": "websocket/offline", "trace_id": trace_id}
+            audit = await record_audit(self.store, actor=actor, action="device.command", resource={"device_id": device_id}, request_payload=payload, result_code="device_offline", trace_id=trace_id, channel_used="websocket/offline", raw_receipt=raw, risk_level="high" if not confirm else "medium")
+            result = {"command_id": None, "status": "device_offline", "device_id": device_id, "raw_receipt": raw, "readback": None, "audit_id": audit["audit_id"], "trace_id": trace_id}
+            await self._log_command_history(device_id=device_id, action=action, payload=payload, result=result)
+            await save_idempotency_result(self.store, scope="device-command", actor=actor, key=key, result=result)
+            raise DeviceOfflineError(result)
+        raw = await device_transport.execute_via_ws(device_id, "system", action, {**parameters, "trace_id": trace_id}, timeout=max(1, int(timeout_ms / 1000)))
+        raw = dict(raw or {})
+        data = raw.get("data") if isinstance(raw.get("data"), dict) else {}
+        readback = data.get("readback") or data.get("db_readback") or raw.get("readback")
+        command_id = raw.get("command_id") or data.get("command_id")
+        rpc_success = raw.get("code") == 200 and data.get("success") is True
+        verified = rpc_success and bool(command_id) and isinstance(readback, dict) and bool(readback.get("verified", True))
+        result = {
+            "command_id": command_id,
+            "device_id": device_id,
+            "status": "succeeded" if verified else ("rpc_succeeded" if rpc_success else "failed"),
+            "raw_receipt": raw,
+            "readback": readback,
+            "trace_id": trace_id,
+        }
+        await self._log_command_history(device_id=device_id, action=action, payload=payload, result=result)
+        audit = await record_audit(self.store, actor=actor, action="device.command", resource={"device_id": device_id, "command_id": command_id}, request_payload=payload, result_code="succeeded" if verified else "business_verification_failed", trace_id=trace_id, readback=readback, channel_used="websocket/agent", raw_receipt=raw, risk_level="high" if not confirm else "medium")
+        result["audit_id"] = audit["audit_id"]
+        await save_idempotency_result(self.store, scope="device-command", actor=actor, key=key, result=result)
+        if not verified:
+            raise DeviceCommandVerificationError(result)
+        return result
+
+    async def _groups_for(self, device_id: str) -> list[dict]:
+        try:
+            groups = await self.store.list("device_groups")
+        except ControlPlaneStorageUnavailable:
+            return []
+        return [{"group_id": item.get("group_id"), "name": item.get("name"), "version": item.get("version", 1)} for item in groups if device_id in (item.get("device_ids") or [])]
+
+    @staticmethod
+    def _wechat_value(item: dict, key: str) -> Any:
+        quick = item.get("quick_status") or item.get("last_status") or {}
+        return item.get(key) or quick.get(key)
+
+    @staticmethod
+    def _source_at(item: dict) -> Optional[str]:
+        return item.get("source_at") or item.get("updated_at") or item.get("last_heartbeat")
+
+
+device_console_service = DeviceConsoleService()