copilot_xhigh_patch.py

#!/usr/bin/env python3
"""Patch an installed @github/copilot CLI so xhigh effort is not filtered out.

Default mode is dry-run. Use --apply to modify files.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import platform
import re
import shutil
import stat
import subprocess
import sys
import tempfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any


PACKAGE_NAME = "@github/copilot"
MANIFEST_NAME = "xhigh-patch-manifest.json"
PATCH_MARKER = "# copilot-xhigh-patch native wrapper"
LOADER_MARKER = "// copilot-xhigh-patch force-js-loader"

EXACT_XHIGH_FILTER = b's=s.filter(l=>l!=="xhigh")'
EXACT_NEVER_FILTER = b's=s.filter(l=>l!=="never")'

FILTER_XHIGH_RE = re.compile(
    rb'\.filter\(\s*([A-Za-z_$][A-Za-z0-9_$]*)\s*=>\s*\1\s*!==\s*([\'"])xhigh\2\s*\)'
)
FILTER_NEVER_RE = re.compile(
    rb'\.filter\(\s*([A-Za-z_$][A-Za-z0-9_$]*)\s*=>\s*\1\s*!==\s*([\'"])never\2\s*\)'
)

SEMANTIC_ANCHORS = (
    b"supportedReasoningEfforts",
    b"xhigh",
    b"business",
    b"enterprise",
)


class PatchError(RuntimeError):
    """Raised when the patch cannot be applied safely."""


@dataclass
class PatchEntry:
    path: str
    strategy: str
    status: str
    backup_path: str | None = None
    original_sha256: str | None = None
    patched_sha256: str | None = None
    details: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class PatchProfile:
    name: str
    known_version: bool
    expected_js_targets: tuple[str, ...] = ("app.js", "sdk/index.js")


@dataclass
class PatchContext:
    root: Path
    version: str
    profile: PatchProfile
    dry_run: bool
    force: bool
    entries: list[PatchEntry] = field(default_factory=list)

    def add(self, entry: PatchEntry) -> None:
        self.entries.append(entry)


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def sha256_file(path: Path) -> str:
    return sha256_bytes(path.read_bytes())


def load_json(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise PatchError(f"Missing file: {path}") from exc
    except json.JSONDecodeError as exc:
        raise PatchError(f"Invalid JSON in {path}: {exc}") from exc
    if not isinstance(value, dict):
        raise PatchError(f"Expected JSON object in {path}")
    return value


def is_package_root(path: Path) -> bool:
    package_json = path / "package.json"
    if not package_json.is_file():
        return False
    try:
        return load_json(package_json).get("name") == PACKAGE_NAME
    except PatchError:
        return False


def find_parent_package_root(path: Path) -> Path | None:
    current = path if path.is_dir() else path.parent
    for candidate in (current, *current.parents):
        if is_package_root(candidate):
            return candidate
    return None


def npm_root_global() -> Path | None:
    try:
        proc = subprocess.run(
            ["npm", "root", "-g"],
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=20,
        )
    except (OSError, subprocess.SubprocessError):
        return None
    output = proc.stdout.strip()
    return Path(output) if output else None


def find_package_root(explicit: str | None) -> Path:
    candidates: list[Path] = []

    if explicit:
        candidates.append(Path(explicit).expanduser())

    which = shutil.which("copilot")
    if which:
        candidates.append(Path(which))
        try:
            candidates.append(Path(which).resolve())
        except OSError:
            pass

    npm_root = npm_root_global()
    if npm_root:
        candidates.append(npm_root / "@github" / "copilot")

    seen: set[Path] = set()
    for candidate in candidates:
        try:
            resolved = candidate.resolve()
        except OSError:
            resolved = candidate.absolute()
        if resolved in seen:
            continue
        seen.add(resolved)

        if is_package_root(resolved):
            return resolved
        parent = find_parent_package_root(resolved)
        if parent:
            return parent

    searched = ", ".join(str(c) for c in candidates) or "(none)"
    raise PatchError(f"Could not find @github/copilot package root. Searched: {searched}")


def detect_node_platform() -> str:
    if sys.platform.startswith("darwin"):
        return "darwin"
    if sys.platform.startswith("linux"):
        return "linux"
    if sys.platform.startswith(("win32", "cygwin", "msys")):
        return "win32"
    return sys.platform


def detect_node_arch() -> str:
    machine = platform.machine().lower()
    if machine in {"x86_64", "amd64"}:
        return "x64"
    if machine in {"arm64", "aarch64"}:
        return "arm64"
    return machine


def atomic_write(path: Path, data: bytes, mode: int | None = None) -> None:
    tmp_path: Path | None = None
    fd = -1
    try:
        fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
        tmp_path = Path(tmp_name)
        with os.fdopen(fd, "wb") as handle:
            fd = -1
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        if mode is not None:
            os.chmod(tmp_path, mode)
        os.replace(tmp_path, path)
    finally:
        if fd != -1:
            os.close(fd)
        if tmp_path and tmp_path.exists():
            try:
                tmp_path.unlink()
            except OSError:
                pass


def backup_once(path: Path, backup_path: Path | None = None) -> Path:
    backup = backup_path or path.with_name(path.name + ".xhigh-patch.bak")
    if backup.exists():
        return backup
    shutil.copy2(path, backup)
    return backup


def semantic_anchor_report(data: bytes) -> dict[str, bool]:
    return {anchor.decode("ascii"): anchor in data for anchor in SEMANTIC_ANCHORS}


def select_patch_profile(version: str) -> PatchProfile:
    known_versions = {"1.0.40", "1.0.41-1"}
    if version in known_versions:
        return PatchProfile(name=f"known-{version}", known_version=True)
    return PatchProfile(name=f"generic-unknown-{version}", known_version=False)


def replace_regex_candidate(data: bytes, path: Path) -> tuple[bytes, dict[str, Any]]:
    matches = list(FILTER_XHIGH_RE.finditer(data))
    if len(matches) != 1:
        contexts = []
        for match in matches[:5]:
            start = max(0, match.start() - 80)
            end = min(len(data), match.end() + 80)
            contexts.append(data[start:end].decode("utf-8", errors="replace"))
        raise PatchError(
            f"{path} has {len(matches)} xhigh filter candidates; refusing to guess"
            + (f". Candidates: {contexts}" if contexts else "")
        )

    match = matches[0]
    fragment = match.group(0)
    patched_fragment = fragment.replace(b"xhigh", b"never", 1)
    patched = data[: match.start()] + patched_fragment + data[match.end() :]
    return patched, {"regex_match": fragment.decode("utf-8", errors="replace")}


def patch_js_file(ctx: PatchContext, relative_path: str) -> None:
    path = ctx.root / relative_path
    if not path.is_file():
        raise PatchError(f"Missing target JS file: {path}")

    data = path.read_bytes()
    original_sha = sha256_bytes(data)
    old_count = data.count(EXACT_XHIGH_FILTER)
    new_count = data.count(EXACT_NEVER_FILTER)
    regex_old_count = len(FILTER_XHIGH_RE.findall(data))
    regex_new_count = len(FILTER_NEVER_RE.findall(data))
    anchors = semantic_anchor_report(data)

    details: dict[str, Any] = {
        "profile": ctx.profile.name,
        "exact_xhigh_count": old_count,
        "exact_never_count": new_count,
        "regex_xhigh_count": regex_old_count,
        "regex_never_count": regex_new_count,
        "anchors": anchors,
    }

    if old_count == 0 and regex_old_count == 0 and (new_count > 0 or regex_new_count > 0):
        ctx.add(
            PatchEntry(
                path=str(path),
                strategy="js-byte-replace",
                status="already-patched",
                original_sha256=original_sha,
                patched_sha256=original_sha,
                details=details,
            )
        )
        return

    if old_count == 1:
        patched = data.replace(EXACT_XHIGH_FILTER, EXACT_NEVER_FILTER, 1)
        strategy = "js-byte-replace"
    elif old_count > 1 and ctx.force:
        patched = data.replace(EXACT_XHIGH_FILTER, EXACT_NEVER_FILTER)
        strategy = "js-byte-replace-force"
    elif old_count > 1:
        raise PatchError(f"{path} has {old_count} exact xhigh filters; refusing to guess")
    else:
        missing = [name for name, present in anchors.items() if not present]
        if missing and not ctx.force:
            raise PatchError(f"{path} missing semantic anchors {missing}; refusing regex fallback")
        patched, regex_details = replace_regex_candidate(data, path)
        details.update(regex_details)
        strategy = "js-regex-filter"

    patched_sha = sha256_bytes(patched)
    if patched_sha == original_sha:
        raise PatchError(f"Patch produced no changes for {path}")

    backup_path = path.with_name(path.name + ".xhigh-patch.bak")
    if not ctx.dry_run:
        backup_once(path, backup_path)
        mode = stat.S_IMODE(path.stat().st_mode)
        atomic_write(path, patched, mode=mode)

    ctx.add(
        PatchEntry(
            path=str(path),
            strategy=strategy,
            status="would-patch" if ctx.dry_run else "patched",
            backup_path=str(backup_path),
            original_sha256=original_sha,
            patched_sha256=patched_sha,
            details=details,
        )
    )


def find_native_binary(root: Path) -> Path | None:
    package_name = f"copilot-{detect_node_platform()}-{detect_node_arch()}"
    package_dir = root / "node_modules" / "@github" / package_name
    candidates = [package_dir / "copilot"]
    if detect_node_platform() == "win32":
        candidates.insert(0, package_dir / "copilot.exe")
    for candidate in candidates:
        if candidate.exists():
            return candidate
    return None


def posix_native_wrapper(root: Path) -> bytes:
    # The optional package directory is <root>/node_modules/@github/copilot-<platform>-<arch>.
    return f"""#!/bin/sh
{PATCH_MARKER}
set -eu
SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
ROOT=$(CDPATH= cd -- "$SCRIPT_DIR/../../.." && pwd)
exec node "$ROOT/index.js" "$@"
""".encode("utf-8")


def patch_posix_native_binary(ctx: PatchContext, native_path: Path) -> None:
    data = native_path.read_bytes()
    original_sha = sha256_bytes(data)
    marker_bytes = PATCH_MARKER.encode("utf-8")

    if data.startswith(b"#!") and marker_bytes in data[:512]:
        ctx.add(
            PatchEntry(
                path=str(native_path),
                strategy="native-wrapper",
                status="already-patched",
                original_sha256=original_sha,
                patched_sha256=original_sha,
                details={"platform": detect_node_platform(), "arch": detect_node_arch()},
            )
        )
        return

    backup_path = native_path.with_name(native_path.name + ".xhigh-patch.original")
    wrapper = posix_native_wrapper(ctx.root)
    patched_sha = sha256_bytes(wrapper)

    if not ctx.dry_run:
        backup_once(native_path, backup_path)
        atomic_write(native_path, wrapper, mode=0o755)

    ctx.add(
        PatchEntry(
            path=str(native_path),
            strategy="native-wrapper",
            status="would-patch" if ctx.dry_run else "patched",
            backup_path=str(backup_path),
            original_sha256=original_sha,
            patched_sha256=patched_sha,
            details={"platform": detect_node_platform(), "arch": detect_node_arch()},
        )
    )


def force_js_loader_content() -> bytes:
    return f"""#!/usr/bin/env node
{LOADER_MARKER}

import {{ dirname, join }} from "node:path";
import {{ fileURLToPath, pathToFileURL }} from "node:url";

const root = dirname(fileURLToPath(import.meta.url));
const major = parseInt(process.versions.node.split(".")[0], 10);

if (major < 24) {{
  console.error(`GitHub Copilot CLI requires Node.js v24 or higher. Currently using v${{process.versions.node}}.`);
  process.exit(1);
}}

try {{
  await import(pathToFileURL(join(root, "index.js")).href);
}} catch (error) {{
  console.error("Failed to load GitHub Copilot CLI:", error);
  process.exit(1);
}}
""".encode("utf-8")


def patch_npm_loader_force_js(ctx: PatchContext) -> None:
    path = ctx.root / "npm-loader.js"
    if not path.is_file():
        raise PatchError(f"Missing npm loader: {path}")

    data = path.read_bytes()
    original_sha = sha256_bytes(data)
    marker_bytes = LOADER_MARKER.encode("utf-8")

    if marker_bytes in data[:512]:
        ctx.add(
            PatchEntry(
                path=str(path),
                strategy="loader-force-js",
                status="already-patched",
                original_sha256=original_sha,
                patched_sha256=original_sha,
            )
        )
        return

    backup_path = path.with_name(path.name + ".xhigh-patch.bak")
    patched = force_js_loader_content()
    patched_sha = sha256_bytes(patched)

    if not ctx.dry_run:
        backup_once(path, backup_path)
        mode = stat.S_IMODE(path.stat().st_mode)
        atomic_write(path, patched, mode=mode)

    ctx.add(
        PatchEntry(
            path=str(path),
            strategy="loader-force-js",
            status="would-patch" if ctx.dry_run else "patched",
            backup_path=str(backup_path),
            original_sha256=original_sha,
            patched_sha256=patched_sha,
        )
    )


def write_manifest(ctx: PatchContext) -> None:
    manifest_path = ctx.root / MANIFEST_NAME
    manifest = {
        "tool": "copilot_xhigh_patch.py",
        "package_name": PACKAGE_NAME,
        "package_root": str(ctx.root),
        "package_version": ctx.version,
        "patch_profile": ctx.profile.name,
        "dry_run": ctx.dry_run,
        "entries": [entry.__dict__ for entry in ctx.entries],
    }
    if ctx.dry_run:
        return
    atomic_write(
        manifest_path,
        (json.dumps(manifest, indent=2, sort_keys=True) + "\n").encode("utf-8"),
        mode=0o644,
    )


def print_summary(ctx: PatchContext) -> None:
    action = "Dry-run plan" if ctx.dry_run else "Patch result"
    print(f"{action} for {PACKAGE_NAME} {ctx.version} at {ctx.root}")
    print(f"Patch profile: {ctx.profile.name}")
    for entry in ctx.entries:
        backup = f" backup={entry.backup_path}" if entry.backup_path else ""
        print(f"- {entry.status}: {entry.strategy}: {entry.path}{backup}")
        if entry.details:
            print(f"  details: {json.dumps(entry.details, sort_keys=True)}")
    if ctx.dry_run:
        print("No files were modified. Re-run with --apply to patch.")
    else:
        print(f"Manifest written to {ctx.root / MANIFEST_NAME}")


def verify_js_postconditions(root: Path) -> list[str]:
    errors: list[str] = []
    for relative in ("app.js", "sdk/index.js"):
        path = root / relative
        data = path.read_bytes()
        if FILTER_XHIGH_RE.search(data):
            errors.append(f"{path} still contains an xhigh-removing filter")
        if not FILTER_NEVER_RE.search(data):
            errors.append(f"{path} does not contain the patched never filter")
        for anchor in (b"supportedReasoningEfforts", b"xhigh"):
            if anchor not in data:
                errors.append(f"{path} missing semantic anchor {anchor.decode('ascii')}")
    return errors


def verify_native_postcondition(native_path: Path | None) -> list[str]:
    if native_path is None or detect_node_platform() == "win32":
        return []
    data = native_path.read_bytes()
    if not (data.startswith(b"#!") and PATCH_MARKER.encode("utf-8") in data[:512]):
        return [f"{native_path} is not a patched native wrapper"]
    return []


def restore_from_manifest(root: Path, dry_run: bool) -> None:
    manifest_path = root / MANIFEST_NAME
    manifest = load_json(manifest_path)
    entries = manifest.get("entries")
    if not isinstance(entries, list):
        raise PatchError(f"Invalid manifest entries in {manifest_path}")

    restored = 0
    for entry in reversed(entries):
        if not isinstance(entry, dict):
            continue
        path_value = entry.get("path")
        backup_value = entry.get("backup_path")
        if not path_value or not backup_value:
            continue
        path = Path(path_value)
        backup_path = Path(backup_value)
        if not backup_path.is_file():
            raise PatchError(f"Missing backup for restore: {backup_path}")
        print(f"{'Would restore' if dry_run else 'Restoring'} {path} from {backup_path}")
        if not dry_run:
            shutil.copy2(backup_path, path)
        restored += 1

    print(f"{'Would restore' if dry_run else 'Restored'} {restored} files")
    if not dry_run:
        manifest_path.unlink(missing_ok=True)


def rollback_entries(ctx: PatchContext) -> None:
    for entry in reversed(ctx.entries):
        if entry.status != "patched" or not entry.backup_path:
            continue
        path = Path(entry.path)
        backup_path = Path(entry.backup_path)
        if backup_path.is_file():
            shutil.copy2(backup_path, path)


def build_arg_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Patch @github/copilot CLI so xhigh reasoning effort is not filtered out.",
    )
    parser.add_argument("--package-root", help="Path to node_modules/@github/copilot")
    parser.add_argument("--apply", action="store_true", help="Modify files. Default is dry-run.")
    parser.add_argument("--restore", action="store_true", help="Restore files from xhigh patch manifest.")
    parser.add_argument("--force", action="store_true", help="Allow fallback patching when safety checks are not exact.")
    parser.add_argument(
        "--no-native-wrapper",
        action="store_true",
        help="Do not replace the POSIX optionalDependency native binary with a JS wrapper.",
    )
    parser.add_argument(
        "--force-js-loader",
        action="store_true",
        help="Patch npm-loader.js to skip optionalDependency native binary and load index.js directly.",
    )
    parser.add_argument(
        "--skip-postcheck",
        action="store_true",
        help="Skip post-patch verification checks.",
    )
    return parser


def run(args: argparse.Namespace) -> int:
    root = find_package_root(args.package_root)
    package_json = load_json(root / "package.json")
    version = str(package_json.get("version", "unknown"))
    dry_run = not args.apply

    if args.restore:
        restore_from_manifest(root, dry_run=dry_run)
        return 0

    profile = select_patch_profile(version)
    ctx = PatchContext(root=root, version=version, profile=profile, dry_run=dry_run, force=args.force)

    for target in profile.expected_js_targets:
        patch_js_file(ctx, target)

    native_path = find_native_binary(root)
    if native_path and not args.no_native_wrapper:
        if detect_node_platform() == "win32":
            ctx.add(
                PatchEntry(
                    path=str(native_path),
                    strategy="native-wrapper",
                    status="skipped",
                    details={"reason": "Windows .exe cannot be safely replaced with a shebang wrapper"},
                )
            )
            patch_npm_loader_force_js(ctx)
        else:
            patch_posix_native_binary(ctx, native_path)
    elif native_path is None:
        ctx.add(
            PatchEntry(
                path=str(root / "node_modules" / "@github"),
                strategy="native-wrapper",
                status="skipped",
                details={"reason": "platform optionalDependency binary not found"},
            )
        )

    if args.force_js_loader and detect_node_platform() != "win32":
        patch_npm_loader_force_js(ctx)

    if not dry_run and not args.skip_postcheck:
        errors = verify_js_postconditions(root)
        errors.extend(verify_native_postcondition(native_path if not args.no_native_wrapper else None))
        if errors:
            rollback_entries(ctx)
            raise PatchError("Post-patch verification failed:\n" + "\n".join(f"- {error}" for error in errors))

    write_manifest(ctx)
    print_summary(ctx)
    return 0


def main() -> int:
    parser = build_arg_parser()
    args = parser.parse_args()
    try:
        return run(args)
    except PatchError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())