宜博 / 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 FULLY-SYNTHETIC voice message to a WeChat contact (no recording, no hot-swap).

        Fires SEND_VOICE_CGI → the hook chunks the SILK, builds WeChat's own
        /cgi-bin/micromsg-bin/uploadvoice request (audio inline), dispatches it via the
        NetSceneQueue, and inserts a local sent bubble. Result: the recipient gets a
        playable voice AND the sender sees a sent bubble. Verified end-to-end on WeChat
        8.0.74 (recipient + Mac both received). Delivery is async; confirm via UI if needed.

        The SILK is auto-staged into WeChat's data dir (its uid + SELinux context) so the
        WeChat process can read it — a device /sdcard path alone would EACCES.

        Args:
            to: Contact wxid as in rcontact.username (e.g. 'filehelper', 'wxid_xxx',
                'yibooo'). Test with 'filehelper' first to avoid account risk.
            silk_path: Absolute device path to a SILK V3 file (Tencent variant, starts
                       with \\x02#!SILK_V3), e.g. edge-tts → ffmpeg 16k mono → silk-wasm.
            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_cgi_via_hook
            result = send_voice_cgi_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()