#!/usr/bin/env python3
"""contractpin - pin the contract of every MCP tool you depend on.

Single-file CLI. Python 3.9+, standard library only.

    contractpin init   <server>   write contractpin.lock from a live MCP server
    contractpin verify [server]   re-enumerate and diff against the lock
    contractpin canon  <file>     print canonical JSON + digest (conformance harness)

Spec: SPEC.md in this directory. License: MIT.
"""

from __future__ import annotations

import argparse
import base64
import decimal
import hashlib
import io
import json
import os
import shlex
import subprocess
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

__version__ = "0.1.0"

LOCK_FORMAT = "1"
CANONICALIZATION = "contractpin-canonical-json/1"
DIGEST_ALGORITHM = "sha256"
DEFAULT_LOCK = "contractpin.lock"
DEFAULT_PROTOCOL_VERSION = "2025-06-18"

# --- SPEC: the contract fields, in the order the spec lists them -------------
REQUIRED_CONTRACT_FIELDS: Tuple[str, ...] = ("name", "description", "inputSchema")
OPTIONAL_CONTRACT_FIELDS: Tuple[str, ...] = ("outputSchema", "annotations")

EXIT_OK = 0
EXIT_DRIFT = 1
EXIT_ERROR = 2


class ContractpinError(Exception):
    """Fatal, user-facing error (exit code 2)."""


# ---------------------------------------------------------------------------
# 1. Canonical JSON  (see SPEC.md section "Canonicalization")
# ---------------------------------------------------------------------------


def _canonical_number(value: float) -> str:
    """Serialise a JSON number per RFC 8785 s3.2.2.3 (ECMAScript Number::toString)."""
    if value != value or value in (float("inf"), float("-inf")):
        raise ContractpinError("NaN and Infinity are not representable in canonical JSON")
    if value == 0:
        # -0.0 and 0.0 both serialise as "0"
        return "0"
    sign = "-" if value < 0 else ""
    magnitude = abs(value)

    # repr() gives the shortest decimal string that round-trips, which is the
    # same digit sequence ECMAScript uses.
    dec = decimal.Decimal(repr(magnitude))
    _, digit_tuple, exponent = dec.as_tuple()
    digits = list(digit_tuple)
    assert isinstance(exponent, int)
    while len(digits) > 1 and digits[-1] == 0:
        digits.pop()
        exponent += 1
    s = "".join(str(d) for d in digits)
    k = len(s)
    n = k + exponent  # value == s * 10**(n - k)

    if k <= n <= 21:
        return sign + s + "0" * (n - k)
    if 0 < n <= 21:
        return sign + s[:n] + "." + s[n:]
    if -6 < n <= 0:
        return sign + "0." + "0" * (-n) + s
    e = n - 1
    esign = "+" if e >= 0 else "-"
    mantissa = s if k == 1 else s[0] + "." + s[1:]
    return sign + mantissa + "e" + esign + str(abs(e))


def _canonical_string(value: str) -> str:
    # json.dumps(ensure_ascii=False) implements exactly the RFC 8785 s3.2.2.2
    # rules: \" \\ \b \f \n \r \t shorthands, \u00xx (lowercase hex) for the
    # remaining C0 controls, every other code point emitted literally.
    return json.dumps(value, ensure_ascii=False)


def _member_sort_key(key: str) -> bytes:
    # RFC 8785: sort object members by their UTF-16 code units. Comparing the
    # big-endian UTF-16 encoding byte-wise is equivalent.
    return key.encode("utf-16-be", errors="surrogatepass")


def _write_canonical(value: Any, out: io.StringIO) -> None:
    if value is True:
        out.write("true")
    elif value is False:
        out.write("false")
    elif value is None:
        out.write("null")
    elif isinstance(value, str):
        out.write(_canonical_string(value))
    elif isinstance(value, int):  # bool already handled above
        out.write(str(value))
    elif isinstance(value, float):
        out.write(_canonical_number(value))
    elif isinstance(value, (list, tuple)):
        out.write("[")
        for index, item in enumerate(value):
            if index:
                out.write(",")
            _write_canonical(item, out)
        out.write("]")
    elif isinstance(value, dict):
        out.write("{")
        keys = list(value.keys())
        for key in keys:
            if not isinstance(key, str):
                raise ContractpinError("canonical JSON requires string object keys")
        keys.sort(key=_member_sort_key)
        for index, key in enumerate(keys):
            if index:
                out.write(",")
            out.write(_canonical_string(key))
            out.write(":")
            _write_canonical(value[key], out)
        out.write("}")
    else:
        raise ContractpinError(f"value of type {type(value).__name__} is not JSON")


def canonical_json(value: Any) -> bytes:
    """Return the canonical UTF-8 serialisation of a JSON value."""
    buf = io.StringIO()
    _write_canonical(value, buf)
    return buf.getvalue().encode("utf-8")


def digest(value: Any) -> str:
    """`sha256:<64 lowercase hex>` over the canonical serialisation of `value`."""
    return "sha256:" + hashlib.sha256(canonical_json(value)).hexdigest()


# ---------------------------------------------------------------------------
# 2. Tool contracts and lock construction
# ---------------------------------------------------------------------------


def tool_contract(tool: Dict[str, Any]) -> Dict[str, Any]:
    """Project an MCP tool definition onto the fields contractpin pins."""
    contract: Dict[str, Any] = {}
    for field in REQUIRED_CONTRACT_FIELDS:
        contract[field] = tool.get(field)
    for field in OPTIONAL_CONTRACT_FIELDS:
        if tool.get(field) is not None:
            contract[field] = tool[field]
    return contract


def tool_entry(tool: Dict[str, Any]) -> Dict[str, Any]:
    contract = tool_contract(tool)
    return {
        "digest": digest(contract),
        # Advisory only. Never an input to `digest`; used to name what moved.
        "fields": {field: digest(contract[field]) for field in contract},
    }


def server_digest(tools: Dict[str, Dict[str, Any]]) -> str:
    return digest({name: entry["digest"] for name, entry in tools.items()})


def build_server_record(
    tools: Sequence[Dict[str, Any]], source: Dict[str, Any]
) -> Dict[str, Any]:
    entries: Dict[str, Dict[str, Any]] = {}
    for tool in tools:
        name = tool.get("name")
        if not isinstance(name, str) or not name:
            raise ContractpinError("server returned a tool with no name")
        if name in entries:
            raise ContractpinError(f"server returned duplicate tool name: {name!r}")
        entries[name] = tool_entry(tool)
    record = dict(source)
    record["tool_count"] = len(entries)
    record["digest"] = server_digest(entries)
    record["tools"] = entries
    return record


def new_lock() -> Dict[str, Any]:
    return {
        "contractpin": LOCK_FORMAT,
        "algorithm": DIGEST_ALGORITHM,
        "canonicalization": CANONICALIZATION,
        "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "servers": {},
    }


def read_lock(path: str) -> Dict[str, Any]:
    if not os.path.exists(path):
        raise ContractpinError(f"lock file not found: {path} (run `contractpin init` first)")
    with open(path, "r", encoding="utf-8") as handle:
        try:
            lock = json.load(handle)
        except json.JSONDecodeError as exc:
            raise ContractpinError(f"{path} is not valid JSON: {exc}") from exc
    if not isinstance(lock, dict) or "servers" not in lock:
        raise ContractpinError(f"{path} is not a contractpin lock file")
    if str(lock.get("contractpin")) != LOCK_FORMAT:
        raise ContractpinError(
            f"{path} is contractpin format {lock.get('contractpin')!r}, this CLI speaks {LOCK_FORMAT!r}"
        )
    return lock


def write_lock(path: str, lock: Dict[str, Any]) -> None:
    # The lock file's own formatting is deliberately git-friendly and is NOT
    # part of any digest. See SPEC.md.
    text = json.dumps(lock, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
    with open(path, "w", encoding="utf-8", newline="\n") as handle:
        handle.write(text)


# ---------------------------------------------------------------------------
# 3. MCP clients (stdio + streamable HTTP), stdlib only
# ---------------------------------------------------------------------------

CLIENT_INFO = {"name": "contractpin", "version": __version__}


class StdioClient:
    def __init__(
        self,
        argv: Sequence[str],
        env: Optional[Dict[str, str]] = None,
        cwd: Optional[str] = None,
        timeout: float = 30.0,
        protocol_version: str = DEFAULT_PROTOCOL_VERSION,
    ) -> None:
        self.argv = list(argv)
        self.timeout = timeout
        self.protocol_version = protocol_version
        merged = dict(os.environ)
        if env:
            merged.update(env)
        try:
            self.proc = subprocess.Popen(
                self.argv,
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                env=merged,
                cwd=cwd,
                text=True,
                encoding="utf-8",
                bufsize=1,
            )
        except OSError as exc:
            raise ContractpinError(f"cannot start {self.argv[0]!r}: {exc}") from exc
        self._next_id = 0

    def _send(self, message: Dict[str, Any]) -> None:
        assert self.proc.stdin is not None
        try:
            self.proc.stdin.write(json.dumps(message) + "\n")
            self.proc.stdin.flush()
        except (BrokenPipeError, ValueError) as exc:
            raise ContractpinError(f"server closed its input: {exc}") from exc

    def _read_result(self, request_id: int) -> Dict[str, Any]:
        assert self.proc.stdout is not None
        while True:
            line = self.proc.stdout.readline()
            if not line:
                stderr = ""
                if self.proc.stderr is not None:
                    stderr = self.proc.stderr.read() or ""
                raise ContractpinError(
                    "server exited before answering"
                    + (f"; stderr:\n{stderr.strip()}" if stderr.strip() else "")
                )
            line = line.strip()
            if not line:
                continue
            try:
                message = json.loads(line)
            except json.JSONDecodeError:
                continue  # servers that print junk on stdout: skip, don't crash
            if not isinstance(message, dict) or message.get("id") != request_id:
                continue  # notification or a response to something else
            if "error" in message:
                err = message["error"] or {}
                raise ContractpinError(
                    f"server error {err.get('code')}: {err.get('message')}"
                )
            return message.get("result") or {}

    def request(self, method: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        self._next_id += 1
        request_id = self._next_id
        self._send(
            {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params or {}}
        )
        return self._read_result(request_id)

    def notify(self, method: str, params: Optional[Dict[str, Any]] = None) -> None:
        self._send({"jsonrpc": "2.0", "method": method, "params": params or {}})

    def initialize(self) -> Dict[str, Any]:
        result = self.request(
            "initialize",
            {
                "protocolVersion": self.protocol_version,
                "capabilities": {},
                "clientInfo": CLIENT_INFO,
            },
        )
        self.notify("notifications/initialized")
        return result

    def close(self) -> None:
        try:
            if self.proc.stdin:
                self.proc.stdin.close()
        except OSError:
            pass
        try:
            self.proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self.proc.kill()
            self.proc.wait(timeout=5)
        for stream in (self.proc.stdout, self.proc.stderr):
            try:
                if stream:
                    stream.close()
            except OSError:
                pass


class HttpClient:
    """Minimal MCP streamable-HTTP client (JSON or SSE response bodies)."""

    def __init__(
        self,
        url: str,
        headers: Optional[Dict[str, str]] = None,
        timeout: float = 30.0,
        protocol_version: str = DEFAULT_PROTOCOL_VERSION,
    ) -> None:
        self.url = url
        self.headers = dict(headers or {})
        self.timeout = timeout
        self.protocol_version = protocol_version
        self.session_id: Optional[str] = None
        self._next_id = 0

    def _post(self, message: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        body = json.dumps(message).encode("utf-8")
        request = urllib.request.Request(self.url, data=body, method="POST")
        request.add_header("Content-Type", "application/json")
        request.add_header("Accept", "application/json, text/event-stream")
        request.add_header("User-Agent", f"contractpin/{__version__}")
        request.add_header("MCP-Protocol-Version", self.protocol_version)
        if self.session_id:
            request.add_header("Mcp-Session-Id", self.session_id)
        for key, value in self.headers.items():
            request.add_header(key, value)
        try:
            with urllib.request.urlopen(request, timeout=self.timeout) as response:
                session = response.headers.get("Mcp-Session-Id")
                if session:
                    self.session_id = session
                content_type = (response.headers.get("Content-Type") or "").lower()
                raw = response.read().decode("utf-8", errors="replace")
        except urllib.error.HTTPError as exc:
            detail = exc.read().decode("utf-8", errors="replace")[:400]
            raise ContractpinError(f"HTTP {exc.code} from {self.url}: {detail}") from exc
        except urllib.error.URLError as exc:
            raise ContractpinError(f"cannot reach {self.url}: {exc.reason}") from exc
        if not raw.strip():
            return None
        if "text/event-stream" in content_type:
            for event in _iter_sse_data(raw):
                try:
                    parsed = json.loads(event)
                except json.JSONDecodeError:
                    continue
                if isinstance(parsed, dict) and "id" in parsed:
                    return parsed
            return None
        return json.loads(raw)

    def request(self, method: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        self._next_id += 1
        message = {
            "jsonrpc": "2.0",
            "id": self._next_id,
            "method": method,
            "params": params or {},
        }
        response = self._post(message)
        if response is None:
            raise ContractpinError(f"empty response to {method}")
        if "error" in response:
            err = response["error"] or {}
            raise ContractpinError(f"server error {err.get('code')}: {err.get('message')}")
        return response.get("result") or {}

    def notify(self, method: str, params: Optional[Dict[str, Any]] = None) -> None:
        self._post({"jsonrpc": "2.0", "method": method, "params": params or {}})

    def initialize(self) -> Dict[str, Any]:
        result = self.request(
            "initialize",
            {
                "protocolVersion": self.protocol_version,
                "capabilities": {},
                "clientInfo": CLIENT_INFO,
            },
        )
        self.notify("notifications/initialized")
        return result

    def close(self) -> None:
        return None


def _iter_sse_data(raw: str) -> Iterable[str]:
    buffer: List[str] = []
    for line in raw.splitlines():
        if line.startswith("data:"):
            buffer.append(line[5:].lstrip())
        elif not line.strip():
            if buffer:
                yield "\n".join(buffer)
                buffer = []
    if buffer:
        yield "\n".join(buffer)


def list_tools(client: Any, page_limit: int = 200) -> List[Dict[str, Any]]:
    client.initialize()
    tools: List[Dict[str, Any]] = []
    cursor: Optional[str] = None
    for _ in range(page_limit):
        params: Dict[str, Any] = {}
        if cursor:
            params["cursor"] = cursor
        result = client.request("tools/list", params)
        page = result.get("tools") or []
        if not isinstance(page, list):
            raise ContractpinError("tools/list did not return a list")
        tools.extend(page)
        cursor = result.get("nextCursor")
        if not cursor:
            return tools
    raise ContractpinError(f"tools/list did not terminate after {page_limit} pages")


# ---------------------------------------------------------------------------
# 4. Target resolution
# ---------------------------------------------------------------------------


def split_command(target: str) -> List[str]:
    """Split a command string into argv, tolerating Windows paths."""
    if os.name == "nt":
        parts = shlex.split(target, posix=False)
        return [p[1:-1] if len(p) > 1 and p[0] == p[-1] == '"' else p for p in parts]
    return shlex.split(target)


def default_server_name(transport: str, target: Sequence[str] | str) -> str:
    if transport == "http":
        from urllib.parse import urlparse

        parsed = urlparse(str(target))
        return parsed.hostname or "server"
    argv = list(target) if not isinstance(target, str) else split_command(target)
    for token in reversed(argv):
        base = os.path.basename(token)
        stem = os.path.splitext(base)[0]
        if stem and not token.startswith("-"):
            return stem
    return "server"


def connect(source: Dict[str, Any], timeout: float, protocol_version: str) -> Any:
    transport = source.get("transport")
    if transport == "http":
        return HttpClient(
            source["url"],
            headers=source.get("headers"),
            timeout=timeout,
            protocol_version=protocol_version,
        )
    if transport == "stdio":
        return StdioClient(
            source["command"],
            env=source.get("env"),
            cwd=source.get("cwd"),
            timeout=timeout,
            protocol_version=protocol_version,
        )
    raise ContractpinError(f"unknown transport {transport!r}")


def enumerate_tools(source: Dict[str, Any], timeout: float, protocol_version: str) -> List[Dict[str, Any]]:
    if source.get("transport") == "file":
        return load_tools_file(source["path"])
    client = connect(source, timeout, protocol_version)
    try:
        return list_tools(client)
    finally:
        client.close()


def load_tools_file(path: str) -> List[Dict[str, Any]]:
    with open(path, "r", encoding="utf-8") as handle:
        data = json.load(handle)
    if isinstance(data, dict):
        data = data.get("tools", data.get("result", {}).get("tools") if isinstance(data.get("result"), dict) else None)
    if not isinstance(data, list):
        raise ContractpinError("expected a JSON array of tools, or an object with a `tools` array")
    return data


def source_from_args(args: argparse.Namespace) -> Dict[str, Any]:
    """Build the connection record stored in (and replayed from) the lock."""
    if getattr(args, "from_json", None):
        return {"transport": "file", "path": args.from_json}

    target: Optional[str] = args.server
    argv: List[str] = list(getattr(args, "argv", []) or [])

    if argv:
        if target:
            argv = [target] + argv
        return _stdio_source(argv, args)
    if not target:
        raise ContractpinError("no server given: pass a URL, a command, or --from-json")
    if target.startswith("http://") or target.startswith("https://"):
        headers = _parse_headers(getattr(args, "header", None) or [])
        source: Dict[str, Any] = {"transport": "http", "url": target}
        if headers:
            source["headers"] = headers
        return source
    return _stdio_source(split_command(target), args)


def _stdio_source(argv: Sequence[str], args: argparse.Namespace) -> Dict[str, Any]:
    source: Dict[str, Any] = {"transport": "stdio", "command": list(argv)}
    env = {}
    for item in getattr(args, "env", None) or []:
        key, _, value = item.partition("=")
        env[key] = value
    if env:
        source["env"] = env
    if getattr(args, "cwd", None):
        source["cwd"] = args.cwd
    return source


def _parse_headers(items: Sequence[str]) -> Dict[str, str]:
    headers: Dict[str, str] = {}
    for item in items:
        key, sep, value = item.partition(":")
        if not sep:
            raise ContractpinError(f"bad --header {item!r}, expected 'Name: value'")
        headers[key.strip()] = value.strip()
    return headers


def connection_summary(source: Dict[str, Any]) -> str:
    transport = source.get("transport")
    if transport == "http":
        return f"http: {source.get('url')}"
    if transport == "file":
        return f"file: {source.get('path')}"
    return "stdio: " + " ".join(source.get("command", []))


# ---------------------------------------------------------------------------
# 5. Commands
# ---------------------------------------------------------------------------


def cmd_init(args: argparse.Namespace) -> int:
    source = source_from_args(args)
    name = args.name or default_server_name(
        source["transport"], source.get("url") or source.get("command") or source.get("path", "")
    )
    tools = enumerate_tools(source, args.timeout, args.protocol_version)
    if not tools:
        raise ContractpinError("server exposed zero tools; refusing to write an empty lock")

    stored = {k: v for k, v in source.items() if k != "headers"}  # never persist secrets
    record = build_server_record(tools, stored)

    lock = new_lock()
    if os.path.exists(args.lock):
        existing = read_lock(args.lock)
        if name in existing["servers"] and not args.force:
            raise ContractpinError(
                f"{args.lock} already pins server {name!r}; re-run with --force to overwrite"
            )
        lock = existing
        lock["generated_at"] = new_lock()["generated_at"]
    lock["servers"][name] = record
    write_lock(args.lock, lock)

    print(f"contractpin {__version__}: pinned {len(tools)} tools from {connection_summary(source)}")
    print(f"  server : {name}")
    print(f"  digest : {record['digest']}")
    print(f"  lock   : {args.lock}")
    if args.print_tools:
        for tool_name in sorted(record["tools"]):
            print(f"    {record['tools'][tool_name]['digest']}  {tool_name}")
    print("\nCommit the lock file. `contractpin verify` in CI now fails on any change.")
    return EXIT_OK


def _diff_server(
    pinned: Dict[str, Any], live_tools: Sequence[Dict[str, Any]]
) -> Dict[str, Any]:
    live: Dict[str, Dict[str, Any]] = {}
    for tool in live_tools:
        name = tool.get("name")
        if isinstance(name, str) and name:
            live[name] = tool_entry(tool)

    pinned_tools: Dict[str, Any] = pinned.get("tools", {})
    added = sorted(set(live) - set(pinned_tools))
    removed = sorted(set(pinned_tools) - set(live))
    drifted: List[Dict[str, Any]] = []
    unchanged: List[str] = []
    for name in sorted(set(pinned_tools) & set(live)):
        if pinned_tools[name].get("digest") == live[name]["digest"]:
            unchanged.append(name)
            continue
        old_fields = pinned_tools[name].get("fields") or {}
        new_fields = live[name]["fields"]
        changed = sorted(
            set(old_fields) ^ set(new_fields)
            | {k for k in set(old_fields) & set(new_fields) if old_fields[k] != new_fields[k]}
        )
        drifted.append(
            {
                "name": name,
                "pinned_digest": pinned_tools[name].get("digest"),
                "live_digest": live[name]["digest"],
                "changed_fields": changed,
            }
        )
    return {
        "added": added,
        "removed": removed,
        "drifted": drifted,
        "unchanged": unchanged,
        "live_digest": server_digest(live),
        "pinned_digest": pinned.get("digest"),
    }


def cmd_verify(args: argparse.Namespace) -> int:
    lock = read_lock(args.lock)
    servers: Dict[str, Any] = lock.get("servers") or {}
    if not servers:
        raise ContractpinError(f"{args.lock} pins no servers")

    wanted = args.server or []
    for name in wanted:
        if name not in servers:
            raise ContractpinError(
                f"{args.lock} does not pin a server named {name!r}; it pins: "
                + ", ".join(sorted(servers))
            )
    selected = wanted or sorted(servers)

    report: Dict[str, Any] = {"lock": args.lock, "servers": {}, "ok": True}
    failures = 0

    for name in selected:
        pinned = servers[name]
        source = {k: v for k, v in pinned.items() if k in ("transport", "url", "command", "env", "cwd", "path")}
        if args.from_json:
            source = {"transport": "file", "path": args.from_json}
        headers = _parse_headers(getattr(args, "header", None) or [])
        if headers and source.get("transport") == "http":
            source["headers"] = headers
        try:
            live_tools = enumerate_tools(source, args.timeout, args.protocol_version)
        except ContractpinError as exc:
            report["servers"][name] = {"error": str(exc)}
            report["ok"] = False
            failures += 1
            if not args.json:
                print(f"server: {name} ({connection_summary(source)})")
                print(f"  ERROR   {exc}")
                print()
            continue

        result = _diff_server(pinned, live_tools)
        fatal = bool(result["drifted"]) or bool(result["removed"])
        if result["added"] and not args.allow_new:
            fatal = True
        result["ok"] = not fatal
        report["servers"][name] = result
        if fatal:
            report["ok"] = False
            failures += 1

        if not args.json:
            print(f"server: {name} ({connection_summary(source)})")
            for item in result["drifted"]:
                fields = ", ".join(item["changed_fields"]) or "digest"
                print(f"  DRIFT   {item['name']}  ({fields} changed)")
            for tool_name in result["added"]:
                label = "ADDED " if not args.allow_new else "new   "
                print(f"  {label}  {tool_name}")
            for tool_name in result["removed"]:
                print(f"  REMOVED {tool_name}")
            if args.verbose:
                for tool_name in result["unchanged"]:
                    print(f"  ok      {tool_name}")
            else:
                print(f"  ok      {len(result['unchanged'])} tools unchanged")
            print()

    if args.json:
        print(json.dumps(report, indent=2, sort_keys=True))
    else:
        if report["ok"]:
            print(f"PASS: {len(selected)} server(s) match {args.lock}")
        else:
            print(f"FAIL: {failures} of {len(selected)} server(s) drifted from {args.lock}")
            print("Review the change, then re-pin with `contractpin init --force` if it is legitimate.")
    return EXIT_OK if report["ok"] else EXIT_DRIFT


def cmd_canon(args: argparse.Namespace) -> int:
    if args.file == "-":
        data = json.load(sys.stdin)
    else:
        with open(args.file, "r", encoding="utf-8") as handle:
            data = json.load(handle)
    if args.contract:
        data = tool_contract(data)
    raw = canonical_json(data)
    if args.bytes:
        sys.stdout.write(base64.b64encode(raw).decode("ascii") + "\n")
    elif hasattr(sys.stdout, "buffer"):
        sys.stdout.buffer.write(raw + b"\n")
        sys.stdout.buffer.flush()
    else:
        sys.stdout.write(raw.decode("utf-8") + "\n")
    sys.stdout.flush()
    print("sha256:" + hashlib.sha256(raw).hexdigest(), file=sys.stderr)
    return EXIT_OK


# ---------------------------------------------------------------------------
# 6. CLI wiring
# ---------------------------------------------------------------------------


def _add_common(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--lock", default=DEFAULT_LOCK, help="lock file path (default: contractpin.lock)")
    parser.add_argument("--timeout", type=float, default=30.0, help="per-request timeout in seconds")
    parser.add_argument(
        "--protocol-version",
        default=DEFAULT_PROTOCOL_VERSION,
        help=f"MCP protocol version to advertise (default: {DEFAULT_PROTOCOL_VERSION})",
    )
    parser.add_argument(
        "--from-json",
        metavar="FILE",
        help="read the tool list from a JSON file instead of connecting",
    )
    parser.add_argument(
        "--header",
        action="append",
        metavar="'Name: value'",
        help="extra HTTP header (repeatable); never written to the lock",
    )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="contractpin",
        description="Pin the contract (name + description + schemas) of every MCP tool you depend on.",
    )
    parser.add_argument("--version", action="version", version=f"contractpin {__version__}")
    sub = parser.add_subparsers(dest="command", required=True)

    init = sub.add_parser("init", help="write contractpin.lock from a live MCP server")
    init.add_argument("server", nargs="?", help="https URL, or the command that starts a stdio server")
    init.add_argument("argv", nargs=argparse.REMAINDER, help="extra argv after `--` for stdio servers")
    init.add_argument("--name", help="name for this server in the lock (default: derived)")
    init.add_argument("--env", action="append", metavar="K=V", help="env var for a stdio server (repeatable)")
    init.add_argument("--cwd", help="working directory for a stdio server")
    init.add_argument("--force", action="store_true", help="overwrite an existing entry")
    init.add_argument("--print-tools", action="store_true", help="list each pinned tool and digest")
    _add_common(init)
    init.set_defaults(func=cmd_init)

    verify = sub.add_parser("verify", help="re-enumerate and diff against the lock")
    verify.add_argument("server", nargs="*", help="server name(s) from the lock (default: all)")
    verify.add_argument("--allow-new", action="store_true", help="do not fail on tools added since pinning")
    verify.add_argument("--json", action="store_true", help="machine-readable report on stdout")
    verify.add_argument("--verbose", action="store_true", help="list every unchanged tool")
    _add_common(verify)
    verify.set_defaults(func=cmd_verify)

    canon = sub.add_parser("canon", help="print canonical JSON and digest of a JSON file (use `-` for stdin)")
    canon.add_argument("file")
    canon.add_argument("--contract", action="store_true", help="project onto the pinned contract fields first")
    canon.add_argument("--bytes", action="store_true", help="print base64 of the canonical bytes")
    canon.set_defaults(func=cmd_canon)

    return parser


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = build_parser().parse_args(argv)
    if getattr(args, "argv", None):
        args.argv = [a for a in args.argv if a != "--"]
    try:
        return int(args.func(args))
    except ContractpinError as exc:
        print(f"contractpin: error: {exc}", file=sys.stderr)
        return EXIT_ERROR
    except KeyboardInterrupt:
        return EXIT_ERROR


if __name__ == "__main__":
    sys.exit(main())