#!/usr/bin/python3
"""
PageHawk MCP server - LOCAL, zero-dependency stdio JSON-RPC 2.0.

Exposes PageHawk's own on-device full-text search index (the same SQLite/FTS5
database the app's Search Workbench builds — see src/app/viewer/SearchIndex.swift)
to the user's AI assistant, so Claude can list, search, and quote from the
documents PageHawk has indexed.

HARD BOUNDARY: this is a LOCAL stdio server. It opens the index database
read-only (sqlite3 URI mode=ro) and reads document files already on disk; it
never writes to the index, never modifies a document, and never talks to
anything beyond this Mac. Its only socket is a loopback (127.0.0.1) call into
the RUNNING PageHawk app's own AgentBridge (token-authenticated, announced in
~/Library/Application Support/PageHawk/agent_bridge.json), which powers the
live tools: list open windows/tabs, open a document, turn to a page, and read
the user's annotations. Those are reversible UI actions a user could perform;
nothing here edits a document or deletes anything. It never spawns extraction
subprocesses — a document that is not yet indexed is reported as such, not
fabricated.

No pip install, no venv, no third-party packages. Runs on the macOS system
/usr/bin/python3. Speaks newline-delimited JSON-RPC 2.0 on stdio (the MCP stdio
transport). stdout carries ONLY protocol JSON; all diagnostics go to stderr.

PII REDACTION, ON BY DEFAULT (atlas findings 406/407 — DEVONthink ships its own MCP
server with "Redact sensitive content" checked by default): every read tool that can
return document/annotation text (pagehawk_search, pagehawk_get_page_text,
pagehawk_get_annotations) masks emails, credit-card numbers, IBAN, API keys/tokens, and
phone numbers before the text leaves this process, via pii_redact.py. A caller may pass
redact=false on any of those three tools to get the raw text back — an explicit,
per-call opt-out, never the default, and every use of it is logged to stderr (never
silent). Asymmetric on purpose, matching DEVONthink's own stated rule: this server has
no write tools at all, so "redact reads, never writes" is true by construction.

Same zero-dependency pattern and detect/explain/act/verify concierge shape as
the rest of the fleet's local connectors (RecapMuse's engine/mcp/server.py,
DisplayDial's server.py, Galley Text's galleytext-mcp.cjs); MCPConnector.swift
in this app's src/app/viewer/ registers this file with Claude Desktop.
"""
from __future__ import annotations

import datetime
import hashlib
import json
import os
import plistlib
import sqlite3
import sys
import urllib.parse
import urllib.request

import pii_redact

SERVER_NAME = "pagehawk"
SERVER_VERSION = "0.3.0"
FALLBACK_PROTOCOL = "2024-11-05"

HOME = os.path.expanduser("~")
DEFAULT_INDEX_DB = os.path.join(HOME, "Library", "Application Support", "PageHawk", "searchindex.db")
DEFAULT_OCR_DIR = os.path.join(HOME, "Library", "Application Support", "PageHawk", "ocr")
DEFAULT_BRIDGE_JSON = os.path.join(HOME, "Library", "Application Support", "PageHawk", "agent_bridge.json")
DEFAULT_PREFS_PLIST = os.path.join(HOME, "Library", "Preferences", "com.epiphani.pagehawk.plist")
APP_PATHS = ["/Applications/PageHawk.app", os.path.join(HOME, "Applications", "PageHawk.app")]

MAX_PAGE_TEXT_BYTES = 8000


def log(*args):
    sys.stderr.write("pagehawk-mcp: " + " ".join(str(a) for a in args) + "\n")
    sys.stderr.flush()


# --------------------------------------------------------------------------- corpus access
def index_db_path() -> str:
    """The real index the app's Search Workbench builds: env override (tests) -> the
    default. Resolved per call so an app-side reindex is picked up without restarting
    this server."""
    env = os.environ.get("PAGEHAWK_INDEX_DB")
    return os.path.expanduser(env) if env else DEFAULT_INDEX_DB


def ocr_cache_dir() -> str:
    env = os.environ.get("PAGEHAWK_OCR_DIR")
    return os.path.expanduser(env) if env else DEFAULT_OCR_DIR


def prefs_plist_path() -> str:
    env = os.environ.get("PAGEHAWK_PREFS_PLIST")
    return os.path.expanduser(env) if env else DEFAULT_PREFS_PLIST


def connect_ro() -> sqlite3.Connection | None:
    """Open the index strictly read-only via a file: URI (WAL mode makes this safe
    alongside the app writing concurrently). None if no index exists yet — never
    creates one; this server never mutates the corpus."""
    path = index_db_path()
    if not os.path.isfile(path):
        return None
    uri = "file:" + urllib.parse.quote(path) + "?mode=ro"
    conn = None
    try:
        conn = sqlite3.connect(uri, uri=True, timeout=4.0)
        conn.row_factory = sqlite3.Row
        conn.execute("SELECT 1 FROM files LIMIT 1")  # confirms the expected schema, not just a file
        return conn
    except sqlite3.Error:
        if conn is not None:
            conn.close()
        return None


def iso(epoch) -> str | None:
    if not epoch:
        return None
    try:
        return datetime.datetime.fromtimestamp(int(epoch), tz=datetime.timezone.utc).isoformat()
    except (OverflowError, OSError, ValueError):
        return None


def app_install() -> tuple[str | None, str | None]:
    """(app_path, version) for the first PageHawk.app found in the usual spots. The
    version is read from the bundle's own Resources/VERSION, the same file the app
    stamps at build time — never guessed."""
    for p in APP_PATHS:
        if os.path.isdir(p):
            vpath = os.path.join(p, "Contents", "Resources", "VERSION")
            version = None
            try:
                with open(vpath, encoding="utf-8") as f:
                    version = f.read().strip() or None
            except OSError:
                pass
            return p, version
    return None, None


def folder_lists() -> tuple[list[str], list[str]]:
    """recentFolders/pinnedFolders as PageHawk itself stores them: plain UserDefaults
    arrays in the app's own preferences domain. Missing or unreadable -> empty lists,
    never an error (this is a nice-to-have on the status tool, not load-bearing)."""
    try:
        with open(prefs_plist_path(), "rb") as f:
            data = plistlib.load(f)
        recent = data.get("recentFolders")
        pinned = data.get("pinnedFolders")
        recent = recent if isinstance(recent, list) else []
        pinned = pinned if isinstance(pinned, list) else []
        return [str(x) for x in recent], [str(x) for x in pinned]
    except Exception:
        return [], []


def ocr_cache_key(path: str) -> str | None:
    """Mirrors main.swift's ocrCacheKey(_:) exactly: SHA256("path|mtime|size") hex,
    so this server can tell whether a scan already has an OCR cache entry without
    ever running OCR itself."""
    try:
        st = os.stat(path)
    except OSError:
        return None
    return hashlib.sha256(f"{path}|{int(st.st_mtime)}|{st.st_size}".encode("utf-8")).hexdigest()


def ocr_cached(path: str) -> bool:
    key = ocr_cache_key(path)
    if not key:
        return False
    return os.path.isfile(os.path.join(ocr_cache_dir(), key + ".json"))


def excerpt(body: str, query: str, radius: int = 60) -> str:
    """A short, honest context window around the match. Falls back to the chunk's
    own opening text when the literal query is not found verbatim in this chunk
    (can happen on ranked-but-not-exact FTS candidates)."""
    flat = body.replace("\n", " ").replace("\r", " ")
    idx = flat.lower().find(query.lower())
    if idx < 0:
        s = flat[:160].strip()
        return s + ("…" if len(flat) > 160 else "")
    lo = max(0, idx - radius)
    hi = min(len(flat), idx + len(query) + radius)
    s = flat[lo:hi].strip()
    return ("…" if lo > 0 else "") + s + ("…" if hi < len(flat) else "")


def redact_wanted(args: dict, tool_name: str) -> bool:
    """True unless the caller explicitly passed redact=false. That opt-out is logged to
    stderr every time — never silent — so a raw, unmasked read of this Mac's documents
    always leaves an audit trail, even though this server itself has no persistent log
    file (stdout is protocol-only)."""
    raw = args.get("redact", True)
    wanted = raw is not False and raw != "false" and raw != 0
    if not wanted:
        log(f"PII redaction explicitly disabled (redact=false) for {tool_name}")
    return wanted


def fts_phrase(query: str) -> str:
    """Escape a term for an FTS5 trigram phrase MATCH: wrap in double quotes, double
    any embedded quote, so the term is matched as a literal substring. This value is
    ALWAYS passed as a bound parameter (`MATCH ?`), never interpolated into the SQL
    text — the SQL string itself never changes shape based on user input."""
    return '"' + query.replace('"', '""') + '"'


# --------------------------------------------------------------------------- live-app bridge
def bridge_json_path() -> str:
    env = os.environ.get("PAGEHAWK_BRIDGE_JSON")
    return os.path.expanduser(env) if env else DEFAULT_BRIDGE_JSON


def bridge_info() -> tuple[dict | None, str]:
    """(info, why_not). The bridge file is written by the running app's AgentBridge
    with a per-launch port + token; a file whose pid is no longer alive is a stale
    leftover from a crash, reported honestly as 'not running'."""
    path = bridge_json_path()
    try:
        with open(path, encoding="utf-8") as f:
            info = json.load(f)
    except OSError:
        return None, ("PageHawk is not running (no live bridge at "
                      f"{path}). The index tools still work; open PageHawk to use the live tools.")
    except ValueError:
        return None, f"The bridge file at {path} is unreadable."
    pid = info.get("pid")
    try:
        os.kill(int(pid), 0)
    except (OSError, TypeError, ValueError):
        return None, ("PageHawk is not running (its last bridge file is stale). "
                      "Open PageHawk to use the live tools.")
    if not info.get("port") or not info.get("token"):
        return None, f"The bridge file at {path} is missing its port/token."
    return info, ""


def bridge_call(endpoint: str, params: dict | None = None, timeout: float = 20.0) -> dict:
    """GET the running app's loopback bridge. Loopback only, token from the bridge
    file, never off this Mac. Every failure returns ok=False with the real reason."""
    info, why = bridge_info()
    if info is None:
        return {"ok": False, "error": why, "app_running": False}
    query = urllib.parse.urlencode({k: v for k, v in (params or {}).items() if v is not None})
    url = f"http://127.0.0.1:{info['port']}{endpoint}" + (f"?{query}" if query else "")
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {info['token']}"})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            body = json.loads(resp.read().decode("utf-8"))
    except Exception as e:  # URLError, timeout, HTTPError, bad JSON — one honest shape
        return {"ok": False, "app_running": True,
                "error": f"The running app's bridge did not answer ({e}). "
                         "It may be mid-launch; try again in a moment."}
    if isinstance(body, dict):
        body.setdefault("app_running", True)
        return body
    return {"ok": False, "app_running": True, "error": "bridge returned a non-object reply"}


# --------------------------------------------------------------------------- tools
def tool_status(args: dict) -> dict:
    app_path, app_version = app_install()
    db = index_db_path()
    conn = connect_ro()
    file_count = 0
    newest_at = None
    if conn is not None:
        try:
            row = conn.execute("SELECT COUNT(*), MAX(indexed_at) FROM files").fetchone()
            file_count = row[0] or 0
            newest_at = row[1]
        except sqlite3.Error:
            pass
        finally:
            conn.close()
    try:
        ocr_count = len([f for f in os.listdir(ocr_cache_dir()) if f.endswith(".json")])
    except OSError:
        ocr_count = 0
    recent, pinned = folder_lists()
    live = bridge_call("/status", timeout=5.0)
    return {
        "server": SERVER_NAME,
        "version": SERVER_VERSION,
        "app_running": bool(live.get("app_running")) and bool(live.get("ok")),
        "live": (live if live.get("ok")
                 else {"ok": False, "why": live.get("error")}),
        "app_installed": app_path is not None,
        "app_path": app_path,
        "app_version": app_version,
        "index_db_path": db,
        "index_present": conn is not None,
        "indexed_files": file_count,
        "index_newest_at": iso(newest_at),
        "ocr_cache_files": ocr_count,
        "recent_folders": recent,
        "pinned_folders": pinned,
        "message": ("Ready." if conn is not None
                    else "No index yet; open a folder in PageHawk to build one."),
        "privacy": "This server reads PageHawk's own local search index and document files "
                   "already on this Mac, read-only, and never uses the network.",
    }


def tool_list_documents(args: dict) -> dict:
    conn = connect_ro()
    if conn is None:
        return {"ok": False, "error": f"No index yet at {index_db_path()}. "
                                      "Open a folder in PageHawk to build one."}
    filt = (args.get("filter") or "").strip()
    limit = max(1, min(int(args.get("limit") or 50), 500))
    try:
        if filt:
            rows = conn.execute(
                "SELECT path, type, size, mtime, indexed_at FROM files WHERE path LIKE ? "
                "ORDER BY indexed_at DESC LIMIT ?",
                (f"%{filt}%", limit),
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT path, type, size, mtime, indexed_at FROM files "
                "ORDER BY indexed_at DESC LIMIT ?",
                (limit,),
            ).fetchall()
    finally:
        conn.close()
    docs = [{
        "path": r["path"], "name": os.path.basename(r["path"]), "type": r["type"],
        "size": r["size"], "mtime": iso(r["mtime"]), "indexed_at": iso(r["indexed_at"]),
    } for r in rows]
    return {"ok": True, "count": len(docs), "documents": docs}


def tool_search(args: dict) -> dict:
    query = (args.get("query") or "").strip()
    if len(query) < 3:
        return {"ok": False, "error": "query must be at least 3 characters "
                                      "(the index uses trigram matching)."}
    conn = connect_ro()
    if conn is None:
        return {"ok": False, "error": f"No index yet at {index_db_path()}. "
                                      "Open a folder in PageHawk to build one."}
    limit_docs = max(1, min(int(args.get("limit") or 20), 100))
    hits_per_doc = max(1, min(int(args.get("hits_per_doc") or 5), 40))
    do_redact = redact_wanted(args, "pagehawk_search")
    expr = fts_phrase(query)
    try:
        rows = conn.execute(
            "SELECT c.path, c.atype, c.anchor, c.page, c.body "
            "FROM chunk_fts f JOIN chunk c ON c.id = f.rowid "
            "WHERE chunk_fts MATCH ? ORDER BY rank LIMIT ?",
            (expr, limit_docs * hits_per_doc * 4),
        ).fetchall()
    except sqlite3.Error as e:
        conn.close()
        return {"ok": False, "error": f"search failed: {e}"}
    order: list[str] = []
    docs: dict[str, dict] = {}
    for r in rows:
        p = r["path"]
        if p not in docs:
            docs[p] = {"path": p, "name": os.path.basename(p), "type": None, "count": 0, "hits": []}
            order.append(p)
        d = docs[p]
        d["count"] += 1
        if len(d["hits"]) < hits_per_doc:
            snippet = excerpt(r["body"] or "", query)
            hit_counts: dict[str, int] = {}
            if do_redact:
                snippet, hit_counts = pii_redact.redact_text(snippet)
            d["hits"].append({
                "atype": r["atype"], "anchor": r["anchor"],
                "page": r["page"] or None,
                "snippet": snippet,
            })
            if hit_counts:
                d["pii_redacted"] = pii_redact.merge_hits(d.get("pii_redacted", {}), hit_counts)
    results = [docs[p] for p in order[:limit_docs]]
    if results:
        placeholders = ",".join("?" * len(results))
        try:
            types = {row["path"]: row["type"] for row in conn.execute(
                f"SELECT path, type FROM files WHERE path IN ({placeholders})",
                [d["path"] for d in results]).fetchall()}
        except sqlite3.Error:
            types = {}
        for d in results:
            d["type"] = types.get(d["path"])
    conn.close()
    return {"ok": True, "query": query, "documents": len(results), "results": results,
            "redacted": do_redact}


def tool_get_document(args: dict) -> dict:
    path = os.path.expanduser((args.get("path") or "").strip())
    if not path:
        return {"ok": False, "error": "path is required (use list_documents or search first)."}
    conn = connect_ro()
    if conn is None:
        return {"ok": False, "error": f"No index yet at {index_db_path()}."}
    try:
        row = conn.execute(
            "SELECT path, type, size, mtime, indexed_at FROM files WHERE path=?", (path,)
        ).fetchone()
        if not row:
            conn.close()
            return {"ok": False, "error": f"{path} is not indexed yet — "
                                          "open its folder in PageHawk to index it."}
        pages = conn.execute(
            "SELECT MIN(page), MAX(page) FROM chunk WHERE path=? AND page>0", (path,)
        ).fetchone()
        chunk_count = conn.execute("SELECT COUNT(*) FROM chunk WHERE path=?", (path,)).fetchone()[0]
    finally:
        conn.close()
    return {
        "ok": True,
        "path": row["path"], "name": os.path.basename(row["path"]), "type": row["type"],
        "size": row["size"], "mtime": iso(row["mtime"]), "indexed_at": iso(row["indexed_at"]),
        "page_range": {"min": pages[0], "max": pages[1]} if pages and pages[0] else None,
        "chunk_count": chunk_count,
        "ocr_cached": ocr_cached(row["path"]),
    }


def tool_get_page_text(args: dict) -> dict:
    path = os.path.expanduser((args.get("path") or "").strip())
    if not path:
        return {"ok": False, "error": "path is required."}
    page_arg = args.get("page")
    conn = connect_ro()
    if conn is None:
        return {"ok": False, "error": f"No index yet at {index_db_path()}."}
    try:
        if not conn.execute("SELECT 1 FROM files WHERE path=?", (path,)).fetchone():
            return {"ok": False, "error": f"{path} is not indexed yet — "
                                          "open its folder in PageHawk to index it."}
        if page_arg is not None and str(page_arg).strip() != "":
            try:
                page = int(page_arg)
            except (TypeError, ValueError):
                return {"ok": False, "error": f"page must be an integer, got {page_arg!r}."}
            rows = conn.execute(
                "SELECT atype, anchor, body FROM chunk WHERE path=? AND page=? ORDER BY id",
                (path, page),
            ).fetchall()
            if not rows:
                return {"ok": False, "error": f"No page {page} indexed for {path}. "
                                              "Use get_document for its page range, or omit "
                                              "page for a non-paginated document (paragraphs/"
                                              "cells/lines)."}
        else:
            page = None
            limit = max(1, min(int(args.get("limit") or 60), 400))
            rows = conn.execute(
                "SELECT atype, anchor, body FROM chunk WHERE path=? ORDER BY id LIMIT ?",
                (path, limit),
            ).fetchall()
            if not rows:
                return {"ok": False, "error": f"No indexed content for {path}."}
    finally:
        conn.close()
    text = "\n".join(r["body"] or "" for r in rows)
    truncated = len(text.encode("utf-8")) > MAX_PAGE_TEXT_BYTES
    if truncated:
        text = text.encode("utf-8")[:MAX_PAGE_TEXT_BYTES].decode("utf-8", errors="ignore")
    do_redact = redact_wanted(args, "pagehawk_get_page_text")
    pii_hits: dict[str, int] = {}
    if do_redact:
        text, pii_hits = pii_redact.redact_text(text)
    result = {
        "ok": True, "path": path, "page": page,
        "atype": rows[0]["atype"] if rows else None,
        "anchors": [r["anchor"] for r in rows],
        "text": text,
        "truncated": truncated,
        "redacted": do_redact,
    }
    if pii_hits:
        result["pii_redacted"] = pii_hits
    return result


# --------------------------------------------------------------------------- live-app tools
def tool_list_open(args: dict) -> dict:
    """Every window/tab the RUNNING app has open, with the live page position for
    reader windows (read from the on-screen viewer, not a cache)."""
    return bridge_call("/windows")


def tool_open_document(args: dict) -> dict:
    path = (args.get("path") or "").strip()
    if not path:
        return {"ok": False, "error": "path is required."}
    return bridge_call("/open", {"path": path, "page": args.get("page")})


def tool_goto_page(args: dict) -> dict:
    page = args.get("page")
    if page is None:
        return {"ok": False, "error": "page is required."}
    return bridge_call("/goto", {"page": page,
                                 "path": (args.get("path") or "").strip() or None,
                                 "window": (args.get("window") or "").strip() or None})


def tool_get_annotations(args: dict) -> dict:
    result = bridge_call("/annotations", {"path": (args.get("path") or "").strip() or None},
                         timeout=30.0)
    do_redact = redact_wanted(args, "pagehawk_get_annotations")
    result["redacted"] = do_redact
    if do_redact and result.get("ok"):
        all_hits: list[dict[str, int]] = []
        for item in result.get("annotations") or []:
            for field in ("text", "note"):
                if item.get(field):
                    item[field], hits = pii_redact.redact_text(item[field])
                    if hits:
                        all_hits.append(hits)
        if result.get("markdown"):
            result["markdown"], md_hits = pii_redact.redact_text(result["markdown"])
            if md_hits:
                all_hits.append(md_hits)
        merged = pii_redact.merge_hits(*all_hits)
        if merged:
            result["pii_redacted"] = merged
    return result


TOOLS = [
    {
        "name": "pagehawk_status",
        "description": "Health check: whether PageHawk is installed and its version, whether "
                       "the search index exists and how many files it covers, OCR cache size, "
                       "recent/pinned folders, and the privacy contract.",
        "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
        "handler": tool_status,
    },
    {
        "name": "pagehawk_list_documents",
        "description": "List documents PageHawk has indexed: path, type, size, modified time, "
                       "when indexed. Optional substring filter on the path/file name.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "filter": {"type": "string", "description": "Optional substring to match against the path."},
                "limit": {"type": "integer", "description": "Max documents to return (default 50, max 500)."},
            },
            "additionalProperties": False,
        },
        "handler": tool_list_documents,
    },
    {
        "name": "pagehawk_search",
        "description": "Full-text search across every document PageHawk has indexed (PDF pages, "
                       "Word/RTF paragraphs, Excel cells, text/Markdown lines). Returns matches "
                       "grouped by document with a snippet and an anchor (page number, cell "
                       "reference, or paragraph/line number) for each hit. By default, emails, "
                       "credit-card numbers, IBAN, API keys/tokens, and phone numbers in each "
                       "snippet are masked (see redact).",
        "inputSchema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Text to find (3+ characters; matched as a literal substring)."},
                "limit": {"type": "integer", "description": "Max documents to return (default 20, max 100)."},
                "hits_per_doc": {"type": "integer", "description": "Max hits per document (default 5, max 40)."},
                "redact": {"type": "boolean", "description": "Mask PII (email/card/IBAN/API-key/phone) in snippets. "
                                                              "Default true. Pass false for raw, unmasked text — "
                                                              "an explicit opt-out that is logged."},
            },
            "required": ["query"],
            "additionalProperties": False,
        },
        "handler": tool_search,
    },
    {
        "name": "pagehawk_get_document",
        "description": "Details for one indexed document by its full path (from list_documents "
                       "or search): type, size, page range, chunk count, and whether it has an "
                       "OCR cache (for scanned PDFs).",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "Full path to the document, as returned by list_documents or search."},
            },
            "required": ["path"],
            "additionalProperties": False,
        },
        "handler": tool_get_document,
    },
    {
        "name": "pagehawk_get_page_text",
        "description": "The indexed text for one page of a document (PDF), or the first N "
                       "indexed chunks for a non-paginated document (Word/Excel/text/Markdown) "
                       "when page is omitted. Capped at ~8KB per response; sets truncated=true "
                       "if cut off. By default, emails, credit-card numbers, IBAN, API keys/"
                       "tokens, and phone numbers in the returned text are masked (see redact).",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "Full path to the document."},
                "page": {"type": "integer", "description": "1-based PDF page number. Omit for non-PDF documents."},
                "limit": {"type": "integer", "description": "When page is omitted: max chunks to return (default 60, max 400)."},
                "redact": {"type": "boolean", "description": "Mask PII (email/card/IBAN/API-key/phone) in the "
                                                              "returned text. Default true. Pass false for raw, "
                                                              "unmasked text — an explicit opt-out that is logged."},
            },
            "required": ["path"],
            "additionalProperties": False,
        },
        "handler": tool_get_page_text,
    },
    {
        "name": "pagehawk_list_open",
        "description": "LIVE: every window and tab the running PageHawk app has open right now — "
                       "window id, kind (reader/grid/timeline/review/welcome), document path(s), "
                       "title, whether it is the key window, and for reader windows the current "
                       "page and page count as shown on screen. Requires the app to be running.",
        "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
        "handler": tool_list_open,
    },
    {
        "name": "pagehawk_open_document",
        "description": "LIVE: open a document (or folder) in the running PageHawk app, exactly as "
                       "a user double-click would — PDFs and Office docs open in a reader tab, "
                       "delimited text in the grid, folders in the review flow. Optional 1-based "
                       "page to land on for PDFs. Multi-tab: an already-busy window is never "
                       "reused; the document gets its own tab/window. Returns the window id.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "Full path to the document or folder to open."},
                "page": {"type": "integer", "description": "Optional 1-based page to open to (PDF readers)."},
            },
            "required": ["path"],
            "additionalProperties": False,
        },
        "handler": tool_open_document,
    },
    {
        "name": "pagehawk_goto_page",
        "description": "LIVE: turn the running app's reader to a page (1-based, clamped to the "
                       "document). Target a specific window by id (from pagehawk_list_open) or by "
                       "document path; with neither, the frontmost reader window is used. Replies "
                       "with the page the viewer actually landed on (verified).",
        "inputSchema": {
            "type": "object",
            "properties": {
                "page": {"type": "integer", "description": "1-based page number to show."},
                "window": {"type": "string", "description": "Optional window id from pagehawk_list_open."},
                "path": {"type": "string", "description": "Optional document path to pick the window by."},
            },
            "required": ["page"],
            "additionalProperties": False,
        },
        "handler": tool_goto_page,
    },
    {
        "name": "pagehawk_get_annotations",
        "description": "The user's annotations and comments on a PDF: highlights/underlines/"
                       "strikeouts WITH the text under each mark recovered from the page text "
                       "layer, plus notes and their contents, per page, with author when present. "
                       "Best source first: the live open reader window (captures unsaved marks), "
                       "else PageHawk's autosaved markup sidecar, else the file itself. Omit path "
                       "to use the document currently front in the app. This is how the user's "
                       "in-document comments reach the AI — check it when asked what the user "
                       "flagged, noted, or commented on. By default, emails, credit-card numbers, "
                       "IBAN, API keys/tokens, and phone numbers in the recovered text/notes/"
                       "markdown are masked (see redact).",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "Full path to the PDF. Omit for the currently front document."},
                "redact": {"type": "boolean", "description": "Mask PII (email/card/IBAN/API-key/phone) in the "
                                                              "recovered annotation text, notes, and markdown. "
                                                              "Default true. Pass false for raw, unmasked text — "
                                                              "an explicit opt-out that is logged."},
            },
            "additionalProperties": False,
        },
        "handler": tool_get_annotations,
    },
]

TOOL_BY_NAME = {t["name"]: t for t in TOOLS}


# --------------------------------------------------------------------------- JSON-RPC plumbing
def send(msg: dict) -> None:
    sys.stdout.write(json.dumps(msg, ensure_ascii=False) + "\n")
    sys.stdout.flush()


def reply(rid, result) -> None:
    send({"jsonrpc": "2.0", "id": rid, "result": result})


def reply_error(rid, code: int, message: str) -> None:
    send({"jsonrpc": "2.0", "id": rid, "error": {"code": code, "message": message}})


def handle_message(msg: dict) -> None:
    is_notification = msg.get("id") is None
    method, params, rid = msg.get("method"), msg.get("params") or {}, msg.get("id")
    try:
        if method == "initialize":
            reply(rid, {
                "protocolVersion": params.get("protocolVersion") or FALLBACK_PROTOCOL,
                "capabilities": {"tools": {"listChanged": False}},
                "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION},
                "instructions": "On-device access to PageHawk: its document search index (always "
                                "available) plus LIVE control of the running app. Start with "
                                "pagehawk_status; use pagehawk_search to find documents, then "
                                "pagehawk_get_document / pagehawk_get_page_text for detail. "
                                "Live tools (app must be running): pagehawk_list_open (what is on "
                                "screen, per tab, with pages), pagehawk_open_document, "
                                "pagehawk_goto_page, and pagehawk_get_annotations (the user's "
                                "highlights with recovered text + note contents — read these when "
                                "asked what the user commented on). A document that has not been "
                                "opened in PageHawk yet will not be indexed — say so rather than "
                                "guessing its contents. Privacy: pagehawk_search, "
                                "pagehawk_get_page_text, and pagehawk_get_annotations mask emails, "
                                "credit-card numbers, IBAN, API keys/tokens, and phone numbers by "
                                "default; pass redact=false on any of them for raw text.",
            })
            return
        if method in ("notifications/initialized", "initialized"):
            return
        if method == "ping":
            reply(rid, {})
            return
        if method == "tools/list":
            reply(rid, {"tools": [{k: t[k] for k in ("name", "description", "inputSchema")}
                                  for t in TOOLS]})
            return
        if method == "tools/call":
            tool = TOOL_BY_NAME.get(params.get("name"))
            if not tool:
                reply_error(rid, -32602, f"Unknown tool: {params.get('name')}")
                return
            try:
                result = tool["handler"](params.get("arguments") or {})
            except Exception as e:
                reply(rid, {"content": [{"type": "text",
                                         "text": f"Error running {params.get('name')}: {e}"}],
                            "isError": True})
                return
            reply(rid, {"content": [{"type": "text",
                                     "text": json.dumps(result, ensure_ascii=False, indent=2)}],
                        "structuredContent": result})
            return
        if not is_notification:
            reply_error(rid, -32601, f"Method not found: {method}")
    except Exception as e:
        log("handler exception:", e)
        if not is_notification:
            reply_error(rid, -32603, f"Internal error: {e}")


def main() -> None:
    log(f"ready; index at {index_db_path()}")
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except ValueError:
            log("bad JSON line skipped:", line[:120])
            continue
        if isinstance(msg, list):
            for m in msg:
                handle_message(m)
        else:
            handle_message(msg)


if __name__ == "__main__":
    main()
