宜博 / pixelphone

"""
Android MCP Server
Exposes Android phone data to Hermes via MCP stdio protocol.

Tools:
  get_events       — poll event bus (incremental, since_id)
  get_latest       — most recent N events
  get_stats        — event queue stats
  query_wechat     — SQLCipher query on WeChat EnMicroMsg.db (requires root)
  list_event_apps  — distinct apps that have posted events
  get_ui_tree      — dump current foreground UI tree via uiautomator (requires root)
  frida_call       — inject Frida JS script into a running app (requires root + frida-server)

Usage (spawned by Hermes as MCP subprocess):
  /path/to/venv/bin/python android_mcp.py

Config via env vars:
  EVENT_BUS_DB      path to events.db  (default: ~/.hermes/event_bus/events.db)
  WECHAT_DB         path to EnMicroMsg.db (default: auto-detect)
  WECHAT_KEY        SQLCipher key (default: read from ~/.hermes/event_bus/wechat_key)
  SQLCIPHER_BIN     path to sqlcipher binary (default: auto-detect)
"""

from __future__ import annotations

import json
import logging
import os
import re
import sqlite3
import subprocess
import sys
import time
from pathlib import Path
from typing import Any, Optional

logger = logging.getLogger(__name__)


def _load_hermes_env():
    """Load ~/.hermes/.env before any module-level env reads."""
    termux_home = Path("/data/data/com.termux/files/home")
    home = termux_home if termux_home.exists() else Path(os.environ.get("HOME", str(termux_home)))
    env_file = home / ".hermes" / ".env"
    if not env_file.exists():
        return
    for raw in env_file.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        k = k.strip(); v = v.strip().strip('"').strip("'")
        if k and not os.environ.get(k):
            os.environ[k] = v

_load_hermes_env()

# ── paths ──────────────────────────────────────────────────────────────────────

_TERMUX_HOME = Path("/data/data/com.termux/files/home")
_HERMES_BASE = (_TERMUX_HOME if _TERMUX_HOME.exists()
                else Path(os.environ.get("HOME", str(_TERMUX_HOME)))) / ".hermes"
HERMES_HOME = Path(os.environ.get("HERMES_HOME", str(_HERMES_BASE)))
EVENT_BUS_DB = Path(os.environ.get("EVENT_BUS_DB", str(HERMES_HOME / "event_bus" / "events.db")))
# Directory holding the connector/persona/cloud_knowledge/memory_graph modules.
# On-device this is the deploy dir (~/.hermes/event_bus); in-repo it is the dir
# containing this script. Tools do `sys.path.insert(0, EVENT_BUS_DIR)` to import them.
EVENT_BUS_DIR = os.environ.get("EVENT_BUS_DIR", str(Path(__file__).resolve().parent))

WECHAT_DB_CANDIDATES = [
    "/data/data/com.tencent.mm/MicroMsg",  # needs root
]
WECHAT_KEY_FILE = HERMES_HOME / "event_bus" / "wechat_key"

SQLCIPHER_CANDIDATES = [
    "/data/data/com.termux/files/usr/bin/sqlcipher",
    "/usr/bin/sqlcipher",
]


# ── helpers ────────────────────────────────────────────────────────────────────

def _event_db() -> sqlite3.Connection:
    conn = sqlite3.connect(EVENT_BUS_DB)
    conn.row_factory = sqlite3.Row
    return conn


def _read_wechat_key() -> str:
    env_key = os.environ.get("WECHAT_KEY", "")
    if env_key:
        return env_key
    if WECHAT_KEY_FILE.exists():
        return WECHAT_KEY_FILE.read_text().strip()
    # fallback to known key (update via `echo <key> > ~/.hermes/event_bus/wechat_key`)
    return "e03ce81"


def _find_wechat_db() -> Optional[str]:
    env_path = os.environ.get("WECHAT_DB", "")
    if env_path and Path(env_path).exists():
        return env_path
    try:
        result = subprocess.run(
            ["su", "-c", f"find {WECHAT_DB_CANDIDATES[0]} -name 'EnMicroMsg.db' 2>/dev/null | head -1"],
            capture_output=True, text=True, timeout=5
        )
        path = result.stdout.strip()
        if path:
            return path
    except Exception:
        pass
    return None


def _find_sqlcipher() -> Optional[str]:
    for p in SQLCIPHER_CANDIDATES:
        if Path(p).exists():
            return p
    return None


def _inline_sql_param(value: Any) -> str:
    """Render a single bind value as a safe SQL literal.

    The sqlcipher CLI is driven via a piped script and does not bind '?'
    placeholders, so callers that need parameters must inline them here.
    Strings are single-quote escaped; numbers are emitted bare; bools/None map
    to SQLite literals. Anything else is coerced to a quoted string.
    """
    if value is None:
        return "NULL"
    if isinstance(value, bool):
        return "1" if value else "0"
    if isinstance(value, int):
        return str(value)
    if isinstance(value, float):
        return repr(value)
    text = str(value).replace("'", "''")
    return f"'{text}'"


def _bind_sql(sql: str, params: Optional[list[Any]]) -> str:
    """Substitute ordered '?' placeholders in *sql* with inline literals.

    Only positional '?' markers are supported (matching sqlite3 style). Raises
    ValueError if the placeholder count does not match the params length.
    """
    if not params:
        if "?" in sql:
            raise ValueError("SQL has '?' placeholders but no params were provided")
        return sql
    out: list[str] = []
    it = iter(params)
    placeholders = 0
    for ch in sql:
        if ch == "?":
            placeholders += 1
            try:
                out.append(_inline_sql_param(next(it)))
            except StopIteration:
                raise ValueError("more '?' placeholders than params")
        else:
            out.append(ch)
    remaining = list(it)
    if remaining:
        raise ValueError(f"more params ({placeholders + len(remaining)}) than '?' placeholders ({placeholders})")
    return "".join(out)


def _run_sqlcipher(sql: str, db_path: str, key: str, params: Optional[list[Any]] = None) -> dict[str, Any]:
    sqlcipher = _find_sqlcipher()
    if not sqlcipher:
        return {"error": "sqlcipher binary not found", "candidates": SQLCIPHER_CANDIDATES}

    try:
        bound_sql = _bind_sql(sql, params)
    except ValueError as e:
        return {"error": f"parameter binding failed: {e}", "sql": sql}

    script = (
        f'PRAGMA key="{key}";\n'
        f'PRAGMA cipher_compatibility=1;\n'
        f'PRAGMA cipher_page_size=1024;\n'
        f'.mode json\n'
        f'{bound_sql.rstrip(";").strip()};\n'
    )
    try:
        result = subprocess.run(
            ["su", "-c", f'{sqlcipher} "{db_path}"'],
            input=script,
            capture_output=True, text=True, timeout=15
        )
        if result.returncode != 0:
            return {"error": result.stderr.strip() or "sqlcipher error", "sql": sql}

        # Parse JSON output (.mode json). PRAGMA key/cipher_* run before
        # `.mode json` takes effect, so sqlcipher emits one or more "ok" lines
        # ahead of the JSON payload — strip everything before the first '['/'{'.
        raw = result.stdout.strip()
        if not raw:
            return {"rows": [], "count": 0}
        starts = [i for i in (raw.find("["), raw.find("{")) if i != -1]
        if not starts:
            # No JSON body at all (e.g. only "ok" from the pragmas) => empty set.
            return {"rows": [], "count": 0}
        payload = raw[min(starts):]
        try:
            rows = json.loads(payload)
            if not isinstance(rows, list):
                rows = [rows]
            return {"rows": rows, "count": len(rows)}
        except json.JSONDecodeError:
            # Fall back to raw output for inspection.
            return {"raw": raw}

    except subprocess.TimeoutExpired:
        return {"error": "query timed out (15s)"}
    except Exception as e:
        return {"error": str(e)}


# ── MCP server ─────────────────────────────────────────────────────────────────

def _build_server():
    from mcp.server.fastmcp import FastMCP

    mcp = FastMCP(
        "android-phone",
        instructions=(
            "Provides real-time data from the Android phone: "
            "system events (notifications, UI, clipboard), "
            "WeChat message history, and event queue stats. "
            "Use get_events(since_id=<last_seen_id>) for incremental polling. "
            "Use query_wechat(sql=...) for structured WeChat data."
        ),
    )

    # ── tool: get_events ──────────────────────────────────────────────────────

    @mcp.tool()
    def get_events(
        since_id: int = 0,
        limit: int = 50,
        type_filter: str = "",
        app_filter: str = "",
        source_filter: str = "",
    ) -> str:
        """
        Poll the Android event bus for new events.

        Returns events with id > since_id. Store the last returned id and
        pass it back next call to get only new events (incremental polling).

        Args:
            since_id: Return only events with id greater than this value (default 0 = all).
            limit: Maximum number of events to return (max 500, default 50).
            type_filter: Filter by event type, e.g. 'notification', 'ui_event', 'clipboard'.
            app_filter: Filter by Android package name, e.g. 'com.tencent.mm'.
            source_filter: Filter by event source: 'apk' (real-time from APK) or 'monitor' (dumpsys polling).
        """
        limit = min(max(1, limit), 500)
        sql = "SELECT id, type, app, data, ts, source FROM events WHERE id > ?"
        params: list[Any] = [since_id]
        if type_filter:
            sql += " AND type = ?"
            params.append(type_filter)
        if app_filter:
            sql += " AND app = ?"
            params.append(app_filter)
        if source_filter:
            sql += " AND source = ?"
            params.append(source_filter)
        sql += " ORDER BY id ASC LIMIT ?"
        params.append(limit)

        try:
            with _event_db() as conn:
                rows = conn.execute(sql, params).fetchall()
            events = [
                {"id": r["id"], "type": r["type"], "app": r["app"],
                 "data": json.loads(r["data"]), "ts": r["ts"],
                 "source": r["source"] or "monitor"}
                for r in rows
            ]
            result = {
                "events": events,
                "count": len(events),
                "next_since_id": events[-1]["id"] if events else since_id,
                "has_more": len(events) == limit,
            }
        except Exception as e:
            result = {"error": str(e), "events": [], "count": 0}

        return json.dumps(result, ensure_ascii=False)

    # ── tool: get_latest ──────────────────────────────────────────────────────

    @mcp.tool()
    def get_latest(
        limit: int = 20,
        type_filter: str = "",
        app_filter: str = "",
        source_filter: str = "",
    ) -> str:
        """
        Get the most recent events from the Android event bus.

        Useful for getting context on what's happening on the phone right now,
        without needing to track since_id.

        Args:
            limit: Number of recent events to return (max 200, default 20).
            type_filter: Filter by event type, e.g. 'notification', 'ui_event'.
            app_filter: Filter by Android package name, e.g. 'com.tencent.mm'.
            source_filter: Filter by source: 'apk' (real-time) or 'monitor' (dumpsys).
        """
        limit = min(max(1, limit), 200)
        sql = "SELECT id, type, app, data, ts, source FROM events WHERE 1=1"
        params: list[Any] = []
        if type_filter:
            sql += " AND type = ?"
            params.append(type_filter)
        if app_filter:
            sql += " AND app = ?"
            params.append(app_filter)
        if source_filter:
            sql += " AND source = ?"
            params.append(source_filter)
        sql += " ORDER BY id DESC LIMIT ?"
        params.append(limit)

        try:
            with _event_db() as conn:
                rows = conn.execute(sql, params).fetchall()
            events = [
                {"id": r["id"], "type": r["type"], "app": r["app"],
                 "data": json.loads(r["data"]), "ts": r["ts"],
                 "source": r["source"] or "monitor"}
                for r in reversed(rows)
            ]
            result = {"events": events, "count": len(events)}
        except Exception as e:
            result = {"error": str(e), "events": [], "count": 0}

        return json.dumps(result, ensure_ascii=False)

    # ── tool: get_stats ───────────────────────────────────────────────────────

    @mcp.tool()
    def get_stats() -> str:
        """
        Get Android event bus statistics: total event count, latest event,
        database path, and available event types.
        """
        try:
            with _event_db() as conn:
                total = conn.execute("SELECT COUNT(*) FROM events").fetchone()[0]
                last = conn.execute(
                    "SELECT id, type, app, ts FROM events ORDER BY id DESC LIMIT 1"
                ).fetchone()
                types = conn.execute(
                    "SELECT type, COUNT(*) as cnt FROM events GROUP BY type ORDER BY cnt DESC"
                ).fetchall()
            result = {
                "total_events": total,
                "last_event_id": last["id"] if last else None,
                "last_event_type": last["type"] if last else None,
                "last_event_app": last["app"] if last else None,
                "last_event_ts": last["ts"] if last else None,
                "event_types": [{"type": r["type"], "count": r["cnt"]} for r in types],
                "db_path": str(EVENT_BUS_DB),
            }
        except Exception as e:
            result = {"error": str(e)}
        return json.dumps(result, ensure_ascii=False)

    # ── tool: list_event_apps ─────────────────────────────────────────────────

    @mcp.tool()
    def list_event_apps() -> str:
        """
        List all Android apps that have posted events to the bus,
        with event counts per app. Useful for knowing what's active.
        """
        try:
            with _event_db() as conn:
                rows = conn.execute(
                    "SELECT app, COUNT(*) as cnt FROM events WHERE app IS NOT NULL "
                    "GROUP BY app ORDER BY cnt DESC"
                ).fetchall()
            result = {"apps": [{"app": r["app"], "count": r["cnt"]} for r in rows]}
        except Exception as e:
            result = {"error": str(e)}
        return json.dumps(result, ensure_ascii=False)

    # ── tool: get_ui_tree ─────────────────────────────────────────────────────

    @mcp.tool()
    def get_ui_tree(package: str = "") -> str:
        """
        Dump the current foreground app's UI tree via uiautomator.
        Returns XML with all visible UI nodes (text, resource-id, bounds, etc.).
        Requires root. Use this to understand what's on screen right now.

        Args:
            package: Optional package name filter. If provided, only nodes from
                     this app are returned (e.g. 'com.tencent.mm').
        """
        tmp = "/sdcard/agent_ui_dump.xml"
        try:
            result = subprocess.run(
                ["su", "-c", f"uiautomator dump {tmp} 2>/dev/null && cat {tmp}"],
                capture_output=True, text=True, timeout=10
            )
            xml = result.stdout.strip()
            if not xml:
                return json.dumps({"error": "uiautomator dump returned empty output",
                                   "stderr": result.stderr[:200]})
            # Strip the "UI hierchary dumped to..." line if present
            if xml.startswith("UI hierchary") or xml.startswith("UI hierarchy"):
                xml = xml.split("\n", 1)[-1].strip()
            if package:
                # Quick filter: return only if package appears in tree
                if package not in xml:
                    return json.dumps({"error": f"package {package!r} not found in UI tree",
                                       "hint": "App may not be in foreground"})
            return json.dumps({"xml": xml, "length": len(xml)}, ensure_ascii=False)
        except subprocess.TimeoutExpired:
            return json.dumps({"error": "uiautomator timed out (10s)"})
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── tool: frida_call ──────────────────────────────────────────────────────

    @mcp.tool()
    def frida_call(
        package: str,
        script: str,
        timeout: int = 10,
    ) -> str:
        """
        Inject a Frida JavaScript script into a running Android app and return
        the output. Requires root and frida-server running on device.

        The script should call send() to return data. Example:
          script="Java.perform(function(){ send(Java.use('android.os.Build').MODEL); })"

        Args:
            package: Target app package name, e.g. 'com.tencent.mm'.
            script:  Frida JavaScript to inject. Use send() to return values.
            timeout: Max seconds to wait for script output (default 10).
        """
        # Delegates to frida_bridge.py on Mac via ADB reverse tcp:8767 tcp:8767
        # The bridge uses frida Python (17.9.10) to connect to frida-server on device via USB
        BRIDGE_URL = "http://127.0.0.1:8767/frida_call"
        import urllib.request
        payload = json.dumps({"package": package, "script": script, "timeout": timeout}).encode()
        try:
            req = urllib.request.Request(BRIDGE_URL, data=payload,
                                         headers={"Content-Type": "application/json"})
            with urllib.request.urlopen(req, timeout=timeout + 10) as resp:
                return resp.read().decode()
        except Exception as e:
            return json.dumps({"error": f"frida_bridge unreachable: {e}. Ensure frida_bridge.py is running on Mac and adb reverse tcp:8767 tcp:8767 is set."})

    # ── tool: query_wechat ────────────────────────────────────────────────────

    @mcp.tool()
    def query_wechat(
        sql: str,
        key: str = "",
        db_path: str = "",
    ) -> str:
        """
        Run a read-only SQL query on WeChat's EnMicroMsg.db (SQLCipher encrypted).
        Requires root. Returns JSON rows.

        Useful tables:
          message  — columns: msgId, type, isSend, talker, content, createTime
          rcontact — columns: conRemark, nickname, username
          chatroom — columns: displayname, chatroomname

        Example:
          sql="SELECT talker, content, createTime FROM message ORDER BY createTime DESC LIMIT 20"

        Args:
            sql: SQL SELECT query to run.
            key: SQLCipher key (optional, reads from ~/.hermes/event_bus/wechat_key if omitted).
            db_path: Path to EnMicroMsg.db (optional, auto-detected if omitted).
        """
        # Safety: only allow SELECT
        stripped = sql.strip().upper()
        if not stripped.startswith("SELECT") and not stripped.startswith("PRAGMA"):
            return json.dumps({"error": "Only SELECT and PRAGMA queries are allowed."})

        effective_key = key or _read_wechat_key()
        effective_db = db_path or _find_wechat_db()
        if not effective_db:
            return json.dumps({"error": "WeChat DB not found. Set WECHAT_DB env or pass db_path."})

        return json.dumps(_run_sqlcipher(sql, effective_db, effective_key), ensure_ascii=False)

    # ── tool: get_wechat_messages ─────────────────────────────────────────────

    @mcp.tool()
    def get_wechat_messages(
        contact: str = "",
        limit: int = 20,
        since_ts: int = 0,
        include_groups: bool = False,
    ) -> str:
        """
        Get WeChat messages from the local SQLite database (requires root).

        Returns messages from the message table with sender, content, and timestamp.
        Optionally filter by contact wxid/username, or restrict to direct messages only.

        Args:
            contact: Filter by talker wxid, e.g. 'wxid_abc123' or '@chatroom' suffix for groups.
                     Leave empty to return recent messages from all chats.
            limit: Number of messages to return (default 20, max 200).
            since_ts: Only return messages after this Unix timestamp in milliseconds.
            include_groups: If False (default), exclude group chat messages (talker ending in @chatroom).
        """
        limit = min(max(1, limit), 200)
        sql_parts = ["SELECT msgId, type, isSend, talker, content, createTime FROM message WHERE 1=1"]
        params: list[Any] = []

        if contact:
            sql_parts.append("AND talker = ?")
            params.append(contact)
        elif not include_groups:
            sql_parts.append("AND talker NOT LIKE '%@chatroom'")

        if since_ts:
            sql_parts.append("AND createTime > ?")
            params.append(since_ts)

        sql_parts.append("ORDER BY createTime DESC LIMIT ?")
        params.append(limit)

        sql = " ".join(sql_parts)
        key = _read_wechat_key()
        db = _find_wechat_db()
        if not db:
            return json.dumps({"error": "WeChat DB not found."})

        result = _run_sqlcipher(sql, db, key, params)
        if "rows" in result:
            # Reverse to chronological order
            result["rows"] = list(reversed(result["rows"]))
        return json.dumps(result, ensure_ascii=False)

    # ── tool: get_wechat_contacts ─────────────────────────────────────────────

    @mcp.tool()
    def get_wechat_contacts(
        query: str = "",
        limit: int = 50,
    ) -> str:
        """
        Search WeChat contacts from the local SQLite database (requires root).

        Returns contacts with wxid, nickname, remark name, and type info.

        Args:
            query: Optional search string matched against nickname, remark (conRemark), or username (wxid).
                   Leave empty to return all contacts.
            limit: Maximum number of contacts to return (default 50, max 500).
        """
        limit = min(max(1, limit), 500)

        # Columns verified against real WeChat 8.x rcontact schema (no imgFlag).
        if query:
            sql = (
                "SELECT username, alias, nickname, conRemark, type "
                "FROM rcontact "
                "WHERE (nickname LIKE ? OR conRemark LIKE ? OR username LIKE ?) "
                "AND type != 0 "
                "ORDER BY conRemark, nickname LIMIT ?"
            )
            like = f"%{query}%"
            params: list[Any] = [like, like, like, limit]
        else:
            sql = (
                "SELECT username, alias, nickname, conRemark, type "
                "FROM rcontact WHERE type != 0 "
                "ORDER BY conRemark, nickname LIMIT ?"
            )
            params = [limit]

        key = _read_wechat_key()
        db = _find_wechat_db()
        if not db:
            return json.dumps({"error": "WeChat DB not found."})

        return json.dumps(_run_sqlcipher(sql, db, key, params), ensure_ascii=False)

    # ── tool: get_wechat_chatrooms ────────────────────────────────────────────

    @mcp.tool()
    def get_wechat_chatrooms(limit: int = 30) -> str:
        """
        List WeChat group chats (chatrooms) with display name and member count.
        Requires root.

        Args:
            limit: Maximum number of chatrooms to return (default 30, max 200).
        """
        limit = min(max(1, limit), 200)
        # Columns verified against real WeChat 8.x chatroom schema
        # (memberCount/roomowner, not membercount/chatroomowner).
        sql = (
            "SELECT chatroomname, displayname, memberCount, roomowner "
            "FROM chatroom ORDER BY memberCount DESC LIMIT ?"
        )
        key = _read_wechat_key()
        db = _find_wechat_db()
        if not db:
            return json.dumps({"error": "WeChat DB not found."})
        return json.dumps(_run_sqlcipher(sql, db, key, [limit]), ensure_ascii=False)

    # ── tool: search_memory ───────────────────────────────────────────────────

    @mcp.tool()
    def search_memory(
        query: str,
        person: str = "",
        since: int = 0,
        until: int = 0,
        source_app: str = "",
        limit: int = 20,
    ) -> str:
        """
        Search the personal Memory Graph for messages and entities.

        Uses FTS5 keyword search with fallback to LIKE for CJK text.

        Args:
            query:      Keyword or phrase to search for.
            person:     Filter by contact name/wxid (optional).
            since:      Unix timestamp lower bound (optional).
            until:      Unix timestamp upper bound (optional).
            source_app: Filter by app, e.g. 'wechat' or 'telegram' (optional).
            limit:      Max results to return (default 20, max 100).
        """
        try:
            import sys
            _base = os.path.dirname(__file__)
            if _base not in sys.path:
                sys.path.insert(0, _base)
            from memory_graph import search_memory as _search, get_stats
            limit = min(max(1, limit), 100)
            results = _search(query, person=person, since=since, until=until,
                              source_app=source_app, limit=limit)
            return json.dumps({
                "results": results,
                "count": len(results),
                "stats": get_stats(),
            }, ensure_ascii=False, default=str)
        except ImportError:
            return json.dumps({"error": "memory_graph.py not found in event_bus dir"})
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def ingest_wechat_to_memory(limit: int = 2000) -> str:
        """
        Bulk-ingest WeChat contacts and text messages into the Memory Graph.
        Run this once to populate the graph, then use search_memory to query.

        Args:
            limit: Max number of recent text messages to ingest (default 2000).
        """
        try:
            import sys
            _base = os.path.dirname(__file__)
            if _base not in sys.path:
                sys.path.insert(0, _base)
            from memory_graph import init_db, ingest_wechat_contacts, ingest_wechat_messages, get_stats

            init_db()
            key = _read_wechat_key()
            db = _find_wechat_db()
            if not db:
                return json.dumps({"error": "WeChat DB not found"})

            # Contacts. _run_sqlcipher returns {"rows": [...], "count": N},
            # so extract the rows list before handing off to the ingester.
            contacts_sql = "SELECT username, nickname, conRemark, type FROM rcontact WHERE type != 0 AND username != '' LIMIT 1000"
            contacts = _run_sqlcipher(contacts_sql, db, key)
            c_rows = contacts.get("rows", []) if isinstance(contacts, dict) else []
            c_result = ingest_wechat_contacts(c_rows)

            # Text messages only (type=1)
            limit = min(max(1, limit), 5000)
            msgs_sql = (
                "SELECT msgId, type, isSend, talker, content, createTime "
                "FROM message WHERE type=1 AND content != '' "
                "AND content NOT LIKE '%<msg>%' "
                "ORDER BY createTime DESC LIMIT ?"
            )
            messages = _run_sqlcipher(msgs_sql, db, key, [limit])
            m_rows = messages.get("rows", []) if isinstance(messages, dict) else []
            m_result = ingest_wechat_messages(m_rows)

            if contacts.get("error") or messages.get("error"):
                return json.dumps({
                    "error": contacts.get("error") or messages.get("error"),
                    "contacts": c_result,
                    "messages": m_result,
                }, ensure_ascii=False)

            return json.dumps({
                "contacts": c_result,
                "messages": m_result,
                "stats": get_stats(),
            }, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── tool: semantic_search ─────────────────────────────────────────────────

    @mcp.tool()
    def semantic_search_memory(query: str, limit: int = 10) -> str:
        """
        Semantic similarity search over Memory Graph using Kimi embeddings.
        Finds messages semantically related to the query, even without exact keywords.

        Args:
            query: Natural language query (Chinese/English).
            limit: Max results to return (default 10).
        """
        try:
            import sys
            _base = os.path.dirname(__file__)
            if _base not in sys.path:
                sys.path.insert(0, _base)
            import memory_graph as mg

            # Inject API key from env
            env_path = os.path.expanduser("~/.hermes/.env")
            if os.path.exists(env_path):
                for line in open(env_path):
                    line = line.strip()
                    if line and not line.startswith("#") and "=" in line:
                        k, v = line.split("=", 1)
                        os.environ.setdefault(k.strip(), v.strip())
            mg.KIMI_API_KEY = os.environ.get("KIMI_API_KEY", "")
            mg.KIMI_BASE_URL = os.environ.get("KIMI_BASE_URL", "https://api.kimi.com/coding/v1")

            if not mg.KIMI_API_KEY:
                return json.dumps({"error": "KIMI_API_KEY not set in ~/.hermes/.env"})

            results = mg.semantic_search(query, limit=min(max(1, limit), 50))
            return json.dumps({
                "results": results,
                "count": len(results),
                "query": query,
            }, ensure_ascii=False, default=str)
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def embed_memory_messages(limit: int = 100) -> str:
        """
        Generate Kimi embeddings for un-embedded messages in Memory Graph.
        Call this after ingest_wechat_to_memory to enable semantic search.

        Args:
            limit: Max number of messages to embed per call (default 100, max 500).
        """
        try:
            import sys
            _base = os.path.dirname(__file__)
            if _base not in sys.path:
                sys.path.insert(0, _base)
            import memory_graph as mg

            env_path = os.path.expanduser("~/.hermes/.env")
            if os.path.exists(env_path):
                for line in open(env_path):
                    line = line.strip()
                    if line and not line.startswith("#") and "=" in line:
                        k, v = line.split("=", 1)
                        os.environ.setdefault(k.strip(), v.strip())
            mg.KIMI_API_KEY = os.environ.get("KIMI_API_KEY", "")
            mg.KIMI_BASE_URL = os.environ.get("KIMI_BASE_URL", "https://api.kimi.com/coding/v1")

            result = mg.embed_messages(limit=min(max(1, limit), 500))
            stats = mg.get_stats()
            return json.dumps({"result": result, "stats": stats}, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── tool: send_wechat_message ─────────────────────────────────────────────

    @mcp.tool()
    def send_wechat_message(to: str, content: str, method: str = "auto",
                            timeout: int = 30) -> str:
        """
        Send a WeChat text message. Tries the fast internal hook first, falls
        back to UI automation if the hook is unavailable.

        Hook path (method='hook' or 'auto'): dispatches via the wechathook
        LSPosed module — no UI disruption, < 1 s, works for any wxid/alias.
        Requires wechathook module loaded and WeChat running.

        UI path (method='ui' or fallback): drives WeChat's search box with
        ADBKeyboard; verifies delivery via WeChat's message DB (isSend=1).
        Requires ADBKeyboard APK installed on device.

        Args:
            to: Contact wxid / nickname / remark, or 'filehelper' / '文件传输助手'
                for File Transfer Helper. Nicknames are resolved via WeChat DB.
            content: Message text (CJK supported).
            method: 'auto' (hook→UI fallback), 'hook' (hook only), 'ui' (UI only).
            timeout: Seconds to wait for chat window in UI mode (default 30).
        """
        import sys
        _base = os.path.dirname(os.path.abspath(__file__))
        if _base not in sys.path:
            sys.path.insert(0, _base)

        try:
            # ── Resolve nickname → wxid (needed for hook path) ───────────────
            wxid = to.strip()
            if method != "ui":
                from wechat_send import _resolve_contact
                resolved = _resolve_contact(to)
                if resolved.get("resolved"):
                    wxid = resolved["talker"] or wxid

            # ── Hook path ────────────────────────────────────────────────────
            if method in ("hook", "auto"):
                from wechat_hook_send import send_text_via_hook
                result = send_text_via_hook(wxid, content)
                if result.get("status") == "sent":
                    return json.dumps(result, ensure_ascii=False)
                if method == "hook":
                    return json.dumps(result, ensure_ascii=False)
                # auto: fall through to UI

            # ── UI path ──────────────────────────────────────────────────────
            from wechat_send import send_wechat_message_ui
            result = send_wechat_message_ui(to=to, content=content, timeout=timeout)
            return json.dumps(result, ensure_ascii=False)

        except Exception as e:
            return json.dumps({"status": "error", "error": str(e)})

    # ── tool: send_wechat_image ───────────────────────────────────────────────

    @mcp.tool()
    def send_wechat_image(to: str, image_path: str) -> str:
        """
        Send an image to a WeChat contact via the internal hook API (no UI simulation).

        Fires SEND_IMAGE_INTERNAL → WeChat's own image plugin (b80.u.oj inserts the
        message + ImgInfo + builds CDN descriptors) → CDN service (v70.w.oj) uploads
        the full image via WeChat's own CDN. Receiver sees the full-resolution
        original (not a stuck thumbnail). Fire-and-forget: delivery is async after
        the broadcast; confirm via a WeChat UI screenshot if needed.

        Preferred over the old share-intent path — no contact-picker, no wrong-contact
        risk. Verified 2026-07-07 on WeChat 8.0.74 (filehelper + real contact).

        Args:
            to: Contact wxid as stored in rcontact.username (e.g. 'filehelper',
                'wxid_xxx', 'yibooo'). Nicknames are NOT resolved. Test with
                'filehelper' first to avoid account risk.
            image_path: Absolute device path to a PNG or JPEG. Staged into /sdcard
                        automatically so root-only source paths also work.
        """
        import sys
        _base = os.path.dirname(os.path.abspath(__file__))
        if _base not in sys.path:
            sys.path.insert(0, _base)
        try:
            from wechat_hook_send import send_image_via_hook
            result = send_image_via_hook(wxid=to, image_path=image_path)
            return json.dumps(result, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"status": "error", "error": str(e)})

    # ── tool: send_wechat_voice ──────────────────────────────────────────────

    @mcp.tool()
    def send_wechat_voice(to: str, silk_path: str, duration_ms: int = 3000) -> str:
        """
        Send a voice message to a WeChat contact via the internal hook API (no UI simulation).

        Uses WeChat's w11.r0 factory (MsgType 34). The hook injects the correct
        voice file ID into the DB insert so the message is stored with the right
        SILK file reference and is immediately playable.

        Args:
            to: Contact wxid (e.g. 'filehelper'). IMPORTANT: test only with 'filehelper'.
            silk_path: Absolute device path to a SILK V3 file (Tencent variant,
                       starts with \\x02#!SILK_V3). Generate with silkv3-encoder
                       or convert from PCM via silk-v3-encoder.
            duration_ms: Voice duration in milliseconds shown in WeChat UI.
        """
        import sys
        _base = os.path.dirname(os.path.abspath(__file__))
        if _base not in sys.path:
            sys.path.insert(0, _base)
        try:
            from wechat_hook_send import send_voice_via_hook
            result = send_voice_via_hook(wxid=to, silk_path=silk_path, duration_ms=duration_ms)
            return json.dumps(result, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"status": "error", "error": str(e)})

    # ── tool: send_wechat_voice_tts ──────────────────────────────────────────

    @mcp.tool()
    def send_wechat_voice_tts(to: str, text: str, voice: str = "zh-CN-XiaoxiaoNeural") -> str:
        """
        Generate a TTS voice message and send it to a WeChat contact via the
        internal hook API. No UI simulation required.

        Full pipeline (runs on-device):
          text → edge-tts (MP3) → ffmpeg (16kHz PCM) → silk-wasm (SILK V3)
          → SEND_VOICE_INTERNAL broadcast → WeChat DB insert (status=4, delivered)

        **DO NOT implement SILK encoding yourself.** This tool handles the entire
        pipeline. Just pass the text you want spoken.

        On-device dependencies (install once in Termux if missing):
          pip install edge-tts
          pkg install ffmpeg nodejs
          npm install -g silk-wasm

        Args:
            to:    Contact wxid (e.g. 'filehelper'). IMPORTANT: test only with 'filehelper'.
            text:  Chinese text to synthesize and send as a voice message.
            voice: edge-tts voice name (default: zh-CN-XiaoxiaoNeural female).
                   Alternatives: zh-CN-YunxiNeural (male), zh-CN-YunyangNeural (news male).
        """
        import sys
        _base = os.path.dirname(os.path.abspath(__file__))
        if _base not in sys.path:
            sys.path.insert(0, _base)
        try:
            from wechat_hook_send import send_voice_tts_via_hook
            result = send_voice_tts_via_hook(wxid=to, text=text, voice=voice)
            return json.dumps(result, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"status": "error", "error": str(e)})

    # ── tool: send_message (semantic router) ─────────────────────────────────

    @mcp.tool()
    def send_message(to: str, content: str, app: str = "auto") -> str:
        """
        Send a message to a contact, automatically routing to the right app.

        Args:
            to:      Contact name, wxid, phone number, or username.
            content: Message text to send.
            app:     'wechat', 'telegram', or 'auto' (default: auto-detect from contact data).
        """
        # For now, route all to WeChat (Telegram support added in S6-02).
        resolved_app = app
        if app == "auto":
            # filehelper aliases are always WeChat; otherwise confirm the contact
            # exists in WeChat's address book (F2 resolution) before routing there.
            if to.strip().lower() in ("filehelper", "文件传输助手", "文件传输") or \
               to.strip() in ("文件传输助手", "文件传输"):
                resolved_app = "wechat"
            else:
                key = _read_wechat_key()
                db = _find_wechat_db()
                if db:
                    escaped = to.replace("'", "''")
                    rows = _run_sqlcipher(
                        f"SELECT username FROM rcontact WHERE type != 0 AND "
                        f"(username='{escaped}' OR nickname LIKE '%{escaped}%' OR conRemark LIKE '%{escaped}%') LIMIT 1",
                        db, key
                    )
                    if isinstance(rows, dict) and rows.get("rows"):
                        resolved_app = "wechat"
            if resolved_app == "auto":
                resolved_app = "wechat"  # default

        if resolved_app == "wechat":
            # Delegate the actual send (real UI automation + DB-verified status).
            return send_wechat_message(to=to, content=content)
        return json.dumps({"error": f"app '{resolved_app}' not supported yet (WeChat only)"})

    # ── tool: create_event ────────────────────────────────────────────────────

    @mcp.tool()
    def create_event(title: str, start_time: str, end_time: str = "",
                     description: str = "", location: str = "") -> str:
        """
        Create a calendar event via Android ContentProvider (CalendarContract).

        Args:
            title:       Event title.
            start_time:  ISO 8601 or natural time string, e.g. '2026-06-15 14:00'.
            end_time:    Event end time (optional, defaults to +1 hour).
            description: Event description (optional).
            location:    Event location (optional).
        """
        import re as _re
        from datetime import datetime, timedelta

        def parse_ts(s: str) -> int:
            for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M", "%Y-%m-%d"):
                try:
                    return int(datetime.strptime(s.strip(), fmt).timestamp() * 1000)
                except ValueError:
                    continue
            return int(datetime.now().timestamp() * 1000)

        start_ms = parse_ts(start_time)
        end_ms = parse_ts(end_time) if end_time else start_ms + 3600000

        try:
            # Find primary calendar account
            cal_query = subprocess.run(
                ["su", "-c", "content query --uri content://com.android.calendar/calendars --projection _id,account_name,name 2>/dev/null | head -5"],
                capture_output=True, text=True, timeout=10
            )
            # Insert event (uses default calendar ID=1)
            insert_cmd = (
                f"content insert --uri content://com.android.calendar/events "
                f"--bind dtstart:l:{start_ms} "
                f"--bind dtend:l:{end_ms} "
                f"--bind title:s:{repr(title)} "
                f"--bind description:s:{repr(description)} "
                f"--bind eventLocation:s:{repr(location)} "
                f"--bind calendar_id:l:1 "
                f"--bind eventTimezone:s:Asia/Shanghai "
                f"--bind hasAlarm:l:0"
            )
            result = subprocess.run(
                ["su", "-c", insert_cmd],
                capture_output=True, text=True, timeout=10
            )
            if result.returncode == 0:
                event_id = result.stdout.strip().replace("Result: id=", "").strip()
                return json.dumps({"status": "created", "event_id": event_id,
                                   "title": title, "start_ms": start_ms, "end_ms": end_ms})
            else:
                return json.dumps({"error": result.stderr[:300] or "insert failed"})
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── tool: read_document ───────────────────────────────────────────────────

    @mcp.tool()
    def read_document(query: str = "", source: str = "", limit: int = 20) -> str:
        """
        Search Memory Graph and recent events for relevant documents/messages.

        Args:
            query:  Search keyword or phrase.
            source: Filter by app source ('wechat', 'event_bus', etc.).
            limit:  Max results (default 20).
        """
        results = []

        # Search Memory Graph
        if query:
            try:
                import sys
                _base = os.path.dirname(__file__)
                if _base not in sys.path:
                    sys.path.insert(0, _base)
                from memory_graph import search_memory as _search
                mem_results = _search(query, source_app=source, limit=limit)
                for r in mem_results:
                    results.append({
                        "source": r.get("app", "wechat"),
                        "content": r.get("content", ""),
                        "talker": r.get("talker", ""),
                        "nickname": r.get("nickname", ""),
                        "ts": r.get("ts", 0),
                        "type": "message",
                    })
            except Exception as e:
                results.append({"error": f"memory_graph: {e}"})

        # Also search event bus
        try:
            with sqlite3.connect(EVENT_BUS_DB) as conn:
                sql = "SELECT type, app, data, ts FROM events WHERE 1=1"
                params = []
                if query:
                    sql += " AND data LIKE ?"
                    params.append(f"%{query}%")
                if source:
                    sql += " AND app = ?"
                    params.append(source)
                sql += f" ORDER BY ts DESC LIMIT {limit}"
                for row in conn.execute(sql, params).fetchall():
                    results.append({
                        "source": "event_bus",
                        "type": row[0],
                        "app": row[1],
                        "data": row[2][:200] if row[2] else "",
                        "ts": row[3],
                    })
        except Exception:
            pass

        return json.dumps({
            "results": results[:limit],
            "count": len(results),
            "query": query,
        }, ensure_ascii=False, default=str)

    # ── Sprint 8: Connector tools ──────────────────────────────────────────────

    @mcp.tool()
    def connector_read_wechat_messages(contact: str = "", limit: int = 50,
                                        since_ts: int = 0) -> str:
        """Read WeChat messages via WeChatConnector (config-driven, via Knowledge Base)."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from connectors import WeChatConnector
            wc = WeChatConnector()
            rows = wc.read_messages(contact=contact, limit=limit, since_ts=since_ts)
            return json.dumps({"rows": rows, "count": len(rows)}, ensure_ascii=False, default=str)
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def connector_read_wechat_contacts(limit: int = 100) -> str:
        """Read WeChat contacts via WeChatConnector."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from connectors import WeChatConnector
            wc = WeChatConnector()
            rows = wc.read_contacts(limit=limit)
            return json.dumps({"rows": rows, "count": len(rows)}, ensure_ascii=False, default=str)
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def run_discovery_agent(app: str = "wechat") -> str:
        """Run the Discovery Sub-agent for an app — detects version, DB schema, validates hooks."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from connectors import DiscoveryAgent
            agent = DiscoveryAgent(app)
            result = agent.run()
            return json.dumps(result, ensure_ascii=False, indent=2)
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def get_connector_kb(app: str = "wechat") -> str:
        """Return the Connector Knowledge Base for an app."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from connectors import _load_kb
            return json.dumps(_load_kb(app), ensure_ascii=False, indent=2)
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── Sprint 10: Persona tools ───────────────────────────────────────────────

    @mcp.tool()
    def get_persona() -> str:
        """Return the user's persona profile from ~/.hermes/persona.yaml."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from persona import get_persona as _get_persona
            return json.dumps(_get_persona(), ensure_ascii=False, indent=2)
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def update_persona(key: str, value: str) -> str:
        """Update a persona field. key supports dot-notation e.g. 'preferences.work_hours'."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from persona import update_persona as _update_persona
            import json as _json
            try:
                parsed_value = _json.loads(value)
            except Exception:
                parsed_value = value
            result = _update_persona(key, parsed_value)
            return json.dumps({"ok": True, "updated": result}, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def get_persona_prompt() -> str:
        """Generate the Hermes system prompt fragment from persona.yaml."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from persona import generate_system_prompt_fragment
            return generate_system_prompt_fragment()
        except Exception as e:
            return f"[persona error: {e}]"

    @mcp.tool()
    def infer_persona_from_behavior() -> str:
        """Auto-update persona (important_contacts, preferred_apps) from memory_graph data."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from persona import infer_from_memory_graph
            result = infer_from_memory_graph()
            return json.dumps(result, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── Sprint 11: Cloud Knowledge tools ──────────────────────────────────────

    @mcp.tool()
    def upload_operation_knowledge(intent: str, app: str, app_version: str,
                                    method: str, success_rate: float = 1.0) -> str:
        """Upload a successful operation's structural knowledge to cloud."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from cloud_knowledge import make_knowledge_entry, upload_knowledge
            entry = make_knowledge_entry(
                intent=intent, app=app, app_version=app_version,
                execution_path=[{"method": method, "step": 1, "result": "success"}],
                success_rate=success_rate,
            )
            result = upload_knowledge(entry)
            return json.dumps(result, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def pull_cloud_knowledge(app: str, intent: str, app_version: str = "") -> str:
        """Pull verified execution paths from cloud Knowledge Base."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from cloud_knowledge import pull_knowledge
            entries = pull_knowledge(app, intent, app_version)
            return json.dumps({"entries": entries, "count": len(entries)}, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def retry_cloud_uploads() -> str:
        """Retry any locally-pending cloud knowledge uploads."""
        try:
            sys.path.insert(0, EVENT_BUS_DIR)
            from cloud_knowledge import retry_pending_uploads
            result = retry_pending_uploads()
            return json.dumps(result, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── Sprint 6-03: System Contacts ContentProvider ──────────────────────────

    @mcp.tool()
    def get_system_contacts(query: str = "", limit: int = 50) -> str:
        """Read system contacts from Android ContactsContract ContentProvider.
        Returns name, phone numbers, and contact ID.
        Optionally filter by name query."""
        try:
            import re as _re

            def _parse_content_rows(output: str) -> list[dict]:
                rows = []
                current: dict = {}
                for line in output.splitlines():
                    m = _re.match(r"\s*Row:\s*\d+\s+(.+)", line)
                    if m:
                        if current:
                            rows.append(current)
                        current = {}
                        for kv in m.group(1).split(", "):
                            if "=" in kv:
                                k, _, v = kv.partition("=")
                                current[k.strip()] = v.strip()
                if current:
                    rows.append(current)
                return rows

            # Query names + IDs
            cmd_contacts = "content query --uri content://com.android.contacts/contacts --projection display_name:_id"
            r = subprocess.run(["sh", "-c", cmd_contacts], capture_output=True, text=True, timeout=15)
            contacts = _parse_content_rows(r.stdout)

            # Filter by query
            if query:
                q_lower = query.lower()
                contacts = [c for c in contacts if q_lower in c.get("display_name", "").lower()]

            contacts = contacts[:limit]
            if not contacts:
                return json.dumps({"contacts": [], "count": 0})

            # Enrich with phone numbers
            cmd_phones = ("content query --uri content://com.android.contacts/data "
                          "--projection display_name:data1:mimetype "
                          "--where \"mimetype='vnd.android.cursor.item/phone_v2'\"")
            rp = subprocess.run(["sh", "-c", cmd_phones], capture_output=True, text=True, timeout=15)
            phones_raw = _parse_content_rows(rp.stdout)

            phone_map: dict[str, list[str]] = {}
            for row in phones_raw:
                name = row.get("display_name", "")
                phone = row.get("data1", "")
                if name and phone:
                    phone_map.setdefault(name, []).append(phone)

            result = []
            for c in contacts:
                name = c.get("display_name", "")
                result.append({
                    "id": c.get("_id", ""),
                    "name": name,
                    "phones": phone_map.get(name, []),
                })

            return json.dumps({"contacts": result, "count": len(result)}, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── Sprint 6-04: System Calendar ContentProvider ──────────────────────────

    @mcp.tool()
    def get_calendar_events(days_ahead: int = 7, days_back: int = 1) -> str:
        """Read system calendar events from CalendarContract ContentProvider.
        Returns title, start/end times, location, description for upcoming events."""
        try:
            import time as _time
            now_ms = int(_time.time() * 1000)
            start_ms = now_ms - days_back * 86400000
            end_ms = now_ms + days_ahead * 86400000

            cmd = (f"content query --uri content://com.android.calendar/events "
                   f"--projection title:dtstart:dtend:eventLocation:description "
                   f"--where \"dtstart>={start_ms} AND dtstart<={end_ms}\" "
                   f"--sort \"dtstart ASC\"")
            r = subprocess.run(["sh", "-c", cmd], capture_output=True, text=True, timeout=15)

            import re as _re
            events = []
            for line in r.stdout.splitlines():
                m = _re.match(r"\s*Row:\s*\d+\s+(.+)", line)
                if m:
                    ev: dict = {}
                    for kv in m.group(1).split(", "):
                        if "=" in kv:
                            k, _, v = kv.partition("=")
                            ev[k.strip()] = v.strip()
                    if ev:
                        # Convert ms timestamps
                        for tf in ("dtstart", "dtend"):
                            if tf in ev:
                                try:
                                    import datetime as _dt
                                    ev[tf + "_iso"] = _dt.datetime.fromtimestamp(
                                        int(ev[tf]) / 1000).isoformat()
                                except Exception:
                                    pass
                        events.append(ev)

            return json.dumps({"events": events, "count": len(events)}, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── Sprint 5-03: Create task via Android Tasks/Reminders ─────────────────

    @mcp.tool()
    def create_task(title: str, deadline: str = "", note: str = "") -> str:
        """Create a task/reminder in the system Tasks app (Google Tasks or Samsung Reminders).
        deadline: ISO datetime string or empty. note: optional description."""
        try:
            import time as _time
            import datetime as _dt

            # Parse deadline
            due_ms = ""
            if deadline:
                try:
                    dt = _dt.datetime.fromisoformat(deadline.replace("Z", "+00:00"))
                    due_ms = str(int(dt.timestamp() * 1000))
                except Exception:
                    pass

            # Try Google Tasks ContentProvider first
            bind = f"--bind title:s:{title}"
            if note:
                bind += f" --bind notes:s:{note}"
            if due_ms:
                bind += f" --bind due:l:{due_ms}"

            r = subprocess.run(
                ["sh", "-c", f"content insert --uri content://com.google.android.gms.tasks/tasks {bind}"],
                capture_output=True, text=True, timeout=10
            )

            if "error" not in r.stderr.lower() and r.returncode == 0:
                return json.dumps({"ok": True, "method": "google_tasks", "title": title})

            # Fallback: use am start to open Tasks app with pre-filled title
            intent_cmd = (f"am start -a android.intent.action.INSERT "
                          f"-d content://com.android.task/ "
                          f"--es title {title!r}")
            if note:
                intent_cmd += f" --es description {note!r}"
            r2 = subprocess.run(["sh", "-c", intent_cmd], capture_output=True, text=True, timeout=10)

            if "Success" in r2.stdout or r2.returncode == 0:
                return json.dumps({"ok": True, "method": "intent", "title": title,
                                   "note": "Task created via Intent; verify in Tasks app"})

            return json.dumps({"ok": False, "error": r2.stderr[:200] or r.stderr[:200],
                               "hint": "No Tasks account configured on device"})
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── Sprint 6-01: MMKV reader ───────────────────────────────────────────────

    @mcp.tool()
    def read_wechat_mmkv(mmkv_file: str = "mmkv.default", max_keys: int = 50) -> str:
        """Read key-value pairs from WeChat MMKV storage files.
        MMKV files are in /data/data/com.tencent.mm/files/mmkv/.
        Returns readable string and integer key-value pairs."""
        try:
            mmkv_dir = "/data/data/com.tencent.mm/files/mmkv"
            target = f"{mmkv_dir}/{mmkv_file}"

            r = subprocess.run(["su", "-c", f"ls {mmkv_dir}"],
                               capture_output=True, text=True, timeout=5)
            available = [f for f in r.stdout.splitlines() if not f.endswith(".crc")]

            # Copy file to readable location
            tmp = f"/data/local/tmp/mmkv_read_{mmkv_file.replace('/', '_')}"
            cp = subprocess.run(["su", "-c", f"cp {target} {tmp} && chmod 644 {tmp}"],
                                capture_output=True, text=True, timeout=5)
            if cp.returncode != 0:
                return json.dumps({"error": f"Cannot read {target}", "available": available})

            # Parse MMKV binary format
            # MMKV format: 4-byte header (size), then protobuf-like key-value pairs
            # Key: varint len + bytes; Value: varint len + bytes (or int32/int64 varint)
            with open(tmp, "rb") as f:
                raw = f.read()

            # Skip 4-byte header
            pos = 4
            results = []
            seen_keys: set = set()

            def read_varint(data: bytes, offset: int):
                result = 0
                shift = 0
                while offset < len(data):
                    b = data[offset]
                    offset += 1
                    result |= (b & 0x7F) << shift
                    shift += 7
                    if not (b & 0x80):
                        break
                return result, offset

            while pos < len(raw) and len(results) < max_keys:
                if pos >= len(raw):
                    break
                try:
                    key_len, pos = read_varint(raw, pos)
                    if key_len <= 0 or key_len > 256 or pos + key_len > len(raw):
                        pos += 1
                        continue
                    key = raw[pos:pos + key_len]
                    pos += key_len
                    try:
                        key_str = key.decode("utf-8")
                    except Exception:
                        key_str = key.hex()

                    val_len, pos = read_varint(raw, pos)
                    if val_len < 0 or pos + val_len > len(raw):
                        pos += 1
                        continue
                    val = raw[pos:pos + val_len]
                    pos += val_len

                    if key_str in seen_keys:
                        continue
                    seen_keys.add(key_str)

                    # Try to decode value as string, then int
                    try:
                        val_str = val.decode("utf-8")
                        if val_str.isprintable():
                            results.append({"key": key_str, "value": val_str, "type": "string"})
                            continue
                    except Exception:
                        pass
                    if len(val) <= 8:
                        try:
                            int_val = int.from_bytes(val, "little")
                            results.append({"key": key_str, "value": int_val, "type": "int"})
                            continue
                        except Exception:
                            pass
                    results.append({"key": key_str, "value": val.hex()[:64], "type": "bytes"})
                except Exception:
                    pos += 1
                    continue

            # Cleanup temp
            subprocess.run(["su", "-c", f"rm -f {tmp}"], capture_output=True, timeout=3)

            return json.dumps({
                "file": mmkv_file,
                "available_files": available,
                "entries": results,
                "count": len(results),
            }, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── Sprint 6-02: Telegram SQLite connector ────────────────────────────────

    @mcp.tool()
    def get_telegram_messages(limit: int = 50, query: str = "", chat_id: int = 0) -> str:
        """Read messages from Telegram's local SQLite database (cache4.db).
        Requires Telegram to be installed and logged in.
        query: optional text filter. chat_id: filter by specific chat (0 = all).
        Returns messages with sender name, text, date."""
        try:
            TG_PKG = "org.telegram.messenger.web"
            TG_DB = f"/data/data/{TG_PKG}/files/cache4.db"
            TG_DB2 = f"/data/data/{TG_PKG}/databases/cache4.db"
            tmp = "/data/local/tmp/tg_cache4.db"

            # Find the database
            db_path = None
            for p in [TG_DB, TG_DB2]:
                r = subprocess.run(["su", "-c", f"test -f {p} && echo exists"],
                                   capture_output=True, text=True, timeout=5)
                if "exists" in r.stdout:
                    db_path = p
                    break

            if not db_path:
                # List all files to help locate
                r2 = subprocess.run(["su", "-c", f"find /data/data/{TG_PKG} -name '*.db' 2>/dev/null"],
                                    capture_output=True, text=True, timeout=10)
                return json.dumps({
                    "error": "Telegram database not found — app may not be logged in yet",
                    "found_dbs": r2.stdout.strip().splitlines(),
                    "hint": "Launch Telegram and log in first",
                })

            # Copy to accessible location
            subprocess.run(["su", "-c", f"cp {db_path} {tmp} && chmod 644 {tmp}"],
                           capture_output=True, text=True, timeout=10)

            import sqlite3 as _sq
            conn = _sq.connect(tmp)
            conn.row_factory = _sq.Row

            # Build query
            sql = """
                SELECT m.mid, m.uid, m.date, m.message,
                       m.from_id,
                       COALESCE(u.first_name || ' ' || COALESCE(u.last_name,''), u.name, '') as sender_name,
                       COALESCE(u.username, '') as sender_username
                FROM messages m
                LEFT JOIN users u ON m.from_id = u.uid
                WHERE m.message != '' AND m.message IS NOT NULL
            """
            params: list = []
            if chat_id:
                sql += " AND m.uid = ?"
                params.append(chat_id)
            if query:
                sql += " AND m.message LIKE ?"
                params.append(f"%{query}%")
            sql += f" ORDER BY m.date DESC LIMIT {limit}"

            try:
                rows = conn.execute(sql, params).fetchall()
            except _sq.OperationalError:
                # Try alternative table/column names
                rows = []
                try:
                    alt = conn.execute(
                        f"SELECT name FROM sqlite_master WHERE type='table'"
                    ).fetchall()
                    tables = [r["name"] for r in alt]
                    conn.close()
                    subprocess.run(["su", "-c", f"rm -f {tmp}"], capture_output=True, timeout=3)
                    return json.dumps({"error": "schema mismatch", "tables": tables})
                except Exception:
                    pass
                conn.close()
                subprocess.run(["su", "-c", f"rm -f {tmp}"], capture_output=True, timeout=3)
                return json.dumps({"error": "Telegram DB schema not recognized"})

            messages = []
            for r in rows:
                import datetime as _dt
                try:
                    ts_iso = _dt.datetime.fromtimestamp(r["date"]).isoformat()
                except Exception:
                    ts_iso = str(r["date"])
                messages.append({
                    "mid": r["mid"],
                    "chat_id": r["uid"],
                    "from_id": r["from_id"],
                    "sender": r["sender_name"].strip() or r["sender_username"] or str(r["from_id"]),
                    "text": r["message"],
                    "ts": r["date"],
                    "ts_iso": ts_iso,
                })

            conn.close()
            subprocess.run(["su", "-c", f"rm -f {tmp}"], capture_output=True, timeout=3)
            return json.dumps({"messages": messages, "count": len(messages)}, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def get_telegram_contacts(query: str = "", limit: int = 100) -> str:
        """Read Telegram contacts from local database. Requires Telegram logged in."""
        try:
            TG_PKG = "org.telegram.messenger.web"
            tmp = "/data/local/tmp/tg_cache4.db"

            r = subprocess.run(["su", "-c",
                                f"find /data/data/{TG_PKG} -name 'cache4.db' 2>/dev/null | head -1"],
                               capture_output=True, text=True, timeout=10)
            db_path = r.stdout.strip()
            if not db_path:
                return json.dumps({"error": "Telegram not logged in — database not found"})

            subprocess.run(["su", "-c", f"cp {db_path} {tmp} && chmod 644 {tmp}"],
                           capture_output=True, text=True, timeout=10)

            import sqlite3 as _sq
            conn = _sq.connect(tmp)
            conn.row_factory = _sq.Row

            sql = "SELECT uid, first_name, last_name, username, phone FROM users WHERE 1=1"
            params: list = []
            if query:
                sql += " AND (first_name LIKE ? OR last_name LIKE ? OR username LIKE ?)"
                params += [f"%{query}%"] * 3
            sql += f" ORDER BY first_name LIMIT {limit}"

            try:
                rows = conn.execute(sql, params).fetchall()
            except _sq.OperationalError as e:
                conn.close()
                subprocess.run(["su", "-c", f"rm -f {tmp}"], capture_output=True, timeout=3)
                return json.dumps({"error": str(e)})

            contacts = [{"uid": r["uid"],
                         "name": f"{r['first_name'] or ''} {r['last_name'] or ''}".strip(),
                         "username": r["username"],
                         "phone": r["phone"]} for r in rows]
            conn.close()
            subprocess.run(["su", "-c", f"rm -f {tmp}"], capture_output=True, timeout=3)
            return json.dumps({"contacts": contacts, "count": len(contacts)}, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

    # ── E1: Telephony tools (phone & SMS) — independent module ────────────────
    try:
        from telephony_tools import register_telephony_tools
        register_telephony_tools(mcp)
    except Exception as exc:  # pragma: no cover - keep server up if module missing
        logger.warning("telephony_tools not registered: %s", exc)

    # ── E2: UI control primitives (tap/swipe/screenshot/...) — independent module ──
    try:
        from ui_control_tools import register_ui_control_tools
        register_ui_control_tools(mcp)
    except Exception as exc:  # pragma: no cover - keep server up if module missing
        logger.warning("ui_control_tools not registered: %s", exc)

    # ── E3: App management & device state (open_app/list_packages/...) — independent module ──
    try:
        from device_tools import register_device_tools
        register_device_tools(mcp)
    except Exception as exc:  # pragma: no cover - keep server up if module missing
        logger.warning("device_tools not registered: %s", exc)

    # ── E5: OCR fallback (ocr_find_text/ocr_click_text) — independent module ──
    try:
        from ocr_tools import register_ocr_tools
        register_ocr_tools(mcp)
    except Exception as exc:  # pragma: no cover - keep server up if module/models missing
        logger.warning("ocr_tools not registered: %s", exc)

    # ── E6: Frida deep-runtime inspection (Intent/Bitmap) — independent module ──
    try:
        from frida_inspect_tools import register_frida_inspect_tools
        register_frida_inspect_tools(mcp)
    except Exception as exc:  # pragma: no cover - keep server up if module missing
        logger.warning("frida_inspect_tools not registered: %s", exc)

    # ── E4: Code deploy tool (Mac → Pixel sync) ─────────────────────────────────
    @mcp.tool()
    def deploy_phoneuse_code(restart: bool = True) -> str:
        """Deploy the latest phoneuse Python code from the Mac repo to the Pixel device.

        This pushes all .py files from the repo directory to the device's
        ~/.hermes/event_bus/ and optionally restarts the MCP server so Hermes
        picks up new tools immediately.

        Args:
            restart: Whether to restart the MCP server after deploy (default True).
        """
        import subprocess as sp
        repo = Path(__file__).resolve().parent
        deploy_script = repo / "deploy.sh"
        if not deploy_script.exists():
            return json.dumps({"ok": False, "error": "deploy.sh not found in repo"})

        args = [str(deploy_script)]
        if not restart:
            args.append("--no-restart")

        try:
            result = sp.run(args, capture_output=True, text=True, timeout=120)
            return json.dumps({
                "ok": result.returncode == 0,
                "stdout": result.stdout,
                "stderr": result.stderr,
                "returncode": result.returncode,
            })
        except sp.TimeoutExpired:
            return json.dumps({"ok": False, "error": "deploy.sh timed out after 120s"})
        except Exception as e:
            return json.dumps({"ok": False, "error": str(e)})

    return mcp


def main():
    verbose = "--verbose" in sys.argv or "-v" in sys.argv
    logging.basicConfig(
        level=logging.INFO if verbose else logging.WARNING,
        stream=sys.stderr,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )
    os.environ.setdefault("HERMES_QUIET", "1")

    try:
        server = _build_server()
    except ImportError as exc:
        sys.stderr.write(f"android_mcp: requires 'mcp' package: {exc}\n")
        sys.exit(2)

    try:
        server.run()
    except KeyboardInterrupt:
        sys.exit(0)
    except Exception as exc:
        logger.exception("android_mcp crashed")
        sys.stderr.write(f"android_mcp error: {exc}\n")
        sys.exit(1)


if __name__ == "__main__":
    main()