#!/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 ipaddress import json import os import pwd import re import shlex import stat import subprocess import sys from pathlib import Path from typing import Any from urllib.parse import urlsplit 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") REMOTE_HOST_RE = re.compile( r"(?=.{1,253}\Z)(?:[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?\.)*" r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?\Z" ) SCP_REMOTE_RE = re.compile( r"(?:(?P[A-Za-z0-9._-]+)@)?" r"(?P[A-Za-z0-9.-]+):(?P[^\s]+)\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" configured_profile = config["profiles"][selected_cli].get(level) if configured_profile is None: raise ConfigError(f"requested profile does not exist: {selected_cli}/{level}") profile = dict(configured_profile) if selected_cli == "codex": if stage == "release": profile["approvalPolicy"] = "on-request" worker_args = [ "--model", profile["model"], "-c", f'model_reasoning_effort="{profile["reasoningEffort"]}"', "--sandbox", profile["permissionMode"], "--ask-for-approval", profile["approvalPolicy"], "--strict-config", ] if stage == "release": worker_args.extend(["-c", 'approvals_reviewer="auto_review"']) 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": 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 _validated_remote_host(value: str, label: str) -> str: host = value.lower() if not REMOTE_HOST_RE.fullmatch(host): raise ConfigError(f"{label} has an unsupported host") if host == "localhost" or host.endswith(".localhost"): raise ConfigError(f"{label} must not target a local host") try: address = ipaddress.ip_address(host) except ValueError: pass else: if not address.is_global: raise ConfigError(f"{label} must not target a private or local address") return host def _remote_url_facts(value: str, label: str) -> dict[str, Any]: if ( not value or value != value.strip() or any(character.isspace() or character == "\x00" for character in value) ): raise ConfigError(f"{label} is not a safe remote URL") if "://" not in value: match = SCP_REMOTE_RE.fullmatch(value) if match is None: raise ConfigError(f"{label} is not a supported HTTPS or SSH URL") scheme = "ssh" username = match.group("username") or "git" host = _validated_remote_host(match.group("host"), label) port = 22 path = "/" + match.group("path").lstrip("/") else: parsed = urlsplit(value) if ( parsed.scheme not in {"https", "ssh"} or not parsed.hostname or parsed.password is not None or parsed.query or parsed.fragment or not parsed.path or parsed.path == "/" ): raise ConfigError(f"{label} is not a supported HTTPS or SSH URL") scheme = parsed.scheme username = parsed.username if scheme == "https" and username is not None: raise ConfigError(f"{label} must not embed HTTPS credentials") host = _validated_remote_host(parsed.hostname, label) try: port = parsed.port or (443 if scheme == "https" else 22) except ValueError as exc: raise ConfigError(f"{label} is not a safe remote URL") from exc path = parsed.path if not path.startswith("/") or _path_has_parent_reference(path): raise ConfigError(f"{label} has an unsafe repository path") user_prefix = f"{username}@" if username is not None else "" canonical = f"{scheme}://{user_prefix}{host}:{port}{path}" return { "url": value, "canonical": canonical, "scheme": scheme, "host": host, "port": port, "path": path, } def _origin_urls(git: Path, worktree: Path, *, push: bool) -> list[str]: argv = [str(git), "-C", str(worktree), "remote", "get-url"] if push: argv.append("--push") argv.extend(["--all", "origin"]) output = _run_control( argv, f"Git origin {'push' if push else 'fetch'} URL check", ) return output.splitlines() def resolve_release_remote(worktree: Path) -> dict[str, Any]: git = resolve_trusted_executable("git") fetch_urls = _origin_urls(git, worktree, push=False) push_urls = _origin_urls(git, worktree, push=True) if len(fetch_urls) != 1: raise ConfigError("release origin must have exactly one fetch URL") if len(push_urls) != 1: raise ConfigError("release origin must have exactly one push URL") fetch = _remote_url_facts(fetch_urls[0], "Git origin fetch URL") push = _remote_url_facts(push_urls[0], "Git origin push URL") if fetch["canonical"] != push["canonical"]: raise ConfigError("release origin fetch and push URLs must match") network_hosts = [push["host"]] if push["host"] == "github.com": network_hosts.extend(["api.github.com", "uploads.github.com"]) return { "name": "origin", "fetchUrl": fetch["url"], "pushUrl": push["url"], "canonical": push["canonical"], "scheme": push["scheme"], "host": push["host"], "port": push["port"], "path": push["path"], "networkHosts": network_hosts, } 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, ) release_remote = None if stage == "release": release_remote = resolve_release_remote(Path(worktree_facts["worktree"])) if profile["cli"] == "codex": network_domains = ", ".join( f'"{host}" = "allow"' for host in release_remote["networkHosts"] ) profile = { **profile, "workerArgs": [ *profile["workerArgs"], "-c", "sandbox_workspace_write.network_access=true", "-c", "features.network_proxy.enabled=true", "-c", "features.network_proxy.allow_local_binding=false", "-c", "features.network_proxy.allow_upstream_proxy=false", "-c", "features.network_proxy.dangerously_allow_all_unix_sockets=false", "-c", "features.network_proxy.dangerously_allow_non_loopback_proxy=false", "-c", "features.network_proxy.unix_sockets={}", "-c", f"features.network_proxy.domains={{ {network_domains} }}", ], } 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, }, } if release_remote is not None: launch_facts["releaseRemote"] = release_remote 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())