#!/usr/bin/env python3 """Command-line control plane for the Codex MMO profile runtime.""" from __future__ import annotations import argparse import difflib import os import re import subprocess import sys import textwrap import traceback from pathlib import Path from typing import Any, Never from mmo_catalog import ( catalog_data, catalog_summary, discover_codex, discover_opencode_go, discover_opencode_zen, discover_openrouter, discover_zai, find_model, list_models, local_inventory_report, refresh_discovery, verify_catalog, ) from mmo_cli_output import ( emit_error, emit_json, emit_scalar, emit_structured, emit_tty_success, emit_usage_error, progress, stdout_is_tty, ) from mmo_diagnostics import ( doctor, profile_validation_report, smoke_profile, tool_mcp_status, ) from mmo_eval import ( compare_runs, discover_suites, list_runs, load_run, run_evaluation, validate_suite, ) from mmo_gateway import ( ensure_gateway, gateway_models, gateway_status, list_gateways, stop_gateway, stop_idle_gateways, ) from mmo_profiles import ( active_profile_id, clone_profile, discover_profiles, install_profile_pack, load_settings, profile_summary, remove_profile, resolve_profile, set_active_profile, ) from mmo_runtime import ( cancel_job, cancel_session, clean_state, compact_session, continue_session, detach_session, iter_session_runs, iter_sessions, launch_interactive, list_jobs, load_session, pause_session, public_run, public_session, read_result, resolve_resume_session, resume_interactive, run_root_exec, stop_session, wait_for_jobs, ) from mmo_snapshot import compile_profile from mmo_tool_mcp import ( load_tool_mcp_registry_with_sources, ) from mmo_util import ( filtered_environment, package_version, shell_exit_status, ) class CLIUsageError(Exception): """A command-line contract error that should exit with status 2.""" def __init__(self, message: str, *, hint: str | None = None) -> None: super().__init__(message) self.hint = hint class MMOArgumentParser(argparse.ArgumentParser): """Argparse with stable no-color output and catchable usage failures.""" def __init__(self, *args: Any, **kwargs: Any) -> None: kwargs.setdefault("allow_abbrev", False) if sys.version_info >= (3, 14): kwargs.setdefault("color", False) super().__init__(*args, **kwargs) def error(self, message: str) -> Never: raise CLIUsageError(message) def _bindings(values: list[str] | None) -> dict[str, str]: result: dict[str, str] = {} for raw in values or []: if "=" not in raw: raise ValueError(f"invalid binding {raw!r}; use AGENT=MODEL") agent, model = raw.split("=", 1) if not agent or not model: raise ValueError(f"invalid binding {raw!r}; use AGENT=MODEL") if agent in result: raise ValueError(f"duplicate binding for agent {agent!r}") result[agent] = model return result _CODEX_OPTIONS_WITH_VALUES = frozenset( { "-a", "--add-dir", "--ask-for-approval", "-C", "--cd", "-c", "--config", "--disable", "--enable", "--local-provider", "-m", "--model", "-p", "--profile", "--remote", "--remote-auth-token-env", "-s", "--sandbox", } ) _CODEX_LONG_OPTIONS_WITH_VALUES = tuple( option for option in _CODEX_OPTIONS_WITH_VALUES if option.startswith("--") ) _CODEX_SHORT_OPTIONS_WITH_VALUES = tuple( option for option in _CODEX_OPTIONS_WITH_VALUES if option.startswith("-") and len(option) == 2 ) def _codex_subcommand(arguments: list[str]) -> str | None: """Return the first Codex positional after its documented global options.""" index = 0 while index < len(arguments): token = arguments[index] if token == "--": return None if ( token in {"-i", "--image"} or token.startswith("--image=") or (token.startswith("-i") and token != "-i") ): index += 1 if token in {"-i", "--image"} and index < len(arguments): index += 1 while index < len(arguments) and not arguments[index].startswith("-"): index += 1 continue if token in _CODEX_OPTIONS_WITH_VALUES: index += 2 continue if any(token.startswith(option + "=") for option in _CODEX_LONG_OPTIONS_WITH_VALUES): index += 1 continue if any( token.startswith(option) and token != option for option in _CODEX_SHORT_OPTIONS_WITH_VALUES ): index += 1 continue if token.startswith("-"): index += 1 continue return token return None def _tail(path: Path, lines: int) -> str: if not path.is_file(): return "" values = path.read_text(encoding="utf-8", errors="replace").splitlines() return "\n".join(values[-lines:]) def _bounded_int(minimum: int, maximum: int) -> Any: def parse(value: str) -> int: try: parsed = int(value) except ValueError as exc: raise argparse.ArgumentTypeError("must be an integer") from exc if not minimum <= parsed <= maximum: raise argparse.ArgumentTypeError(f"must be between {minimum} and {maximum}") return parsed return parse def _bounded_float(minimum: float, maximum: float) -> Any: def parse(value: str) -> float: try: parsed = float(value) except ValueError as exc: raise argparse.ArgumentTypeError("must be a number") from exc if not minimum <= parsed <= maximum: raise argparse.ArgumentTypeError(f"must be between {minimum:g} and {maximum:g}") return parsed return parse def _configured_codex_context(binary: str | None, home: Path | None) -> tuple[str, Path | None]: """Apply explicit option, environment, then settings precedence.""" settings = load_settings() selected_binary = ( binary or os.environ.get("MMO_CODEX_BIN") or str(settings.get("codex_bin", "codex")) ) if home is not None: selected_home = home.expanduser() elif os.environ.get("CODEX_HOME"): selected_home = Path(os.environ["CODEX_HOME"]).expanduser() else: selected_home = Path(str(settings.get("base_codex_home", "~/.codex"))).expanduser() return selected_binary, selected_home def _parser(*, prog: str = "codex-mmoctl") -> MMOArgumentParser: first_workflow = ( (prog, "Launch a new interactive session.") if prog == "codex-mmo" else (f"{prog} profile list", "Inspect installed profiles.") ) workflow_lines = "\n".join( f" {invocation:<36} {description}" for invocation, description in ( first_workflow, (f"{prog} resume --last", "Reattach to the latest resumable session here."), (f"{prog} session list", "Inspect retained session state."), (f"{prog} jobs list --session ID", "Inspect workers belonging to one session."), ) ) parser = MMOArgumentParser( prog=prog, description=( "Run persistent Codex MMO sessions and inspect their profiles, workers, " "gateways, and retained evidence." ), epilog=( "Common workflows:\n" f"{workflow_lines}\n\n" f"Run '{prog} COMMAND --help' for command-specific usage. Structured commands\n" "use human output on a terminal and JSON when redirected; --json forces JSON." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--json", action="store_true", dest="global_json", help="force structured JSON output where the command supports it", ) parser.add_argument( "-q", "--quiet", action="store_true", help="suppress MMO progress messages; requested output and errors remain", ) parser.add_argument( "--debug", action="store_true", help="include exception details and tracebacks when a command fails", ) parser.add_argument( "--version", action="store_true", dest="show_version", help="print the Codex MMO version and exit", ) sub = parser.add_subparsers(dest="command", metavar="COMMAND") def command( actions: argparse._SubParsersAction[Any], name: str, summary: str, *, description: str | None = None, epilog: str | None = None, ) -> MMOArgumentParser: return actions.add_parser( name, help=summary, description=description or summary.capitalize() + ".", epilog=epilog, formatter_class=argparse.RawDescriptionHelpFormatter, ) command(sub, "version", "print the Codex MMO package version") run = command( sub, "run", "launch a new interactive Codex session", description=( "Create a new immutable MMO session and attach the stock Codex TUI. " "Use resume to reattach to existing work." ), epilog=f"Pass Codex-owned options after '--', for example:\n {prog} run -- --search", ) run.add_argument("--profile", "-p", metavar="PROFILE", help="profile ID (default: active)") run.add_argument( "--cwd", "-C", metavar="PATH", default=os.getcwd(), help="session working directory" ) run.add_argument( "--bind", metavar="AGENT=MODEL", action="append", default=[], help="override one profile agent binding; repeat for multiple agents", ) run.add_argument( "codex_args", metavar="CODEX_ARG", nargs=argparse.REMAINDER, help="arguments forwarded to Codex; put '--' before Codex-owned options", ) resume = command( sub, "resume", "reattach to a resumable persistent MMO session", description=( "Reattach to the exact retained session and current root-thread generation. " "Terminal sessions cannot be resumed." ), ) resume.add_argument( "identifier", metavar="SESSION_OR_THREAD_ID", nargs="?", help="session or root thread ID" ) resume.add_argument( "--last", action="store_true", help="select the most recently active resumable session" ) resume.add_argument( "--all", action="store_true", dest="all_cwds", help="with --last, search every working directory instead of the current one", ) resume.add_argument( "--allow-tainted", action="store_true", help="allow reattachment to state marked unsafe after a boundary violation", ) execute = command( sub, "exec", "run one noninteractive root task", description=( "Run one task through the persistent app-server host without a TUI. Supply PROMPT " "or pipe non-empty UTF-8 text on stdin." ), epilog=f"Example:\n printf '%s\\n' 'Inspect the failure' | {prog} exec --profile PROFILE", ) execute.add_argument("prompt", metavar="PROMPT", nargs="?", help="task prompt") execute.add_argument("--profile", "-p", metavar="PROFILE", help="profile ID (default: active)") execute.add_argument( "--cwd", "-C", metavar="PATH", default=os.getcwd(), help="task working directory" ) execute.add_argument( "--bind", metavar="AGENT=MODEL", action="append", default=[], help="override one profile agent binding; repeat for multiple agents", ) execute.add_argument( "--image", metavar="PATH", action="append", default=[], help="attach an image; repeatable" ) execute.add_argument( "--wall-timeout", metavar="SECONDS", type=_bounded_int(1, 172_800), help="detach the caller after this harness limit while preserving active work (1-172800)", ) execute.add_argument( "--sandbox", choices=["read-only", "workspace-write"], default=None, help="override the root task sandbox mode", ) profile = command(sub, "profile", "inspect and manage composition profiles") ps = profile.add_subparsers(dest="profile_command", required=True, metavar="ACTION") command(ps, "list", "list installed profiles") command(ps, "current", "print the active profile ID") show = command(ps, "show", "show one profile") show.add_argument("profile", metavar="PROFILE", help="profile ID or profile-pack path") show.add_argument( "--resolved", action="store_true", help="include inherited defaults and resolved bindings" ) validate = command(ps, "validate", "validate one profile without launching models") validate.add_argument( "profile", metavar="PROFILE", nargs="?", help="profile ID (default: active)" ) validate.add_argument( "--bind", metavar="AGENT=MODEL", action="append", default=[], help="validate a binding override", ) compile_p = command(ps, "compile", "compile one immutable profile snapshot") compile_p.add_argument( "profile", metavar="PROFILE", nargs="?", help="profile ID (default: active)" ) compile_p.add_argument( "--bind", metavar="AGENT=MODEL", action="append", default=[], help="compile a binding override", ) compile_p.add_argument( "--force", action="store_true", help="rebuild the snapshot even when an identical one exists", ) use = command(ps, "use", "select the default profile") use.add_argument("profile", metavar="PROFILE", help="installed profile ID") install = command(ps, "install", "install a static profile pack") install.add_argument("source", metavar="PATH", help="profile-pack directory") install.add_argument( "--replace", action="store_true", help="replace an installed user profile with the same ID" ) remove = command(ps, "remove", "remove an installed user profile") remove.add_argument("profile", metavar="PROFILE", help="user profile ID") clone = command(ps, "clone", "clone a profile into user configuration") clone.add_argument("source", metavar="SOURCE", help="source profile ID") clone.add_argument("destination", metavar="DESTINATION", help="new user profile ID") clone.add_argument( "--replace", action="store_true", help="replace an existing destination profile" ) doctor_p = command(ps, "doctor", "diagnose one profile and its runtime dependencies") doctor_p.add_argument( "profile", metavar="PROFILE", nargs="?", help="profile ID (default: active)" ) doctor_p.add_argument( "--live", action="store_true", help="check live binaries, credentials, and routes" ) doctor_p.add_argument( "--probe", action="store_true", help="with --live, send a real root-model probe request" ) doctor_p.add_argument( "--bind", metavar="AGENT=MODEL", action="append", default=[], help="diagnose a binding override", ) smoke = command(ps, "smoke", "exercise a profile through its declared execution backends") smoke.add_argument("profile", metavar="PROFILE", nargs="?", help="profile ID (default: active)") smoke.add_argument( "--cwd", "-C", metavar="PATH", default=os.getcwd(), help="temporary smoke workspace parent" ) smoke.add_argument( "--bind", metavar="AGENT=MODEL", action="append", default=[], help="smoke-test a binding override", ) smoke_selection = smoke.add_mutually_exclusive_group() smoke_selection.add_argument( "--root-only", action="store_true", help="exercise only the root agent" ) smoke_selection.add_argument( "--workers-only", action="store_true", help="exercise only worker agents" ) tool_mcp = command( sub, "tool-mcp", "inspect operator-owned third-party Tool MCP definitions", ) tms = tool_mcp.add_subparsers(dest="tool_mcp_command", required=True, metavar="ACTION") command(tms, "list", "list configured Tool MCP servers") tool_mcp_show = command(tms, "show", "show one server definition and readiness report") tool_mcp_show.add_argument("server", metavar="SERVER", help="Tool MCP server ID") tool_mcp_validate = command(tms, "validate", "validate one or all Tool MCP definitions") tool_mcp_validate.add_argument( "server", metavar="SERVER", nargs="?", help="server ID (default: all)" ) catalog = command(sub, "catalog", "inspect and verify the provider/model catalog") cs = catalog.add_subparsers(dest="catalog_command", required=True, metavar="ACTION") summary = command(cs, "summary", "summarize routes, models, and resource groups") summary.add_argument("--profile", metavar="PROFILE", help="apply one profile's catalog overlay") models = command(cs, "models", "list catalog models") models.add_argument("--profile", metavar="PROFILE", help="apply one profile's catalog overlay") models.add_argument("--route", metavar="ROUTE", help="filter by route ID") models.add_argument("--inventory", metavar="INVENTORY", help="filter by inventory source") models.add_argument("--query", metavar="TEXT", help="case-insensitive model search") models.add_argument("--availability", metavar="STATUS", help="filter by observed availability") compatibility = models.add_mutually_exclusive_group() compatibility.add_argument( "--agent-compatible", action="store_true", help="show only executable agent models" ) compatibility.add_argument( "--catalog-only", action="store_true", help="show only inventory-only models" ) model = command(cs, "model", "show one route-qualified model") model.add_argument("key", metavar="ROUTE__MODEL_KEY", help="route-qualified model key") model.add_argument("--profile", metavar="PROFILE", help="apply one profile's catalog overlay") routes = command(cs, "routes", "list catalog routes") routes.add_argument("--profile", metavar="PROFILE", help="apply one profile's catalog overlay") resources = command(cs, "resources", "list shared capacity resource groups") resources.add_argument( "--profile", metavar="PROFILE", help="apply one profile's catalog overlay" ) command(cs, "inventory", "verify the bundled local inventory snapshot") verify = command(cs, "verify", "verify local and optional live catalog evidence") verify.add_argument( "--remote", action="store_true", help="query configured remote provider catalogs" ) verify.add_argument( "--codex", action="store_true", help="query the installed Codex model catalog" ) verify.add_argument("--opencode-url", metavar="URL", help="override the OpenCode Go models URL") verify.add_argument( "--opencode-zen-url", metavar="URL", help="override the OpenCode Zen models URL" ) verify.add_argument( "--openrouter-url", metavar="URL", help="override the OpenRouter models URL" ) verify.add_argument( "--zai-coding-url", metavar="URL", help="override the Z.AI Coding models URL" ) verify.add_argument("--codex-bin", metavar="PATH", help="Codex executable override") verify.add_argument("--codex-home", metavar="PATH", type=Path, help="base Codex home override") verify.add_argument( "--timeout", metavar="SECONDS", type=_bounded_float(0.1, 300.0), default=10.0, help="per-request timeout (default: 10)", ) discover = command(cs, "discover", "query one provider's live model inventory") discover.add_argument( "source", metavar="SOURCE", choices=[ "codex", "opencode-go", "opencode-zen", "openrouter", "zai-api", "zai-coding-plan", ], help="provider inventory to query", ) discover.add_argument("--url", metavar="URL", help="provider URL override where supported") discover.add_argument("--opencode-url", metavar="URL", help="OpenCode Go URL override") discover.add_argument("--opencode-zen-url", metavar="URL", help="OpenCode Zen URL override") discover.add_argument("--codex-bin", metavar="PATH", help="Codex executable override") discover.add_argument( "--codex-home", metavar="PATH", type=Path, help="base Codex home override" ) discover.add_argument( "--timeout", metavar="SECONDS", type=_bounded_float(0.1, 300.0), help="request timeout for remote sources (default: 10)", ) refresh = command(cs, "refresh", "refresh retained discovery evidence") refresh.add_argument("--no-remote", action="store_true", help="skip remote provider discovery") refresh.add_argument("--no-codex", action="store_true", help="skip installed Codex discovery") refresh.add_argument( "--install-codex-overlay", action="store_true", help="install the successfully discovered Codex catalog overlay", ) refresh.add_argument( "--opencode-url", metavar="URL", help="override the OpenCode Go models URL" ) refresh.add_argument( "--opencode-zen-url", metavar="URL", help="override the OpenCode Zen models URL" ) refresh.add_argument( "--openrouter-url", metavar="URL", help="override the OpenRouter models URL" ) refresh.add_argument( "--zai-coding-url", metavar="URL", help="override the Z.AI Coding models URL" ) refresh.add_argument("--codex-bin", metavar="PATH", help="Codex executable override") refresh.add_argument("--codex-home", metavar="PATH", type=Path, help="base Codex home override") refresh.add_argument( "--timeout", metavar="SECONDS", type=_bounded_float(0.1, 300.0), default=10.0, help="per-request timeout (default: 10)", ) gateway = command(sub, "gateway", "inspect and control local Switchyard gateways") gs = gateway.add_subparsers(dest="gateway_command", required=True, metavar="ACTION") command(gs, "list", "list retained gateway processes") gateway_summaries = { "start": "start the gateway required by a profile", "stop": "stop the gateway used by a profile", "status": "show gateway status for a profile", "models": "query models exposed by a profile gateway", "logs": "print the tail of a profile gateway log", } for name, summary_text in gateway_summaries.items(): item = command(gs, name, summary_text) item.add_argument( "profile", metavar="PROFILE", nargs="?", help="profile ID (default: active)" ) if name == "logs": item.add_argument( "--lines", metavar="COUNT", type=_bounded_int(1, 100_000), default=100, help="lines to print (default: 100)", ) command(gs, "stop-idle", "stop gateways whose idle timeout has elapsed") session = command(sub, "session", "inspect and control persistent MMO sessions") ss = session.add_subparsers(dest="session_command", required=True, metavar="ACTION") list_s = command(ss, "list", "list retained sessions, newest activity first") list_s.add_argument( "--limit", metavar="COUNT", type=_bounded_int(1, 10_000), default=50, help="maximum sessions (default: 50)", ) show_s = command(ss, "show", "show lifecycle and recovery details for one session") show_s.add_argument("session_id", metavar="SESSION_ID", help="MMO session ID") runs_s = command(ss, "runs", "list the immutable run belonging to one session") runs_s.add_argument("session_id", metavar="SESSION_ID", help="MMO session ID") runs_s.add_argument( "--limit", metavar="COUNT", type=_bounded_int(1, 10_000), default=50, help="maximum runs (default: 50)", ) cancel_s = command(ss, "cancel", "immediately terminate a session while retaining evidence") cancel_s.add_argument("session_id", metavar="SESSION_ID", help="MMO session ID") detach_s = command(ss, "detach", "disconnect the client while admitted work continues") detach_s.add_argument("session_id", metavar="SESSION_ID", help="MMO session ID") pause_s = command(ss, "pause", "checkpoint and cold-pause a session and its workers") pause_s.add_argument("session_id", metavar="SESSION_ID", help="MMO session ID") compact_s = command(ss, "compact", "compact a paused root thread and return it to cold pause") compact_s.add_argument("session_id", metavar="SESSION_ID", help="paused MMO session ID") continue_s = command(ss, "continue", "reactivate the same paused root and worker threads") continue_s.add_argument("session_id", metavar="SESSION_ID", help="paused MMO session ID") continue_s.add_argument( "--input", metavar="TEXT", help="new operator input delivered on reactivation" ) continue_s.add_argument( "--goal-token-budget", metavar="TOKENS", type=_bounded_int(10_000, 100_000_000), help="increase the total goal budget within the compiled ceiling", ) stop_s = command(ss, "stop", "request evidence finalization, then retire every host") stop_s.add_argument("session_id", metavar="SESSION_ID", help="MMO session ID") stop_s.add_argument( "--grace", metavar="SECONDS", type=_bounded_int(0, 3600), default=120, help="finalization grace before forced retirement (default: 120)", ) jobs = command(sub, "jobs", "inspect worker jobs and retrieve retained evidence") js = jobs.add_subparsers(dest="jobs_command", required=True, metavar="ACTION") list_j = command(js, "list", "list retained jobs, newest first") list_j.add_argument("--session", metavar="SESSION_ID", help="filter by session") list_j.add_argument("--run", metavar="RUN_ID", help="filter by immutable run") list_j.add_argument( "--limit", metavar="COUNT", type=_bounded_int(1, 10_000), default=50, help="maximum jobs (default: 50)", ) status_j = command(js, "status", "show one or more jobs in the supplied order") status_j.add_argument("job_ids", metavar="JOB_ID", nargs="+", help="job IDs") result_j = command( js, "result", "read one losslessly paged job result", epilog="Follow next_cursor until it is null; concatenate text pages in cursor order.", ) result_j.add_argument("job_id", metavar="JOB_ID", help="job ID") result_j.add_argument( "--cursor", metavar="OFFSET", type=_bounded_int(0, 2_147_483_647), default=0, help="character cursor (default: 0)", ) result_j.add_argument( "--max-chars", metavar="COUNT", type=_bounded_int(500, 500_000), help="maximum characters in this page (500-500000)", ) wait_j = command(js, "wait", "wait briefly for one or more jobs without discarding work") wait_j.add_argument("job_ids", metavar="JOB_ID", nargs="+", help="job IDs") wait_j.add_argument( "--timeout", metavar="SECONDS", type=_bounded_int(0, 120), default=30, help="caller wait limit (default: 30; maximum: 120)", ) wait_j.add_argument("--session", metavar="SESSION_ID", required=True, help="owning session ID") cancel_j = command(js, "cancel", "cancel a job and, by default, its descendants") cancel_j.add_argument("job_id", metavar="JOB_ID", help="job ID") cancel_j.add_argument("--no-cascade", action="store_true", help="leave descendant jobs running") cancel_j.add_argument("--reason", metavar="TEXT", help="operator cancellation reason") eval_p = command(sub, "eval", "validate and run profile evaluation suites") es = eval_p.add_subparsers(dest="eval_command", required=True, metavar="ACTION") command(es, "suites", "list installed evaluation suites") ev = command(es, "validate", "validate one evaluation suite without running it") ev.add_argument("suite", metavar="SUITE", help="suite ID or path") er = command(es, "run", "run one profile against an evaluation suite") er.add_argument("--profile", "-p", metavar="PROFILE", required=True, help="profile ID") er.add_argument( "--suite", metavar="SUITE", default="codex-harness", help="suite ID (default: codex-harness)", ) er.add_argument( "--bind", metavar="AGENT=MODEL", action="append", default=[], help="override one profile binding; repeatable", ) er.add_argument( "--wall-timeout", metavar="SECONDS", type=_bounded_int(1, 172_800), help="override each task's external harness limit", ) er.add_argument( "--trial-mode", choices=("development", "release"), default="development", help="use the suite's development or release trial count (default: development)", ) er.add_argument( "--dry-run", action="store_true", help="validate and plan trials without launching agents" ) el = command(es, "list", "list retained evaluation runs") el.add_argument( "--limit", metavar="COUNT", type=_bounded_int(1, 10_000), default=100, help="maximum runs (default: 100)", ) eshow = command(es, "show", "show one retained evaluation run") eshow.add_argument("run_id", metavar="RUN_ID", help="evaluation run ID") ecompare = command(es, "compare", "compare two or more retained evaluation runs") ecompare.add_argument("run_ids", metavar="RUN_ID", nargs="+", help="at least two run IDs") doc = command(sub, "doctor", "diagnose the active or selected profile") doc.add_argument("--profile", "-p", metavar="PROFILE", help="profile ID (default: active)") doc.add_argument( "--live", action="store_true", help="check live binaries, credentials, and routes" ) doc.add_argument( "--probe", action="store_true", help="with --live, send a real root-model probe request" ) doc.add_argument( "--bind", metavar="AGENT=MODEL", action="append", default=[], help="diagnose a binding override", ) validate_all = command( sub, "validate", "validate profiles, catalog inventory, and evaluation suites" ) validate_all.add_argument( "--all-profiles", action="store_true", help="validate every installed profile instead of only the active profile", ) clean = command( sub, "clean", "delete terminal jobs and sessions older than their retention thresholds", description=( "Permanently delete only terminal job and session state older than the selected " "thresholds. Active work is never eligible. Use --dry-run to preview counts." ), ) clean.add_argument( "--job-days", metavar="DAYS", type=_bounded_int(0, 36_500), help="job retention threshold (default: settings.toml)", ) clean.add_argument( "--session-days", metavar="DAYS", type=_bounded_int(0, 36_500), help="session retention threshold (default: settings.toml)", ) clean.add_argument( "--dry-run", action="store_true", help="count eligible records without deleting them" ) command(sub, "prompt", "print the reusable orchestration prompt") auth = command(sub, "auth", "manage built-in Codex authentication in the base Codex home") aus = auth.add_subparsers(dest="auth_command", required=True, metavar="ACTION") command(aus, "status", "show Codex login status for the configured base home") command(aus, "login", "run the interactive Codex login flow in the configured base home") return parser _GLOBAL_FLAGS = frozenset({"--json", "-q", "--quiet", "--debug"}) def _subparser_action(parser: argparse.ArgumentParser) -> argparse._SubParsersAction[Any] | None: return next( (action for action in parser._actions if isinstance(action, argparse._SubParsersAction)), None, ) def _normalize_global_options(argv: list[str]) -> list[str]: """Allow MMO global flags anywhere before the explicit passthrough separator.""" global_flags: list[str] = [] remaining: list[str] = [] passthrough = False for token in argv: if token == "--": passthrough = True remaining.append(token) elif not passthrough and token in _GLOBAL_FLAGS: if token not in global_flags: global_flags.append(token) else: remaining.append(token) return [*global_flags, *remaining] def _route_argv( parser: argparse.ArgumentParser, argv: list[str], *, implicit_run: bool, ) -> list[str]: normalized = _normalize_global_options(argv) if not implicit_run: return normalized split = 0 while split < len(normalized) and normalized[split] in _GLOBAL_FLAGS: split += 1 globals_prefix = normalized[:split] command_argv = normalized[split:] top_level = _subparser_action(parser) choices = set(top_level.choices) if top_level is not None else set() if not command_argv: return [*globals_prefix, "run"] first = command_argv[0] if first in {"-h", "--help", "--version"} or first in choices: return normalized return [*globals_prefix, "run", *command_argv] def _selected_parser( parser: argparse.ArgumentParser, argv: list[str], ) -> argparse.ArgumentParser: current = parser for token in argv: if token == "--": break if token in _GLOBAL_FLAGS or token == "--version": continue action = _subparser_action(current) if action is not None and token in action.choices: current = action.choices[token] return current def _usage_hint( message: str, selected: argparse.ArgumentParser, root: argparse.ArgumentParser, ) -> str | None: candidates: list[str] = [] action = _subparser_action(selected) if action is not None: candidates.extend(action.choices) for parser in (selected, root): candidates.extend(parser._option_string_actions) invalid_choice = re.search(r"invalid choice: ['\"]([^'\"]+)['\"]", message) unknown_option = re.search(r"unrecognized arguments?:\s+(--?[A-Za-z0-9][\w-]*)", message) value = invalid_choice.group(1) if invalid_choice else None if value is None and unknown_option: value = unknown_option.group(1) if value is None: return None matches = difflib.get_close_matches(value, sorted(set(candidates)), n=1, cutoff=0.72) return f"Did you mean {matches[0]!r}?" if matches else None def _command_path(args: argparse.Namespace) -> str: parts = [str(args.command)] if getattr(args, "command", None) else [] for attribute in ( "profile_command", "tool_mcp_command", "catalog_command", "gateway_command", "session_command", "jobs_command", "eval_command", "auth_command", ): value = getattr(args, attribute, None) if value: parts.append(str(value)) return ".".join(parts) def _validate_cli_args(args: argparse.Namespace) -> None: path = _command_path(args) if args.show_version and args.command is not None: raise CLIUsageError("--version cannot be combined with a command") if not args.show_version and args.command is None: raise CLIUsageError( "a command is required", hint="Run with --help to see the available commands.", ) if args.global_json and path in { "run", "resume", "gateway.logs", "auth.status", "auth.login", }: raise CLIUsageError( f"--json is not supported by {path.replace('.', ' ')}", hint="This command attaches a terminal UI or forwards a raw external text stream.", ) if path == "resume": if bool(args.identifier) == bool(args.last): raise CLIUsageError("provide exactly one SESSION_OR_THREAD_ID or --last") if args.all_cwds and not args.last: raise CLIUsageError("--all is valid only with --last") if path in {"doctor", "profile.doctor"} and args.probe and not args.live: raise CLIUsageError( "--probe requires --live", hint="A probe sends a real model request after live dependency checks pass.", ) if path == "eval.compare" and len(args.run_ids) < 2: raise CLIUsageError("compare requires at least two RUN_ID arguments") if path == "catalog.refresh" and args.no_codex and args.install_codex_overlay: raise CLIUsageError("--no-codex cannot be combined with --install-codex-overlay") if path == "catalog.refresh" and args.no_remote: remote_overrides = [ name for name, value in ( ("--opencode-url", args.opencode_url), ("--opencode-zen-url", args.opencode_zen_url), ("--openrouter-url", args.openrouter_url), ("--zai-coding-url", args.zai_coding_url), ) if value is not None ] if remote_overrides: raise CLIUsageError( f"--no-remote cannot be combined with {', '.join(remote_overrides)}" ) if path == "catalog.refresh" and args.no_codex: codex_overrides = [ name for name, value in ( ("--codex-bin", args.codex_bin), ("--codex-home", args.codex_home), ) if value is not None ] if codex_overrides: raise CLIUsageError(f"--no-codex cannot be combined with {', '.join(codex_overrides)}") if path == "catalog.discover": invalid: list[str] = [] if args.source == "codex": invalid.extend( name for name, value in ( ("--url", args.url), ("--opencode-url", args.opencode_url), ("--opencode-zen-url", args.opencode_zen_url), ("--timeout", args.timeout), ) if value is not None ) else: invalid.extend( name for name, value in ( ("--codex-bin", args.codex_bin), ("--codex-home", args.codex_home), ) if value is not None ) if args.source != "opencode-go" and args.opencode_url is not None: invalid.append("--opencode-url") if args.source != "opencode-zen" and args.opencode_zen_url is not None: invalid.append("--opencode-zen-url") if ( args.source == "opencode-go" and args.url is not None and args.opencode_url is not None ): raise CLIUsageError("use either --url or --opencode-url, not both") if ( args.source == "opencode-zen" and args.url is not None and args.opencode_zen_url is not None ): raise CLIUsageError("use either --url or --opencode-zen-url, not both") if invalid: raise CLIUsageError( f"{', '.join(sorted(set(invalid)))} not valid for catalog source {args.source!r}" ) def _exec_prompt(args: argparse.Namespace) -> str: if args.prompt is not None: if not args.prompt.strip(): raise CLIUsageError("PROMPT must contain non-whitespace text") return str(args.prompt) if bool(getattr(sys.stdin, "isatty", lambda: False)()): raise CLIUsageError( "exec requires PROMPT or piped stdin", hint="Supply a positional prompt or pipe non-empty text into codex-mmo exec.", ) prompt = sys.stdin.read() if not prompt.strip(): raise CLIUsageError("piped stdin for exec must contain non-whitespace text") return prompt def _error_category(exc: Exception) -> str: if isinstance(exc, FileNotFoundError): return "not_found" if isinstance(exc, ValueError): return "invalid_input" if isinstance(exc, (RuntimeError, TimeoutError, ConnectionError, OSError)): return "runtime" return "internal" def _error_hint(path: str, exc: Exception) -> str | None: if isinstance(exc, FileNotFoundError): if path.startswith("profile"): return "Run 'codex-mmo profile list' to inspect installed profiles." if path.startswith("session") or path == "resume": return "Run 'codex-mmo session list' to inspect retained and resumable sessions." if path.startswith("jobs"): return "Run 'codex-mmo jobs list' with the owning session or run filter." if path.startswith("catalog"): return "Run 'codex-mmo catalog --help' to inspect discovery and verification options." return None def _broken_pipe_status() -> int: """Retire stdout so interpreter shutdown cannot print another pipe error.""" try: descriptor = os.open(os.devnull, os.O_WRONLY) os.dup2(descriptor, sys.stdout.fileno()) os.close(descriptor) except OSError: pass return 141 def _handle_catalog(args: argparse.Namespace) -> int: """Dispatch catalog commands outside the general control-plane router.""" cmd = args.catalog_command if cmd == "summary": emit_structured( "catalog.summary", catalog_summary(args.profile), force_json=args.global_json ) return 0 if cmd == "models": compatible = True if args.agent_compatible else (False if args.catalog_only else None) emit_structured( "catalog.models", list_models( profile=args.profile, route=args.route, inventory=args.inventory, query=args.query, agent_compatible=compatible, availability=args.availability, ), force_json=args.global_json, ) return 0 if cmd == "model": emit_structured( "catalog.model", find_model(args.key, profile=args.profile), force_json=args.global_json, ) return 0 if cmd in {"routes", "resources"}: emit_structured( f"catalog.{cmd}", catalog_data(args.profile)[cmd], force_json=args.global_json ) return 0 if cmd == "inventory": report = local_inventory_report() emit_structured("catalog.inventory", report, force_json=args.global_json) return 0 if report["passed"] else 1 if cmd == "verify": progress("Verifying catalog evidence...", quiet=args.quiet) codex_binary, codex_home = _configured_codex_context(args.codex_bin, args.codex_home) report = verify_catalog( remote=args.remote or any( value is not None for value in ( args.opencode_url, args.opencode_zen_url, args.openrouter_url, args.zai_coding_url, ) ), include_codex=args.codex or args.codex_bin is not None or args.codex_home is not None, opencode_url=args.opencode_url, opencode_zen_url=args.opencode_zen_url, openrouter_url=args.openrouter_url, zai_coding_url=args.zai_coding_url, codex_binary=codex_binary, codex_home=codex_home, timeout=args.timeout, ) emit_structured("catalog.verify", report, force_json=args.global_json) return 0 if report["passed"] else 1 if cmd == "discover": progress(f"Discovering models from {args.source}...", quiet=args.quiet) if args.source == "codex": codex_binary, codex_home = _configured_codex_context(args.codex_bin, args.codex_home) result = discover_codex(binary=codex_binary, home=codex_home) elif args.source == "opencode-go": result = discover_opencode_go( url=args.url or args.opencode_url, timeout=args.timeout or 10.0 ) elif args.source == "opencode-zen": result = discover_opencode_zen( url=args.url or args.opencode_zen_url, timeout=args.timeout or 10.0 ) elif args.source == "openrouter": result = discover_openrouter(url=args.url, timeout=args.timeout or 10.0) else: result = discover_zai(args.source, url=args.url, timeout=args.timeout or 10.0) emit_structured("catalog.discover", result, force_json=args.global_json) return 0 if result.get("passed") else 1 if cmd == "refresh": progress("Refreshing retained catalog discovery evidence...", quiet=args.quiet) codex_binary, codex_home = _configured_codex_context(args.codex_bin, args.codex_home) result = refresh_discovery( remote=not args.no_remote, include_codex=not args.no_codex, install_codex_overlay=args.install_codex_overlay, opencode_url=args.opencode_url, opencode_zen_url=args.opencode_zen_url, openrouter_url=args.openrouter_url, zai_coding_url=args.zai_coding_url, codex_binary=codex_binary, codex_home=codex_home, timeout=args.timeout, ) emit_structured("catalog.refresh", result, force_json=args.global_json) return 0 if result["report"]["passed"] else 1 raise ValueError(f"unhandled catalog command: {cmd}") def _handle_profile(args: argparse.Namespace) -> int: """Dispatch profile commands at their existing domain boundary.""" cmd = args.profile_command if cmd == "list": items = discover_profiles() if stdout_is_tty() and not args.global_json: active = active_profile_id() items = {key: {**item, "active": key == active} for key, item in items.items()} emit_structured("profile.list", items, force_json=args.global_json) return 0 if cmd == "current": profile_id = active_profile_id() emit_scalar( profile_id, json_value={"profile_id": profile_id}, force_json=args.global_json, ) return 0 if cmd == "show": emit_structured( "profile.show", resolve_profile(args.profile) if args.resolved else profile_summary(args.profile), force_json=args.global_json, ) return 0 if cmd == "validate": result = profile_validation_report( args.profile or active_profile_id(), _bindings(args.bind) ) emit_structured("profile.validate", result, force_json=args.global_json) return 0 if result["valid"] else 1 if cmd == "compile": selected_profile = args.profile or active_profile_id() progress(f"Compiling profile {selected_profile}...", quiet=args.quiet) snap = compile_profile( selected_profile, bindings=_bindings(args.bind), force=args.force, ) emit_structured("profile.compile", snap["manifest"], force_json=args.global_json) return 0 if cmd == "use": selected = profile_summary(args.profile) set_active_profile(args.profile) if selected["maturity"] == "lab": print( f"warning: {args.profile} is an experimental lab profile", file=sys.stderr, ) emit_scalar( args.profile, json_value={"profile_id": args.profile, "selected": True}, force_json=args.global_json, ) return 0 if cmd == "install": profile_id = install_profile_pack(Path(args.source), replace=args.replace) emit_scalar( profile_id, json_value={"profile_id": profile_id, "installed": True}, force_json=args.global_json, ) return 0 if cmd == "remove": remove_profile(args.profile) emit_tty_success( f"Removed profile {args.profile}.", json_value={"profile_id": args.profile, "removed": True}, force_json=args.global_json, ) return 0 if cmd == "clone": destination_path = clone_profile(args.source, args.destination, replace=args.replace) emit_scalar( args.destination, json_value={ "profile_id": args.destination, "path": str(destination_path), "source_profile_id": args.source, "cloned": True, }, force_json=args.global_json, ) return 0 if cmd == "doctor": selected_profile = args.profile or active_profile_id() progress(f"Diagnosing profile {selected_profile}...", quiet=args.quiet) result = doctor( selected_profile, live=args.live, probe=args.probe, bindings=_bindings(args.bind), progress=lambda message: progress(message, quiet=args.quiet), ) emit_structured("profile.doctor", result, force_json=args.global_json) return 0 if result["passed"] else 1 if cmd == "smoke": selected_profile = args.profile or active_profile_id() progress(f"Smoke-testing profile {selected_profile}...", quiet=args.quiet) result = smoke_profile( selected_profile, cwd=args.cwd, bindings=_bindings(args.bind), root_only=args.root_only, workers_only=args.workers_only, progress=lambda message: progress(message, quiet=args.quiet), ) emit_structured("profile.smoke", result, force_json=args.global_json) return 0 if result["passed"] else 1 raise RuntimeError(f"unhandled profile command: {cmd}") def _handle_session(args: argparse.Namespace) -> int: """Dispatch durable session inspection and lifecycle controls.""" cmd = args.session_command if cmd == "list": sessions = sorted( iter_sessions(strict=False), key=lambda item: str( item.get("last_active_at") or item.get("finished_at") or item.get("created_at") or "" ), reverse=True, ) emit_structured( "session.list", [public_session(item) for item in sessions[: args.limit]], force_json=args.global_json, ) return 0 if cmd == "show": emit_structured( "session.show", public_session(load_session(args.session_id), include_details=True), force_json=args.global_json, ) return 0 if cmd == "runs": emit_structured( "session.runs", [public_run(item) for item in iter_session_runs(args.session_id)[: args.limit]], force_json=args.global_json, ) return 0 if cmd == "cancel": progress(f"Cancelling session {args.session_id}...", quiet=args.quiet) emit_structured( "session.cancel", cancel_session(args.session_id), force_json=args.global_json, ) return 0 if cmd == "detach": emit_structured( "session.detach", detach_session(args.session_id), force_json=args.global_json, ) return 0 if cmd == "pause": progress(f"Pausing session {args.session_id}...", quiet=args.quiet) emit_structured( "session.pause", pause_session(args.session_id), force_json=args.global_json, ) return 0 if cmd == "compact": progress(f"Compacting paused session {args.session_id}...", quiet=args.quiet) emit_structured( "session.compact", compact_session(args.session_id), force_json=args.global_json, ) return 0 if cmd == "continue": progress(f"Continuing paused session {args.session_id}...", quiet=args.quiet) emit_structured( "session.continue", continue_session( args.session_id, input_text=args.input, goal_token_budget=args.goal_token_budget, ), force_json=args.global_json, ) return 0 if cmd == "stop": progress( f"Stopping session {args.session_id} with {args.grace}s finalization grace...", quiet=args.quiet, ) emit_structured( "session.stop", stop_session(args.session_id, grace_seconds=args.grace), force_json=args.global_json, ) return 0 raise RuntimeError(f"unhandled session command: {cmd}") def _handle_jobs(args: argparse.Namespace) -> int: """Dispatch worker inspection, waiting, evidence, and cancellation.""" cmd = args.jobs_command if cmd == "list": emit_structured( "jobs.list", list_jobs(session_id=args.session, run_id=args.run, limit=args.limit), force_json=args.global_json, ) return 0 if cmd == "status": rows = list_jobs(job_ids=args.job_ids, limit=len(args.job_ids) + 5) by_id = {str(item["job_id"]): item for item in rows} missing = [item for item in args.job_ids if item not in by_id] if missing: raise FileNotFoundError("unknown jobs: " + ", ".join(missing)) emit_structured( "jobs.status", [by_id[item] for item in args.job_ids], force_json=args.global_json, ) return 0 if cmd == "result": emit_structured( "jobs.result", read_result(args.job_id, max_chars=args.max_chars, cursor=args.cursor), force_json=args.global_json, ) return 0 if cmd == "wait": progress( f"Waiting up to {args.timeout}s for {len(args.job_ids)} job(s)...", quiet=args.quiet, ) result = wait_for_jobs(args.job_ids, session_id=args.session, timeout_seconds=args.timeout) emit_structured("jobs.wait", result, force_json=args.global_json) return 1 if result["unfinished"] else 0 if cmd == "cancel": progress(f"Cancelling job {args.job_id}...", quiet=args.quiet) emit_structured( "jobs.cancel", cancel_job( args.job_id, cascade=not args.no_cascade, reason=args.reason, ), force_json=args.global_json, ) return 0 raise RuntimeError(f"unhandled jobs command: {cmd}") def _handle_eval(args: argparse.Namespace) -> int: """Dispatch evaluation suite and retained-run commands.""" cmd = args.eval_command if cmd == "suites": emit_structured("eval.suites", discover_suites(), force_json=args.global_json) return 0 if cmd == "validate": result = validate_suite(args.suite) emit_structured("eval.validate", result, force_json=args.global_json) return 0 if result["valid"] else 1 if cmd == "run": progress( f"Running evaluation suite {args.suite} for profile {args.profile}...", quiet=args.quiet, ) result = run_evaluation( profile=args.profile, suite=args.suite, bindings=_bindings(args.bind), wall_timeout_override=args.wall_timeout, dry_run=args.dry_run, trial_mode=args.trial_mode, progress=lambda message: progress(message, quiet=args.quiet), ) emit_structured("eval.run", result, force_json=args.global_json) if result["status"] not in {"validated", "completed"}: return 1 if args.trial_mode == "release" and result["status"] == "completed": promotion = result.get("summary", {}).get("promotion", {}) if promotion.get("eligible") and promotion.get("passed") is not True: return 1 return 0 if cmd == "list": emit_structured("eval.list", list_runs(args.limit), force_json=args.global_json) return 0 if cmd == "show": emit_structured("eval.show", load_run(args.run_id), force_json=args.global_json) return 0 if cmd == "compare": emit_structured("eval.compare", compare_runs(args.run_ids), force_json=args.global_json) return 0 raise RuntimeError(f"unhandled eval command: {cmd}") def _handle_gateway(args: argparse.Namespace) -> int: """Dispatch Switchyard gateway inspection and lifecycle commands.""" cmd = args.gateway_command if cmd == "list": emit_structured("gateway.list", list_gateways(), force_json=args.global_json) return 0 if cmd == "stop-idle": progress("Stopping idle gateways...", quiet=args.quiet) emit_structured( "gateway.stop-idle", {"stopped": stop_idle_gateways()}, force_json=args.global_json, ) return 0 profile = args.profile or active_profile_id() snap = compile_profile(profile) snapshot_hash = snap["manifest"]["snapshot_hash"] if cmd == "start": progress(f"Starting gateway for profile {profile}...", quiet=args.quiet) emit_structured( "gateway.start", ensure_gateway(snapshot_hash) or {"status": "not_required"}, force_json=args.global_json, ) return 0 if cmd == "stop": progress(f"Stopping gateway for profile {profile}...", quiet=args.quiet) emit_structured("gateway.stop", stop_gateway(snapshot_hash), force_json=args.global_json) return 0 if cmd == "status": emit_structured( "gateway.status", gateway_status(snapshot_hash), force_json=args.global_json ) return 0 if cmd == "models": emit_structured( "gateway.models", gateway_models(snapshot_hash), force_json=args.global_json ) return 0 if cmd == "logs": state = gateway_status(snapshot_hash) log_path = state.get("log_path") if not isinstance(log_path, str) or not Path(log_path).is_file(): raise FileNotFoundError(f"no gateway log is available for profile {profile!r}") print(_tail(Path(log_path), args.lines)) return 0 raise RuntimeError(f"unhandled gateway command: {cmd}") def main(argv: list[str] | None = None) -> int: raw_argv = list(sys.argv[1:] if argv is None else argv) entrypoint = os.environ.get("MMO_CLI_ENTRYPOINT", "codex-mmoctl") if entrypoint not in {"codex-mmo", "codex-mmoctl"}: entrypoint = "codex-mmoctl" parser = _parser(prog=entrypoint) routed_argv = _route_argv(parser, raw_argv, implicit_run=entrypoint == "codex-mmo") selected_parser = _selected_parser(parser, routed_argv) json_requested = "--json" in routed_argv try: args = parser.parse_args(routed_argv) _validate_cli_args(args) prompt = _exec_prompt(args) if args.command == "exec" else None if hasattr(args, "bind"): _bindings(args.bind) except CLIUsageError as exc: emit_usage_error( selected_parser.format_usage(), str(exc), hint=exc.hint or _usage_hint(str(exc), selected_parser, parser), as_json=json_requested, ) return 2 def structured(command: str, value: Any) -> None: emit_structured(command, value, force_json=args.global_json) try: if args.show_version or args.command == "version": version = package_version() emit_scalar( version, json_value={"version": version}, force_json=args.global_json, ) return 0 if args.command == "run": codex_args = args.codex_args if codex_args and codex_args[0] == "--": codex_args = codex_args[1:] if _codex_subcommand(codex_args) == "resume": raise ValueError("use 'codex-mmo resume' instead of 'codex-mmo run resume'") return shell_exit_status( launch_interactive( profile=args.profile, cwd=args.cwd, bindings=_bindings(args.bind), codex_args=codex_args, ) ) if args.command == "resume": session_id = resolve_resume_session( args.identifier, last=args.last, all_cwds=args.all_cwds, cwd=os.getcwd(), ) return shell_exit_status( resume_interactive(session_id, allow_tainted=args.allow_tainted) ) if args.command == "exec": progress("Running noninteractive root task...", quiet=args.quiet) result = run_root_exec( profile=args.profile, cwd=args.cwd, prompt=str(prompt), bindings=_bindings(args.bind), images=args.image, wall_timeout_seconds=args.wall_timeout, sandbox_mode=args.sandbox, ) if args.global_json: emit_json(result) else: print(result.get("result", "")) return shell_exit_status(int(result.get("exit_code", 1))) if args.command == "profile": return _handle_profile(args) if args.command == "tool-mcp": status = tool_mcp_status() cmd = args.tool_mcp_command if cmd == "list": structured("tool-mcp.list", status) return 0 server_id = args.server if server_id is not None and server_id not in status["servers"]: raise FileNotFoundError(f"unknown Tool MCP server {server_id!r}") if cmd == "show": if server_id is None: raise RuntimeError("tool-mcp show requires a server ID") registry, _sources = load_tool_mcp_registry_with_sources() structured( "tool-mcp.show", { "definition": registry[server_id], "readiness": status["servers"][server_id], }, ) return 0 selected = ( status["servers"] if server_id is None else {server_id: status["servers"][server_id]} ) result = { "registry_root": status["registry_root"], "servers": selected, "passed": all(bool(item["ready"]) for item in selected.values()), } structured("tool-mcp.validate", result) return 0 if result["passed"] else 1 if args.command == "catalog": return _handle_catalog(args) if args.command == "gateway": return _handle_gateway(args) if args.command == "session": return _handle_session(args) if args.command == "jobs": return _handle_jobs(args) if args.command == "eval": return _handle_eval(args) if args.command == "doctor": selected_profile = args.profile or active_profile_id() progress(f"Diagnosing profile {selected_profile}...", quiet=args.quiet) result = doctor( selected_profile, live=args.live, probe=args.probe, bindings=_bindings(args.bind), progress=lambda message: progress(message, quiet=args.quiet), ) structured("doctor", result) return 0 if result["passed"] else 1 if args.command == "validate": progress("Validating Codex MMO configuration...", quiet=args.quiet) profiles = discover_profiles() selected_profiles = sorted(profiles) if args.all_profiles else [active_profile_id()] validation_result: dict[str, Any] = { item: profile_validation_report(item, {}) for item in selected_profiles } validation_result["catalog"] = { "summary": catalog_summary(), "inventory": local_inventory_report(), } validation_result["evaluation_suites"] = { item: validate_suite(item) for item in discover_suites() } validation_result["passed"] = ( all( item["valid"] for key, item in validation_result.items() if key not in {"catalog", "evaluation_suites", "passed"} ) and validation_result["catalog"]["inventory"]["passed"] and all(item["valid"] for item in validation_result["evaluation_suites"].values()) ) structured("validate", validation_result) return 0 if validation_result["passed"] else 1 if args.command == "clean": settings = load_settings() job_days = ( args.job_days if args.job_days is not None else int(settings["job_retention_days"]) ) session_days = ( args.session_days if args.session_days is not None else int(settings["session_retention_days"]) ) counts = clean_state( job_days=job_days, session_days=session_days, dry_run=args.dry_run, ) structured( "clean", { "dry_run": args.dry_run, "job_days": job_days, "session_days": session_days, **counts, }, ) return 0 if args.command == "prompt": prompt_text = textwrap.dedent("""\ Analyze this repository using the active composition profile. Keep the immediate critical path in the root. Spawn independent participants only where they improve latency, specialization, or confidence. Continue useful non-overlapping root work while they run. Reconcile all material results from primary evidence, review changes, and run integrated validation before finalizing. """) emit_scalar( prompt_text.rstrip("\n"), json_value={"prompt": prompt_text.rstrip("\n")}, force_json=args.global_json, ) return 0 if args.command == "auth": settings = load_settings() home = Path(str(settings.get("base_codex_home", "~/.codex"))).expanduser() env = filtered_environment(extra={"CODEX_HOME": str(home)}) codex = os.environ.get("MMO_CODEX_BIN") or str(settings.get("codex_bin", "codex")) auth_command = ( [codex, "login", "status"] if args.auth_command == "status" else [codex, "login"] ) return shell_exit_status(subprocess.call(auth_command, env=env)) raise RuntimeError(f"unhandled command path: {_command_path(args)}") except CLIUsageError as exc: emit_usage_error( selected_parser.format_usage(), str(exc), hint=exc.hint, as_json=args.global_json, ) return 2 except KeyboardInterrupt: return 130 except BrokenPipeError: return _broken_pipe_status() except Exception as exc: category = _error_category(exc) message = str(exc) or type(exc).__name__ hint = _error_hint(_command_path(args), exc) if category == "internal": message = f"unexpected internal failure: {message}" hint = hint or "Rerun with --debug to include a traceback." emit_error( category=category, message=message, hint=hint, as_json=args.global_json, exception_type=type(exc).__name__ if args.debug else None, traceback_text=traceback.format_exc() if args.debug else None, ) return 1 if __name__ == "__main__": raise SystemExit(main())