#!/usr/bin/python3
"""hilt-mcp — CUT D: hilt's Shape C door over MCP stdio.

Ten tools, one per CLI verb: find → describe → run, plus orient, owners,
permissions, vocab, inventory, traps, coverage. Resident schema cost (bytes/4, Sheath's schema_cost.py
heuristic):

    C  lazy core 10 tools  592 tok   budget 600

Both numbers are asserted by tools/test-hilt-mcp.py against a live tools/list,
because a header is a claim like any other (Sheath's said 313 while serving 660).

THE DOOR HOLDS NO LOGIC. Every call is `hilt <verb> --json [flags] -- <args>`
against the `hilt` BESIDE this file (else PATH), and the answer is the CLI's
own JSON, verbatim. The door never opens the index, never renders, never maps
an exit code to a verdict: `hilt --json` carries `verdict` in every answer, so
the exit code simply is not relayed. Falsifier (spec §7 CUT D): any tool
result that differs from `hilt --json` for the same argv — the suite compares
them call by call. What the door adds is transport only:

  * `--` before every caller-supplied value, so an argument can never become a
    flag (`args: ["--no-reload"]` is refused as a bad enum, not honoured).
  * HILT_CALLER=mcp, so runs.jsonl attributes the run to this door.
  * a contract handshake: `hilt contract` must equal HILT_CONTRACT below, or
    every tool answers UNAVAILABLE. A stale installed copy would otherwise give
    old verdicts with nothing in them saying so.
  * isError ONLY when the CLI did not answer (missing, crashed, timed out, not
    JSON). A refusal is an answer: verdict `refused`, isError unset.

Stdlib only. stdout carries PROTOCOL ONLY; diagnostics go to stderr.
"""
import json
import os
import subprocess
import sys

VERSION = "0.1.0"
HILT_CONTRACT = "4"
PROTOCOL = "2025-06-18"
SUPPORTED = {"2024-11-05", "2025-03-26", "2025-06-18"}
TIMEOUT = 180  # run polls up to 10s; `shortcuts run` may take 120

S = {"type": "string"}
TOOLS = [
    {"name": "hilt_find",
     "description": "Search what this Mac can be asked to do (settings, panes, app launch, CLI "
                    "state, Shortcuts) by intent words. Ranked. Start here; never guess an id.",
     "inputSchema": {"type": "object", "properties": {"query": S, "all": {"type": "boolean"},
                                                      "criteria": {"type": "boolean"}},
                     "required": ["query"]}},
    {"name": "hilt_describe",
     "description": "One id's typed params (enums, ranges, path rules), authority, probe and "
                    "reload step. Read before hilt_run.",
     "inputSchema": {"type": "object", "properties": {"id": S}, "required": ["id"]}},
    {"name": "hilt_run",
     "description": "Run one id; verdict from a probe before and after: verified | mismatched | "
                    "unverifiable | failed | refused | query. Refuses sudo, unknown ids and untyped "
                    "args before acting. A launch is verified by pid; windows are Zil's.",
     "inputSchema": {"type": "object",
                     "properties": {"id": S, "args": {"type": "array", "items": S},
                                    "no_reload": {"type": "boolean"}},
                     "required": ["id"]}},
    {"name": "hilt_orient",
     "description": "Index sizes and holes (Spotlight, Full Disk Access, missing CLIs).",
     "inputSchema": {"type": "object", "properties": {}}},
    {"name": "hilt_owners",
     "description": "Who owns a preference domain or plist (OWNED | CONTESTED | UNOWNED | VACUOUS | "
                    "UNVERIFIABLE | REFUSED). Ask before any defaults write.",
     "inputSchema": {"type": "object", "properties": {"target": S}, "required": ["target"]}},
    {"name": "hilt_traps",
     "description": "macOS traps that make ordinary commands lie, each with a falsifier.",
     "inputSchema": {"type": "object", "properties": {"id": S}}},
    {"name": "hilt_permissions",
     "description": "Privacy (TCC) decisions for an app, or every denial if no app: why it keeps "
                    "prompting and the exact Settings pane to fix it. Read-only.",
     "inputSchema": {"type": "object", "properties": {"app": S}}},
    {"name": "hilt_vocab",
     "description": "Budgeted glossary (user app names, network jargon) to bias speech "
                    "recognition: vlab listen --context.",
     "inputSchema": {"type": "object", "properties": {}}},
    {"name": "hilt_inventory",
     "description": "This Mac's hardware, recommended tools (missing, or lacking a needed flag, "
                    "with the brew line) and local models.",
     "inputSchema": {"type": "object", "properties": {}}},
    {"name": "hilt_coverage",
     "description": "Every index count with its denominator.",
     "inputSchema": {"type": "object", "properties": {}}},
]


def argv_for(name, a):
    """Tool call → CLI argv. Shape only: which verb, which flags, which
    positionals. Values are never inspected — typing them is the CLI's job."""
    flags, pos = [], []
    if name == "hilt_find":
        pos = [a["query"]]
        flags = (["--all"] if a.get("all") else []) + (["--criteria"] if a.get("criteria") else [])
    elif name == "hilt_describe":
        pos = [a["id"]]
    elif name == "hilt_run":
        pos = [a["id"], *a.get("args", [])]
        flags = ["--no-reload"] if a.get("no_reload") else []
    elif name == "hilt_owners":
        pos = [a["target"]]
    elif name == "hilt_permissions":
        pos = [a["app"]] if a.get("app") else []
    elif name == "hilt_traps":
        pos = [a["id"]] if a.get("id") else []
    return [name[len("hilt_"):], "--json", *flags, "--", *pos]


def check_args(tool, a):
    """Schema conformance only (JSON-RPC -32602 territory), never meaning."""
    props, req = tool["inputSchema"]["properties"], tool["inputSchema"].get("required", [])
    for k in req:
        if k not in a:
            return f"missing required argument {k!r}"
    for k, v in a.items():
        t = props.get(k)
        if t is None:
            return f"unknown argument {k!r}"
        ok = {"string": lambda x: isinstance(x, str), "boolean": lambda x: isinstance(x, bool),
              "array": lambda x: isinstance(x, list) and all(isinstance(i, str) for i in x)}[t["type"]]
        if not ok(v):
            return f"argument {k!r} must be {t['type']}" + (" of strings" if t["type"] == "array" else "")
    return None


def hilt_bin():
    """The hilt BESIDE this file (installed pair, or the tracked copy in a
    worktree, so the suite tests the code being edited), else PATH."""
    sib = os.path.join(os.path.dirname(os.path.abspath(__file__)), "hilt")
    return sib if os.path.exists(sib) else "hilt"


def unavailable(why):
    return {"verdict": "UNAVAILABLE", "detail": why + " This is a door failure, NOT an answer "
            "about your query."}


def cli(argv):
    """(payload, ok). ok=False means the CLI did not answer."""
    env = dict(os.environ, HILT_CALLER="mcp")
    try:
        r = subprocess.run([hilt_bin(), *argv], capture_output=True, text=True,
                           timeout=TIMEOUT, env=env, stdin=subprocess.DEVNULL)
    except FileNotFoundError:
        return unavailable("hilt is not beside this server nor on PATH."), False
    except PermissionError:
        return unavailable(f"{hilt_bin()} is not executable (chmod +x; a copy or clone may "
                           "have dropped the mode bits)."), False
    except subprocess.TimeoutExpired:
        return unavailable(f"hilt {argv[0]} timed out after {TIMEOUT}s."), False
    try:
        payload = json.loads(r.stdout)
    except json.JSONDecodeError:
        return unavailable(f"hilt {argv[0]} exited {r.returncode} without JSON: "
                           f"{(r.stdout or r.stderr).strip()[:600]}"), False
    if not isinstance(payload, dict) or "verdict" not in payload:
        return unavailable(f"hilt {argv[0]} answered without a verdict (contract drift?)."), False
    return payload, True


_contract = {}


def contract_refusal():
    if "v" not in _contract:
        p, ok = cli(["contract", "--json"])
        got = p.get("contract") if ok else None
        _contract["v"] = None if got == HILT_CONTRACT else unavailable(
            f"hilt speaks contract {got}; this door was written against {HILT_CONTRACT}. "
            "Reinstall both from the hilt project (`install -m755 tools/hilt tools/hilt-mcp "
            "~/.local/bin/`), then `hilt build`.")
    return _contract["v"]


def call(name, a):
    bad = contract_refusal()
    if bad:
        return bad, False
    return cli(argv_for(name, a))


def handle(msg):
    mid, method = msg.get("id"), msg.get("method")
    if mid is None:
        return None  # notification

    def ok(result):
        return {"jsonrpc": "2.0", "id": mid, "result": result}

    def err(code, message):
        return {"jsonrpc": "2.0", "id": mid, "error": {"code": code, "message": message}}

    if method == "initialize":
        want = (msg.get("params") or {}).get("protocolVersion")
        return ok({"protocolVersion": want if want in SUPPORTED else PROTOCOL,
                   "capabilities": {"tools": {}},
                   "serverInfo": {"name": "hilt", "version": VERSION}})
    if method == "ping":
        return ok({})
    if method == "tools/list":
        return ok({"tools": TOOLS})
    if method == "tools/call":
        p = msg.get("params") or {}
        tool = next((t for t in TOOLS if t["name"] == p.get("name")), None)
        if not tool:
            return err(-32602, f"unknown tool {p.get('name')!r}")
        a = p.get("arguments") or {}
        why = check_args(tool, a) if isinstance(a, dict) else "arguments must be an object"
        if why:
            return err(-32602, why)
        payload, answered = call(tool["name"], a)
        body = {"content": [{"type": "text", "text": json.dumps(payload, default=str)}]}
        if not answered:
            body["isError"] = True
        return ok(body)
    return err(-32601, f"method not found: {method}")


def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except json.JSONDecodeError as e:
            sys.stderr.write(f"[hilt-mcp] unparseable line dropped: {e}\n")
            continue
        resp = handle(msg)
        if resp is not None:
            sys.stdout.write(json.dumps(resp) + "\n")
            sys.stdout.flush()


if __name__ == "__main__":
    main()
