.pouch/skills/orc/scripts/resolve_profile.py

#!/usr/bin/env python3
"""Validate shared ORC config and build a worktree-bound Agent launch plan."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import pwd
import re
import shlex
import stat
import subprocess
import sys
from pathlib import Path
from typing import Any

LEVELS = ("low", "mid", "high")
STAGES = ("code", "release", "deb", "docker")
CLIS = ("codex", "cursor-agent")
EXPECTED_EFFORT = {"low": "low", "mid": "medium", "high": "high"}
TOP_LEVEL_KEYS = {
    "version",
    "cliPolicy",
    "defaultLevel",
    "stageDefaults",
    "worktreePolicy",
    "profiles",
}
PROFILE_KEYS = {
    "model",
    "reasoningEffort",
    "permissionMode",
    "approvalPolicy",
}
MODEL_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}\Z")
MAX_CONFIG_SIZE = 64 * 1024
MAX_CONTROL_OUTPUT = 64 * 1024
CONTROL_TIMEOUT_SECONDS = 15
SHARED_CONFIG_PATH = Path(__file__).resolve().parents[1] / "config.yaml"
COMMON_ENVIRONMENT_NAMES = {
    "COLORTERM",
    "LANG",
    "LC_ALL",
    "NO_COLOR",
    "TERM",
    "TZ",
}
MODEL_AUTH_ENVIRONMENT = {
    "codex-login": frozenset(),
    "openai": frozenset({"OPENAI_API_KEY"}),
    "azure-openai": frozenset({"AZURE_OPENAI_API_KEY"}),
    "cursor-login": frozenset(),
    "cursor-api-key": frozenset({"CURSOR_API_KEY"}),
}
CLI_MODEL_AUTH = {
    "codex": frozenset({"codex-login", "openai", "azure-openai"}),
    "cursor-agent": frozenset({"cursor-login", "cursor-api-key"}),
}
DEFAULT_MODEL_AUTH = {
    "codex": "codex-login",
    "cursor-agent": "cursor-login",
}
REMOTE_AUTH_ENVIRONMENT = {
    "none": frozenset(),
    "github-token": frozenset({"GITHUB_TOKEN"}),
    "gitlab-token": frozenset({"GITLAB_TOKEN"}),
    "gitea-token": frozenset({"GITEA_TOKEN"}),
    "forgejo-token": frozenset({"FORGEJO_TOKEN"}),
    "ssh-agent": frozenset({"SSH_AUTH_SOCK"}),
    "deb-token": frozenset({"DEB_TOKEN"}),
}
STAGE_REMOTE_AUTH = {
    "code": frozenset(
        {
            "none",
            "github-token",
            "gitlab-token",
            "gitea-token",
            "forgejo-token",
            "ssh-agent",
        }
    ),
    "release": frozenset(
        {
            "none",
            "github-token",
            "gitlab-token",
            "gitea-token",
            "forgejo-token",
            "ssh-agent",
        }
    ),
    "deb": frozenset({"none", "deb-token", "ssh-agent"}),
    "docker": frozenset({"none"}),
}


class ConfigError(ValueError):
    """ORC configuration or launch state is invalid or unsafe."""


def _json_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ConfigError("JSON-compatible YAML contains a duplicate key")
        result[key] = value
    return result


def _read_bounded_regular_file(path: Path) -> tuple[str, dict[str, int]]:
    absolute = path.absolute()
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor: int | None = None
    try:
        descriptor = os.open(absolute, flags)
        with os.fdopen(descriptor, "rb") as stream:
            descriptor = None
            before = os.fstat(stream.fileno())
            if not stat.S_ISREG(before.st_mode):
                raise ConfigError(f"config must be a regular file: {absolute}")
            if before.st_size > MAX_CONFIG_SIZE:
                raise ConfigError(f"config exceeds {MAX_CONFIG_SIZE} bytes")
            content = stream.read(MAX_CONFIG_SIZE + 1)
            after = os.fstat(stream.fileno())
    except OSError as exc:
        raise ConfigError(f"cannot safely read config {absolute}: {exc}") from exc
    finally:
        if descriptor is not None:
            os.close(descriptor)

    if len(content) > MAX_CONFIG_SIZE:
        raise ConfigError(f"config exceeds {MAX_CONFIG_SIZE} bytes")
    identity_before = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
    identity_after = (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
    if identity_before != identity_after:
        raise ConfigError("config changed while it was being read")
    try:
        text = content.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise ConfigError("config must be valid UTF-8") from exc
    return text, {
        "device": before.st_dev,
        "inode": before.st_ino,
        "size": before.st_size,
        "mtimeNs": before.st_mtime_ns,
    }


def _mapping(value: Any, path: str) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ConfigError(f"{path} must be a mapping")
    return value


def _exact_keys(
    value: dict[str, Any],
    *,
    required: set[str],
    path: str,
) -> None:
    missing = sorted(required - set(value))
    unknown = sorted(set(value) - required)
    if missing:
        raise ConfigError(f"{path} is missing fields: {', '.join(missing)}")
    if unknown:
        raise ConfigError(f"{path} has unknown fields: {', '.join(unknown)}")


def _enum(value: Any, allowed: tuple[str, ...] | set[str], path: str) -> str:
    if not isinstance(value, str) or value not in allowed:
        raise ConfigError(f"{path} must be one of: {', '.join(sorted(allowed))}")
    return value


def validate_config(document: Any) -> dict[str, Any]:
    config = _mapping(document, "config")
    _exact_keys(config, required=TOP_LEVEL_KEYS, path="config")

    version = config["version"]
    if isinstance(version, bool) or version != 2:
        raise ConfigError("version must be integer 2")

    if config["cliPolicy"] != "current-host":
        raise ConfigError("cliPolicy must be current-host in ORC v2")
    _enum(config["defaultLevel"], LEVELS, "defaultLevel")

    stage_defaults = _mapping(config["stageDefaults"], "stageDefaults")
    unknown_stages = sorted(set(stage_defaults) - set(STAGES))
    if unknown_stages:
        raise ConfigError(
            f"stageDefaults has unknown fields: {', '.join(unknown_stages)}"
        )
    for stage, level in stage_defaults.items():
        _enum(level, LEVELS, f"stageDefaults.{stage}")

    if config["worktreePolicy"] != "registered-same-repository":
        raise ConfigError(
            "worktreePolicy must be registered-same-repository in ORC v2"
        )

    profiles = _mapping(config["profiles"], "profiles")
    _exact_keys(profiles, required=set(CLIS), path="profiles")
    for cli in CLIS:
        cli_profiles = _mapping(profiles[cli], f"profiles.{cli}")
        _exact_keys(cli_profiles, required=set(LEVELS), path=f"profiles.{cli}")
        for level in LEVELS:
            path = f"profiles.{cli}.{level}"
            profile = _mapping(cli_profiles[level], path)
            _exact_keys(profile, required=PROFILE_KEYS, path=path)
            model = profile["model"]
            if not isinstance(model, str) or not MODEL_RE.fullmatch(model):
                raise ConfigError(f"{path}.model is not a safe exact model ID")
            if cli == "codex":
                expected_effort = EXPECTED_EFFORT[level]
                if profile["reasoningEffort"] != expected_effort:
                    raise ConfigError(
                        f"{path}.reasoningEffort must be {expected_effort}"
                    )
                _enum(
                    profile["approvalPolicy"],
                    {"untrusted", "on-request", "never"},
                    f"{path}.approvalPolicy",
                )
            else:
                if profile["reasoningEffort"] is not None:
                    raise ConfigError(f"{path}.reasoningEffort: Cursor requires null")
                model_tokens = set(re.split(r"[^a-z0-9]+", model.lower()))
                expected_token = "medium" if level == "mid" else level
                if model != "auto" and expected_token not in model_tokens:
                    raise ConfigError(
                        f"{path}.model must encode the {expected_token} effort level"
                    )
                if profile["approvalPolicy"] != "auto-review":
                    raise ConfigError(
                        f"{path}.approvalPolicy must be auto-review for Cursor"
                    )
            if profile["permissionMode"] != "workspace-write":
                raise ConfigError(
                    f"{path}.permissionMode must be workspace-write in ORC v2"
                )
    return config


def load_config_snapshot(path: Path) -> tuple[dict[str, Any], dict[str, Any]]:
    raw, identity = _read_bounded_regular_file(path)
    try:
        document = json.loads(raw, object_pairs_hook=_json_object)
    except ConfigError:
        raise
    except Exception as exc:
        raise ConfigError("invalid JSON-compatible YAML config") from exc
    config = validate_config(document)
    snapshot = {
        "sha256": hashlib.sha256(raw.encode("utf-8")).hexdigest(),
        **identity,
    }
    return config, snapshot


def load_config(path: Path) -> dict[str, Any]:
    config, _ = load_config_snapshot(path)
    return config


def resolve_profile(
    config: dict[str, Any],
    *,
    stage: str,
    global_level: str | None = None,
    stage_level: str | None = None,
    host_cli: str | None = None,
    model_auth: str | None = None,
    remote_auth: str = "none",
) -> dict[str, Any]:
    _enum(stage, STAGES, "stage")
    if global_level is not None:
        _enum(global_level, LEVELS, "global level")
    if stage_level is not None:
        _enum(stage_level, LEVELS, "stage level")
    _enum(remote_auth, STAGE_REMOTE_AUTH[stage], f"{stage} remote auth")

    if host_cli is None:
        raise ConfigError("host CLI is required; resolve it from the current Agent")
    selected_cli = _enum(host_cli, CLIS, "host CLI")
    cli_source = "runtime.host"
    selected_auth = model_auth or DEFAULT_MODEL_AUTH[selected_cli]
    _enum(selected_auth, set(MODEL_AUTH_ENVIRONMENT), "model auth")
    if selected_auth not in CLI_MODEL_AUTH[selected_cli]:
        raise ConfigError(
            f"model auth {selected_auth} is not valid for {selected_cli}"
        )

    if stage_level is not None:
        level, source = stage_level, "request.stage"
    elif global_level is not None:
        level, source = global_level, "request.global"
    elif stage in config["stageDefaults"]:
        level, source = config["stageDefaults"][stage], f"config.stageDefaults.{stage}"
    else:
        level, source = config["defaultLevel"], "config.defaultLevel"

    profile = config["profiles"][selected_cli].get(level)
    if profile is None:
        raise ConfigError(f"requested profile does not exist: {selected_cli}/{level}")
    if selected_cli == "codex":
        worker_args = [
            "--model",
            profile["model"],
            "-c",
            f'model_reasoning_effort="{profile["reasoningEffort"]}"',
            "--sandbox",
            profile["permissionMode"],
            "--ask-for-approval",
            profile["approvalPolicy"],
            "--strict-config",
        ]
    else:
        worker_args = [
            "--model",
            profile["model"],
            "--auto-review",
            "--sandbox",
            "enabled",
        ]
    return {
        "stage": stage,
        "cli": selected_cli,
        "cliSelectionSource": cli_source,
        "level": level,
        "selectionSource": source,
        "modelAuth": selected_auth,
        "remoteAuth": remote_auth,
        "profile": dict(profile),
        "workerArgs": worker_args,
    }


def _path_has_parent_reference(value: str) -> bool:
    return ".." in Path(value).parts


def _assert_no_symlink_components(path: Path, label: str) -> None:
    if not path.is_absolute():
        raise ConfigError(f"{label} must be absolute: {path}")
    current = Path(path.anchor)
    for part in path.parts[1:]:
        current /= part
        try:
            metadata = os.lstat(current)
        except OSError as exc:
            raise ConfigError(f"{label} does not exist: {current}") from exc
        if stat.S_ISLNK(metadata.st_mode):
            raise ConfigError(f"{label} must not contain symlinks: {current}")


def canonical_directory(value: str | Path, label: str) -> Path:
    text = str(value)
    if (
        not text
        or text != text.strip()
        or any(character in text for character in ("\x00", "\n", "\r"))
        or _path_has_parent_reference(text)
    ):
        raise ConfigError(f"{label} must be a safe canonical absolute path")
    raw = Path(text)
    _assert_no_symlink_components(raw, label)
    try:
        resolved = raw.resolve(strict=True)
    except OSError as exc:
        raise ConfigError(f"{label} does not exist: {raw}") from exc
    if resolved != raw or resolved == Path(resolved.anchor) or not resolved.is_dir():
        raise ConfigError(f"{label} must be a canonical non-root directory: {resolved}")
    return resolved


def account_identity() -> tuple[Path, str]:
    account = pwd.getpwuid(os.getuid())
    account_home = Path(account.pw_dir).resolve(strict=True)
    if not account_home.is_dir():
        raise ConfigError("current account home is unavailable")
    return account_home, account.pw_name


def trusted_path_entries() -> list[Path]:
    account_home, _ = account_identity()
    candidates = [
        account_home / ".local" / "bin",
        account_home / ".local" / "share" / "mise" / "shims",
        account_home / ".cargo" / "bin",
        Path("/home/linuxbrew/.linuxbrew/bin"),
        Path("/usr/local/bin"),
        Path("/usr/bin"),
        Path("/bin"),
    ]
    result: list[Path] = []
    for candidate in candidates:
        try:
            resolved = candidate.resolve(strict=True)
        except OSError:
            continue
        if resolved.is_dir() and resolved not in result:
            result.append(resolved)
    return result


def _trusted_executable(path: Path, expected_name: str) -> Path | None:
    try:
        candidate_metadata = os.lstat(path)
        resolved = path.resolve(strict=True)
        metadata = resolved.stat()
    except OSError:
        return None
    if not (
        stat.S_ISREG(candidate_metadata.st_mode)
        or stat.S_ISLNK(candidate_metadata.st_mode)
    ):
        return None
    if not stat.S_ISREG(metadata.st_mode) or not os.access(resolved, os.X_OK):
        return None
    if metadata.st_uid not in {0, os.getuid()}:
        return None
    if stat.S_IMODE(metadata.st_mode) & 0o022:
        return None
    if expected_name == "python3":
        if not resolved.name.startswith("python3"):
            return None
    elif resolved.name != expected_name:
        return None
    return resolved


def resolve_trusted_executable(name: str) -> Path:
    if name not in {"codex", "cursor-agent", "git", "orca"}:
        raise ConfigError(f"unsupported executable: {name}")
    search_paths = trusted_path_entries()
    if name == "git":
        search_paths = [
            path
            for path in search_paths
            if str(path) in {"/usr/local/bin", "/usr/bin", "/bin"}
        ]
    for directory in search_paths:
        resolved = _trusted_executable(directory / name, name)
        if resolved is not None:
            return resolved
    raise ConfigError(f"trusted {name} executable was not found in fixed directories")


def resolve_trusted_python() -> Path:
    for candidate in (Path("/usr/bin/python3"), Path("/usr/local/bin/python3")):
        resolved = _trusted_executable(candidate, "python3")
        if resolved is not None and resolved.stat().st_uid == 0:
            return resolved
    raise ConfigError("a root-owned Python executable was not found in fixed paths")


def control_environment() -> dict[str, str]:
    account_home, username = account_identity()
    result = {
        "HOME": str(account_home),
        "LOGNAME": username,
        "PATH": os.pathsep.join(str(path) for path in trusted_path_entries()),
        "USER": username,
    }
    for name, value in os.environ.items():
        if (
            name in COMMON_ENVIRONMENT_NAMES or name.startswith("LC_")
        ) and "\x00" not in value:
            result[name] = value
    return result


def selected_credential_names(
    stage: str,
    *,
    model_auth: str,
    remote_auth: str,
) -> frozenset[str]:
    _enum(stage, STAGES, "stage")
    _enum(model_auth, set(MODEL_AUTH_ENVIRONMENT), "model auth")
    _enum(remote_auth, STAGE_REMOTE_AUTH[stage], f"{stage} remote auth")
    names = MODEL_AUTH_ENVIRONMENT[model_auth] | REMOTE_AUTH_ENVIRONMENT[remote_auth]
    for name in names:
        value = os.environ.get(name)
        if not value or "\x00" in value:
            raise ConfigError(
                f"selected authentication variable is unavailable: {name}"
            )
        if name == "SSH_AUTH_SOCK":
            socket_path = Path(value)
            if (
                not socket_path.is_absolute()
                or value != value.strip()
                or any(character in value for character in ("\n", "\r"))
                or _path_has_parent_reference(value)
            ):
                raise ConfigError("selected SSH_AUTH_SOCK is not a safe absolute path")
            _assert_no_symlink_components(socket_path, "selected SSH_AUTH_SOCK")
            try:
                metadata = os.lstat(socket_path)
            except OSError as exc:
                raise ConfigError("selected SSH_AUTH_SOCK is unavailable") from exc
            if (
                not stat.S_ISSOCK(metadata.st_mode)
                or metadata.st_uid != os.getuid()
                or stat.S_IMODE(metadata.st_mode) & 0o022
            ):
                raise ConfigError("selected SSH_AUTH_SOCK is not a trusted user socket")
    return names


def worker_environment(
    stage: str,
    *,
    model_auth: str,
    remote_auth: str,
) -> dict[str, str]:
    result = control_environment()
    for name in selected_credential_names(
        stage,
        model_auth=model_auth,
        remote_auth=remote_auth,
    ):
        result[name] = os.environ[name]
    return result


def _run_control(argv: list[str], label: str) -> str:
    try:
        completed = subprocess.run(
            argv,
            shell=False,
            check=False,
            capture_output=True,
            text=True,
            timeout=CONTROL_TIMEOUT_SECONDS,
            env=control_environment(),
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise ConfigError(f"{label} could not be executed safely") from exc
    if completed.returncode != 0:
        raise ConfigError(f"{label} failed with exit {completed.returncode}")
    output = completed.stdout.strip()
    if not output or len(output.encode("utf-8")) > MAX_CONTROL_OUTPUT:
        raise ConfigError(f"{label} returned invalid output")
    return output


def _git_path(value: str, cwd: Path, label: str) -> Path:
    candidate = Path(value)
    if not candidate.is_absolute():
        candidate = cwd / candidate
    _assert_no_symlink_components(candidate, label)
    try:
        resolved = candidate.resolve(strict=True)
    except OSError as exc:
        raise ConfigError(f"{label} is invalid") from exc
    if not resolved.is_dir():
        raise ConfigError(f"{label} is not a directory")
    return resolved


def validate_worktree(
    config: dict[str, Any],
    *,
    project_root_value: str | Path,
    worktree_value: str | Path,
) -> dict[str, Any]:
    project_root = canonical_directory(project_root_value, "project root")
    worktree = canonical_directory(worktree_value, "target worktree")
    git = resolve_trusted_executable("git")

    top_level = canonical_directory(
        _run_control(
            [str(git), "-C", str(project_root), "rev-parse", "--show-toplevel"],
            "Git project-root check",
        ),
        "Git project root",
    )
    if top_level != project_root:
        raise ConfigError("project root is not the repository top level")

    common_dir = _git_path(
        _run_control(
            [str(git), "-C", str(project_root), "rev-parse", "--git-common-dir"],
            "Git common-directory check",
        ),
        project_root,
        "Git common directory",
    )
    listed = _run_control(
        [str(git), "-C", str(project_root), "worktree", "list", "--porcelain"],
        "Git worktree listing",
    )
    registered: set[Path] = set()
    for line in listed.splitlines():
        if line.startswith("worktree "):
            try:
                registered.add(
                    canonical_directory(line[9:], "registered worktree")
                )
            except ConfigError:
                continue

    if config["worktreePolicy"] != "registered-same-repository":
        raise ConfigError("unsupported worktree policy")
    if worktree not in registered:
        raise ConfigError("target worktree is not registered in the project repository")

    target_top = canonical_directory(
        _run_control(
            [str(git), "-C", str(worktree), "rev-parse", "--show-toplevel"],
            "Git target-worktree check",
        ),
        "target Git worktree",
    )
    target_common = _git_path(
        _run_control(
            [str(git), "-C", str(worktree), "rev-parse", "--git-common-dir"],
            "target Git common-directory check",
        ),
        worktree,
        "target Git common directory",
    )
    if target_top != worktree or target_common != common_dir:
        raise ConfigError("target worktree does not belong to the project repository")

    worktree_metadata = worktree.stat()
    common_metadata = common_dir.stat()
    return {
        "projectRoot": str(project_root),
        "worktree": str(worktree),
        "gitCommonDir": str(common_dir),
        "worktreeIdentity": {
            "device": worktree_metadata.st_dev,
            "inode": worktree_metadata.st_ino,
            "gitCommonDevice": common_metadata.st_dev,
            "gitCommonInode": common_metadata.st_ino,
        },
    }


def _file_facts(path: Path) -> dict[str, Any]:
    metadata = path.stat()
    return {
        "path": str(path),
        "device": metadata.st_dev,
        "inode": metadata.st_ino,
        "size": metadata.st_size,
        "mtimeNs": metadata.st_mtime_ns,
    }


def _executable_facts(path: Path) -> dict[str, Any]:
    return {
        **_file_facts(path),
        "version": _run_control([str(path), "--version"], "Agent CLI version check"),
    }


def _python_facts(path: Path) -> dict[str, Any]:
    version = _run_control(
        [
            str(path),
            "-I",
            "-S",
            "-c",
            "import sys; print(sys.version.split()[0])",
        ],
        "Python version check",
    )
    return {
        **_file_facts(path),
        "version": version,
    }


def _fingerprint(value: dict[str, Any]) -> str:
    canonical = json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return "sha256:" + hashlib.sha256(canonical).hexdigest()


def build_launch_plan(
    *,
    project_root: str | Path,
    worktree: str | Path,
    stage: str,
    host_cli: str,
    global_level: str | None = None,
    stage_level: str | None = None,
    model_auth: str | None = None,
    remote_auth: str = "none",
) -> dict[str, Any]:
    project = canonical_directory(project_root, "project root")
    _assert_no_symlink_components(SHARED_CONFIG_PATH, "shared config path")
    config, config_snapshot = load_config_snapshot(SHARED_CONFIG_PATH)
    worktree_facts = validate_worktree(
        config,
        project_root_value=project,
        worktree_value=worktree,
    )
    profile = resolve_profile(
        config,
        stage=stage,
        global_level=global_level,
        stage_level=stage_level,
        host_cli=host_cli,
        model_auth=model_auth,
        remote_auth=remote_auth,
    )
    selected_credential_names(
        stage,
        model_auth=profile["modelAuth"],
        remote_auth=remote_auth,
    )
    agent_cli = resolve_trusted_executable(profile["cli"])
    executable = _executable_facts(agent_cli)
    orca = resolve_trusted_executable("orca")
    orca_executable = _file_facts(orca)
    python = resolve_trusted_python()
    python_executable = _python_facts(python)
    script = Path(__file__).resolve(strict=True)
    script_metadata = script.stat()
    script_sha256 = hashlib.sha256(script.read_bytes()).hexdigest()
    launch_facts = {
        **profile,
        **worktree_facts,
        "config": {
            "path": str(SHARED_CONFIG_PATH),
            **config_snapshot,
        },
        "executable": executable,
        "orca": orca_executable,
        "python": python_executable,
        "launcher": {
            "path": str(script),
            "device": script_metadata.st_dev,
            "inode": script_metadata.st_ino,
            "size": script_metadata.st_size,
            "mtimeNs": script_metadata.st_mtime_ns,
            "sha256": script_sha256,
        },
    }
    fingerprint = _fingerprint(launch_facts)
    worker_argv = [str(agent_cli), *profile["workerArgs"]]
    if profile["cli"] == "cursor-agent":
        worker_argv.extend(["--workspace", str(worktree_facts["worktree"])])
    launcher_argv = [
        str(python),
        "-I",
        "-S",
        str(script),
        "_launch",
        "--project-root",
        str(project),
        "--worktree",
        str(worktree_facts["worktree"]),
        "--stage",
        stage,
        "--expected-fingerprint",
        fingerprint,
    ]
    if global_level is not None:
        launcher_argv.extend(["--global-level", global_level])
    if stage_level is not None:
        launcher_argv.extend(["--stage-level", stage_level])
    launcher_argv.extend(["--host-cli", host_cli])
    if model_auth is not None:
        launcher_argv.extend(["--model-auth", model_auth])
    if remote_auth != "none":
        launcher_argv.extend(["--remote-auth", remote_auth])
    terminal_command = shlex.join(launcher_argv)
    worktree_selector = "path:" + str(worktree_facts["worktree"])
    cli_label = "CODEX" if profile["cli"] == "codex" else "CURSOR"
    terminal_title = (
        f"ORC-{stage}-{cli_label}-{profile['level']}-{fingerprint[7:15]}"
    )
    terminal_create_argv = [
        str(orca),
        "terminal",
        "create",
        "--worktree",
        worktree_selector,
        "--title",
        terminal_title,
        "--command",
        terminal_command,
        "--json",
    ]
    return {
        **launch_facts,
        "argv": worker_argv,
        "launchFingerprint": fingerprint,
        "launcherArgv": launcher_argv,
        "terminalCommand": terminal_command,
        "worktreeSelector": worktree_selector,
        "terminalTitle": terminal_title,
        "terminalCreateArgv": terminal_create_argv,
        "terminalCreateShellCommand": shlex.join(terminal_create_argv),
    }


def execute_launch(args: argparse.Namespace) -> int:
    plan = build_launch_plan(
        project_root=args.project_root,
        worktree=args.worktree,
        stage=args.stage,
        host_cli=args.host_cli,
        global_level=args.global_level,
        stage_level=args.stage_level,
        model_auth=args.model_auth,
        remote_auth=args.remote_auth,
    )
    if plan["launchFingerprint"] != args.expected_fingerprint:
        raise ConfigError("launch fingerprint changed; resolve the profile again")
    current = canonical_directory(Path.cwd(), "launcher working directory")
    if current != Path(plan["worktree"]):
        raise ConfigError("launcher working directory does not match target worktree")
    current_python = Path(sys.executable).resolve(strict=True)
    if current_python != Path(plan["python"]["path"]):
        raise ConfigError("launcher Python does not match the resolved interpreter")
    if _python_facts(current_python) != plan["python"]:
        raise ConfigError("Python executable changed before launch")
    executable = Path(plan["executable"]["path"])
    if _executable_facts(executable) != plan["executable"]:
        raise ConfigError("Agent CLI executable changed before launch")
    try:
        os.execve(
            executable,
            plan["argv"],
            worker_environment(
                plan["stage"],
                model_auth=plan["modelAuth"],
                remote_auth=plan["remoteAuth"],
            ),
        )
    except OSError as exc:
        raise ConfigError("Agent worker could not be launched") from exc
    return 1  # pragma: no cover - os.execve does not return on success


def _add_resolution_arguments(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--project-root", required=True)
    parser.add_argument("--worktree", required=True)
    parser.add_argument("--stage", required=True, choices=STAGES)
    parser.add_argument("--global-level", choices=LEVELS)
    parser.add_argument("--stage-level", choices=LEVELS)
    parser.add_argument("--host-cli", required=True, choices=CLIS)
    parser.add_argument(
        "--model-auth",
        choices=tuple(MODEL_AUTH_ENVIRONMENT),
    )
    parser.add_argument(
        "--remote-auth",
        choices=tuple(REMOTE_AUTH_ENVIRONMENT),
        default="none",
    )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)

    validate = subparsers.add_parser("validate", help="validate config only")
    resolve = subparsers.add_parser("resolve", help="resolve a bound launch plan")
    _add_resolution_arguments(resolve)

    launch = subparsers.add_parser("_launch", help=argparse.SUPPRESS)
    _add_resolution_arguments(launch)
    launch.add_argument("--expected-fingerprint", required=True)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        if args.command == "validate":
            config = load_config(SHARED_CONFIG_PATH)
            result: dict[str, Any] = {
                "ok": True,
                "config": str(SHARED_CONFIG_PATH),
                "clis": list(CLIS),
                "levels": list(LEVELS),
                "stages": list(STAGES),
                "cliPolicy": config["cliPolicy"],
                "defaultLevel": config["defaultLevel"],
            }
        elif args.command == "resolve":
            result = build_launch_plan(
                project_root=args.project_root,
                worktree=args.worktree,
                stage=args.stage,
                host_cli=args.host_cli,
                global_level=args.global_level,
                stage_level=args.stage_level,
                model_auth=args.model_auth,
                remote_auth=args.remote_auth,
            )
        else:
            return execute_launch(args)
    except ConfigError as exc:
        print(f"ORC config error: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, ensure_ascii=False, sort_keys=True))
    return 0


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