#!/usr/bin/env python3
"""parable — multi-model coding orchestration dispatcher.

The orchestrating agent (the "brain") never parses provider configs or raw
harness logs; it calls these subcommands and reads their compact reports.

Subcommands:
  config [--validate] [--json]   Show merged config summary the brain reads at session start
  list                           One line per executor: id, model, cost, status, tags
  usage [--all] [--json]         Live subscription headroom per pool (zero model tokens)
  claude [-- <args...>]           Launch Claude Code through the configured localhost proxy
  finalize [--json]               Verify exact catalog ids and synchronize the named cast
  agents sync                     Synchronize Parable's project-local custom agents
  run <executor> <plan> [workdir] [--slug S] [--effort E]   Dispatch a plan to a codex- or pi-backed executor
  resume <run-dir|uuid> <delta prompt>            Continue a prior run's session with a fix-up
  status <run-dir>               6- or 7-line status parsed from the run's event stream
  verify [--when W] [--only a,b] [--targets T] [workdir]   Run configured deterministic checks
  review <executor> [workdir] [--author ID] [--base BRANCH] [--paths P]  Model code-review of the current diff

Requires Python 3.11+ (tomllib). `pip install tomli` covers 3.10.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
import textwrap
import threading
import time
import uuid
from collections.abc import Callable
from datetime import datetime, timezone
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import Request, urlopen

try:
    import tomllib
except ImportError:  # pragma: no cover
    try:
        import tomli as tomllib  # type: ignore
    except ImportError:
        sys.exit("parable requires Python >= 3.11 (or `pip install tomli`)")

SCHEMA_VERSION = 1
SUPPORTED_PROVIDER_TYPES = ("codex", "codex-native", "subagent", "pi", "cursor")
CHECK_WHEN_VALUES = ("post-implement", "pre-commit")
EFFORT_LEVELS = ("minimal", "low", "medium", "high", "xhigh", "max", "ultra")
# "max"/"ultra" ship with GPT-5.6-class models; "ultra" additionally flips codex into
# proactive multi-agent delegation (the model spawns its own subagent threads) — a
# deliberate batch-dispatch setting, not an escalation rung.
PI_THINKING_LEVELS = ("off", "minimal", "low", "medium", "high", "xhigh", "max")  # pi: "off" extra, no "ultra"
CLAUDE_SUBAGENT_EFFORT_LEVELS = ("low", "medium", "high", "xhigh", "max")
PI_API_VALUES = ("openai-completions", "openai-responses", "anthropic-messages")
PI_INSTALL_HINT = ("'pi' not found on PATH — npm i -g @earendil-works/pi-coding-agent "
                   "(requires node >= 22; pi crashes on node 20)")
CLAUDE_AGENT_ALIASES = frozenset((
    "inherit", "sonnet", "opus", "haiku", "best", "sonnet[1m]", "opus[1m]", "opusplan",
))
CLAUDE_CONFIG_FIELDS = frozenset(("base_url", "auth_token_env", "brain_model", "binary"))
PARABLE_AGENT_PREFIX = "parable-"
PARABLE_AGENT_MARKER = "<!-- Generated by @parcha/parable from parable.toml. -->"
CLAUDE_BRAIN_MODELS = {
    "fable": "claude-fable-5",
    "sol": "gpt-5.6-sol",
    "grok": "grok-4.6",
}
CLAUDE_BRAIN_MODES = ("auto", "fable", "sol", "grok", "config")
AUTO_BRAIN_TIGHT_PCT = 80.0
# Real context windows for models routed through the loopback proxy. Claude Code
# assumes 200k (or 1M via the [1m]/beta paths) for models it does not recognize,
# so without this table auto-compact fires far too late for proxied non-Anthropic
# models and sessions die with upstream "input exceeds the context window" 400s.
# Sources: provider documentation plus the pinned CLIProxyAPI registry
# (internal/registry/models/*.json). Sol supports a 1.05M provider window; Parable
# uses the documented 1M operating budget with 900k compaction rather than the
# registry's tuned 372k default. kimi-k3 is ~1M because the pinned proxy normalizes
# the upstream id to bare "k3" (router-for-me/CLIProxyAPI#4418). A `context_ktok`
# on the executor in parable.toml overrides this table.
MODEL_CONTEXT_WINDOWS = {
    "claude-fable-5": 1_000_000,
    "claude-sonnet-5": 1_000_000,
    "claude-opus-4-8": 1_000_000,
    "claude-haiku-4-5-20251001": 200_000,
    "gpt-5.6-sol": 1_000_000,
    "gpt-5.6-terra": 372_000,
    "gpt-5.6-luna": 372_000,
    "gpt-5.5": 272_000,
    "grok-4.6": 500_000,
    "kimi-k3": 1_000_000,
}
# Claude Code's own fallback for unrecognized models; used as the floor when a
# cast model's window is unknown so we never raise the assumed ceiling blindly.
CLAUDE_DEFAULT_CONTEXT_WINDOW = 200_000
CLAUDE_CONTEXT_ENV = "CLAUDE_CODE_MAX_CONTEXT_TOKENS"
CLAUDE_AUTO_COMPACT_WINDOW_ENV = "CLAUDE_CODE_AUTO_COMPACT_WINDOW"
CLAUDE_AUTO_COMPACT_PCT_ENV = "CLAUDE_AUTOCOMPACT_PCT_OVERRIDE"
# Claude Code's default auto-compact point is about 95%. That leaves too little
# room for a tool result between turns when a proxied model has a smaller input
# ceiling than Claude expects. Use a conservative default, while Sol follows the
# documented 1M operating budget and 900k compaction point.
CLAUDE_AUTO_COMPACT_PCT = 75
MODEL_AUTO_COMPACT_PCT = {
    "gpt-5.6-sol": 90,
}
CLAUDE_LONG_CONTEXT_MARKER = "[1m]"
CLAUDE_RESUME_COMPACT_MODEL = "claude-sonnet-5[1m]"
CLAUDE_RESUME_COMPACT_BASE_MODEL = "claude-sonnet-5"
CLAUDE_RESUME_COMPACT_WINDOW = 1_000_000
CLAUDE_RESUME_CHECK_TIMEOUT_SECONDS = 180
CLAUDE_RESUME_COMPACT_TIMEOUT_SECONDS = 900
CLAUDE_RESUME_HEARTBEAT_SECONDS = 60
PARABLE_CONTEXT_FAILURE_RECOVERY_ENV = "PARABLE_CONTEXT_FAILURE_RECOVERY"
CLAUDE_RESUME_COMPACT_PROMPT = (
    "/compact preserve the active task, user requirements, decisions, changed files, "
    "remaining work, and verification evidence"
)
PARABLE_WELCOME_ENV = "PARABLE_WELCOME_MESSAGE"
PARABLE_AGENT_STATE_ENV = "PARABLE_AGENT_STATE_JSON"
PARABLE_WELCOME_PLUGIN = Path(__file__).resolve().parent.parent / "runtime" / "welcome-plugin"
PARABLE_ASCII = (
    "                        _     _            _     ",
    "  _ __   __ _ _ __ __ _| |__ | | ___   ___| |__  ",
    " | '_ \\ / _` | '__/ _` | '_ \\| |/ _ \\ / __| '_ \\ ",
    " | |_) | (_| | | | (_| | |_) | |  __/_\\__ \\ | | |",
    " | .__/ \\__,_|_|  \\__,_|_.__/|_|\\___(_)___/_| |_|",
    " |_|",
)
PARABLE_ANIMALS = {
    "FABLE": "🐢",
    "SOL": "🐘",
    "TERRA": "🦊",
    "LUNA": "🐤",
    "SONNET": "🫏",
    "OPUS": "🦉",
    "HAIKU": "🐦",
    "GROK": "🐺",
    "KIMI": "🐯",
}

# Tier-0 defaults: Claude-native executors need no API keys — they run as
# subagents inside the orchestrating session. Anything with an env_key must
# be declared explicitly by the user.
BUILTIN_DEFAULTS: dict = {
    "parable": {
        "version": SCHEMA_VERSION,
        "log_dir": ".parable",
        "default_executor": "sonnet",
        "default_reviewer": "opus",
        "repo_notes": "",
    },
    "providers": {
        "claude": {"type": "subagent"},
    },
    "executors": {
        "sonnet": {
            "provider": "claude",
            "model": "sonnet",
            "tags": ["implementer", "default"],
            "use_for": (
                "Default implementer: features, bugfixes, tests from a fully-specified plan. "
                "Literal instruction-follower — state scope explicitly in the plan."
            ),
            "avoid_for": "Ambiguous architecture decisions; repo-wide refactors needing huge context.",
        },
        "opus": {
            "provider": "claude",
            "model": "opus",
            "tags": ["smoke-test", "reviewer", "second-opinion"],
            "use_for": (
                "Smoke-testing against the running stack (instruct it explicitly to execute real "
                "requests and return evidence) and second-opinion review of risky diffs."
            ),
            "avoid_for": "Cheap mechanical edits (waste of capability).",
        },
    },
    "checks": {},
    "research": {"provider": "grep.ai"},
    "routing": {
        "mechanical": ["sonnet"],
        "feature": ["sonnet"],
        "refactor_wide": ["sonnet"],
        "gnarly": ["opus"],
        "review": ["opus"],
        "smoke_test": ["opus"],
        "escalation": ["sonnet", "opus"],
    },
}


def utc_stamp() -> str:
    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")


def git_root(start: Path | None = None) -> Path:
    p = subprocess.run(
        ["git", "rev-parse", "--show-toplevel"],
        capture_output=True, text=True, cwd=start or Path.cwd(),
    )
    if p.returncode == 0:
        return Path(p.stdout.strip())
    return Path.cwd()


# ---------------------------------------------------------------------------
# Config loading / merging / validation
# ---------------------------------------------------------------------------

def config_paths(root: Path) -> list[Path]:
    """Candidate config files, lowest precedence first."""
    paths: list[Path] = []
    user_global = Path.home() / ".config" / "parable" / "parable.toml"
    paths.append(user_global)
    paths.append(root / "parable.toml")
    paths.append(root / ".claude" / "parable.toml")
    env = os.environ.get("PARABLE_CONFIG")
    if env:
        paths.append(Path(env))
    return paths


def merge_configs(base: dict, overlay: dict) -> dict:
    """Merge overlay onto base. [executors.*]/[providers.*]/[checks.*] merge
    per-id with overlay winning per-field; [parable]/[routing]/[claude] are
    whole-table overlay-wins."""
    out = json.loads(json.dumps(base))  # deep copy, plain data only
    for section in ("executors", "providers", "checks"):
        for key, val in overlay.get(section, {}).items():
            merged = dict(out.setdefault(section, {}).get(key, {}))
            merged.update(val)
            out[section][key] = merged
    for section in ("parable", "routing", "research", "claude"):
        if section in overlay:
            merged = dict(out.get(section, {}))
            merged.update(overlay[section])
            out[section] = merged
    return out


def load_config(root: Path) -> tuple[dict, list[Path]]:
    cfg = json.loads(json.dumps(BUILTIN_DEFAULTS))
    loaded: list[Path] = []
    for path in config_paths(root):
        if path.is_file():
            with open(path, "rb") as f:
                try:
                    overlay = tomllib.load(f)
                except tomllib.TOMLDecodeError as e:
                    sys.exit(f"parable: invalid TOML in {path}: {e}")
            cfg = merge_configs(cfg, overlay)
            loaded.append(path)
    version = cfg.get("parable", {}).get("version", SCHEMA_VERSION)
    if int(version) != SCHEMA_VERSION:
        sys.exit(f"parable: config schema version {version} not supported (this version supports {SCHEMA_VERSION})")
    return cfg, loaded


def validate_config(cfg: dict) -> list[str]:
    """Return a list of problems (empty = valid)."""
    problems: list[str] = []
    providers = cfg.get("providers", {})
    executors = cfg.get("executors", {})
    for pid, prov in providers.items():
        ptype = prov.get("type")
        if ptype not in SUPPORTED_PROVIDER_TYPES:
            problems.append(
                f"provider '{pid}': unknown type '{ptype}' (this version supports: "
                f"{', '.join(SUPPORTED_PROVIDER_TYPES)})"
            )
        if ptype == "codex":
            if not prov.get("base_url"):
                problems.append(f"provider '{pid}': type=codex requires base_url")
            if not prov.get("env_key"):
                problems.append(f"provider '{pid}': type=codex requires env_key (name of the API-key env var)")
            wire = prov.get("wire_api", "responses")
            if wire != "responses":
                problems.append(
                    f"provider '{pid}': wire_api='{wire}' — codex only supports 'responses' "
                    f"(chat-completions was removed from codex; use a Responses-capable endpoint, "
                    f"a LiteLLM proxy bridge, or a type=\"pi\" provider)"
                )
        if ptype == "pi":
            if not prov.get("base_url"):
                problems.append(f"provider '{pid}': type=pi requires base_url")
            if not prov.get("env_key"):
                problems.append(f"provider '{pid}': type=pi requires env_key (name of the API-key env var)")
            api = prov.get("api", "openai-completions")
            if api not in PI_API_VALUES:
                problems.append(f"provider '{pid}': api='{api}' (supported: {', '.join(PI_API_VALUES)})")
        if ptype == "cursor":
            # cursor-agent authenticates via CURSOR_API_KEY by default; env_key overrides the name.
            if prov.get("base_url"):
                problems.append(f"provider '{pid}': type=cursor takes no base_url (cursor-agent owns its endpoint)")
    for eid, ex in executors.items():
        pid = ex.get("provider")
        if pid not in providers:
            problems.append(f"executor '{eid}': unknown provider '{pid}'")
        if not ex.get("model"):
            problems.append(f"executor '{eid}': missing model")
        effort = ex.get("effort")
        if effort is not None:
            ptype = providers.get(pid, {}).get("type")
            # cursor pins effort inside the model slug (e.g. grok-4.5-high) and some
            # first-party models (composer) have no effort variant — so effort is
            # advisory metadata there, not gated against the codex/pi enum.
            if ptype == "cursor":
                allowed = None
            elif ptype == "pi":
                allowed = PI_THINKING_LEVELS
            elif ptype == "subagent":
                allowed = CLAUDE_SUBAGENT_EFFORT_LEVELS
            else:
                allowed = EFFORT_LEVELS
            if allowed is not None and effort not in allowed:
                problems.append(f"executor '{eid}': effort='{effort}' (allowed for {ptype}: {', '.join(allowed)})")
        window = ex.get("context_ktok")
        if window is not None and (not isinstance(window, int) or isinstance(window, bool) or window <= 0):
            problems.append(f"executor '{eid}': context_ktok must be a positive integer (thousands of tokens)")
    for klass, chain in cfg.get("routing", {}).items():
        if klass == "notes" or not isinstance(chain, list):
            continue
        for eid in chain:
            if eid not in executors:
                problems.append(f"routing.{klass}: unknown executor '{eid}'")
    for cid, check in cfg.get("checks", {}).items():
        if not check.get("run"):
            problems.append(f"check '{cid}': missing run command")
        for w in check.get("when", []):
            if w not in CHECK_WHEN_VALUES:
                problems.append(f"check '{cid}': unknown when '{w}' (use {'/'.join(CHECK_WHEN_VALUES)})")
    for name in (cfg["parable"].get("default_executor"), cfg["parable"].get("default_reviewer")):
        if name and name not in executors:
            problems.append(f"[parable] default executor/reviewer '{name}' is not a configured executor")
    research = cfg.get("research", {}).get("provider", "grep.ai")
    if research not in ("grep.ai", "claude"):
        problems.append(f"[research] provider='{research}' (supported: grep.ai, claude)")
    claude = cfg.get("claude")
    if claude is not None:
        if not isinstance(claude, dict):
            problems.append("[claude] must be a table")
        else:
            unknown = sorted(set(claude) - CLAUDE_CONFIG_FIELDS)
            if unknown:
                problems.append(
                    f"[claude] unknown field(s): {', '.join(unknown)} "
                    f"(supported: {', '.join(sorted(CLAUDE_CONFIG_FIELDS))})"
                )
            base_url = claude.get("base_url")
            if not isinstance(base_url, str) or not base_url:
                problems.append("[claude] base_url is required")
            elif not is_loopback_url(base_url):
                problems.append(
                    "[claude] base_url must be an http(s) loopback URL "
                    "(localhost, 127.0.0.1, or ::1)"
                )
            env_name = claude.get("auth_token_env")
            if not isinstance(env_name, str) or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", env_name):
                problems.append("[claude] auth_token_env must name a valid environment variable")
            if not isinstance(claude.get("brain_model"), str) or not claude.get("brain_model"):
                problems.append("[claude] brain_model is required")
            binary = claude.get("binary", "claude")
            if not isinstance(binary, str) or not binary:
                problems.append("[claude] binary must be a non-empty command name or path")
    return problems


def is_loopback_url(value: str) -> bool:
    """Keep the subscription client token on the operator's machine."""
    try:
        parsed = urlparse(value)
        _ = parsed.port  # force malformed ports to raise
    except ValueError:
        return False
    return (
        parsed.scheme in ("http", "https")
        and parsed.hostname in ("localhost", "127.0.0.1", "::1")
        and parsed.username is None
        and parsed.password is None
        and not parsed.query
        and not parsed.fragment
    )


# ---------------------------------------------------------------------------
# Claude Code subscription launcher + project-local named agents
# ---------------------------------------------------------------------------

def custom_claude_executors(cfg: dict) -> dict[str, dict]:
    """Executors materialized as Claude Code custom agents.

    Bare Claude aliases are already native to Claude Code and need no generated
    file. Arbitrary model ids are preserved exactly in frontmatter.
    """
    result: dict[str, dict] = {}
    providers = cfg.get("providers", {})
    for executor_id, executor in cfg.get("executors", {}).items():
        if executor.get("enabled", True) is False:
            continue
        provider = providers.get(executor.get("provider"), {})
        if provider.get("type") != "subagent":
            continue
        model = executor.get("model")
        if isinstance(model, str) and model.lower() not in CLAUDE_AGENT_ALIASES:
            result[executor_id] = executor
    return result


def agent_slug(executor_id: str) -> str:
    slug = re.sub(r"[^a-z0-9-]+", "-", executor_id.lower()).strip("-")
    if not slug:
        raise ValueError(f"executor id {executor_id!r} cannot form a Claude agent name")
    return PARABLE_AGENT_PREFIX + slug


def render_claude_agent(executor_id: str, executor: dict) -> str:
    """Render safe YAML-frontmatter Markdown accepted by stock Claude Code."""
    name = agent_slug(executor_id)
    model = executor["model"]
    effort = executor.get("effort", "high")
    use_for = str(executor.get("use_for") or f"Tasks routed to the {executor_id} executor.")
    avoid_for = str(executor.get("avoid_for") or "").strip()
    description = use_for.replace("\n", " ").strip()
    body = [
        PARABLE_AGENT_MARKER,
        "",
        f"You are Parable's `{executor_id}` executor. Complete the delegated task in the "
        "working repository and return concise, checkable evidence to the parent.",
    ]
    if use_for:
        body += ["", f"Routing guidance: {use_for.strip()}"]
    if avoid_for:
        body += ["", f"Do not use this lane for: {avoid_for}"]
    return (
        "---\n"
        f"name: {name}\n"
        f"description: {json.dumps(description)}\n"
        f"model: {json.dumps(model)}\n"
        f"effort: {json.dumps(effort)}\n"
        "---\n"
        + "\n".join(body)
        + "\n"
    )


def is_parable_managed_agent(path: Path) -> bool:
    if path.is_symlink() or not path.is_file():
        return False
    try:
        return PARABLE_AGENT_MARKER in path.read_text(errors="replace")[:4096]
    except OSError:
        return False


def sync_claude_agents(root: Path, cfg: dict) -> dict[str, list[str]]:
    """Synchronize only Parable's project-local namespace.

    Unrelated user agents are never read beyond the narrow `parable-*.md`
    ownership check and are never modified.
    """
    executors = custom_claude_executors(cfg)
    names: dict[str, str] = {}
    for executor_id in executors:
        name = agent_slug(executor_id)
        prior = names.get(name)
        if prior is not None:
            raise ValueError(
                f"executors {prior!r} and {executor_id!r} both map to Claude agent {name!r}"
            )
        names[name] = executor_id

    agents_dir = root / ".claude" / "agents"
    agents_dir.mkdir(parents=True, exist_ok=True)
    changed: list[str] = []
    unchanged: list[str] = []
    expected: set[Path] = set()
    for name, executor_id in sorted(names.items()):
        path = agents_dir / f"{name}.md"
        expected.add(path)
        if path.is_symlink():
            raise ValueError(f"refusing to replace symlinked managed-agent path: {path}")
        content = render_claude_agent(executor_id, executors[executor_id])
        if path.is_file() and path.read_text() == content:
            unchanged.append(name)
            continue
        if path.exists() and not is_parable_managed_agent(path):
            raise ValueError(
                f"refusing to overwrite non-Parable agent at namespaced path: {path}"
            )
        path.write_text(content)
        path.chmod(0o644)
        changed.append(name)

    removed: list[str] = []
    for path in sorted(agents_dir.glob(f"{PARABLE_AGENT_PREFIX}*.md")):
        if path not in expected and is_parable_managed_agent(path):
            path.unlink()
            removed.append(path.stem)
    return {"changed": changed, "unchanged": unchanged, "removed": removed}


def claude_configured_models(cfg: dict) -> list[str]:
    """Exact proxy ids that may participate in a configured Claude session."""
    models = [cfg["claude"]["brain_model"]]
    models.extend(ex["model"] for ex in custom_claude_executors(cfg).values())
    return list(dict.fromkeys(models))


def claude_cast_availability(cfg: dict, available: set[str],
                             brain_model: str | None = None
                             ) -> dict[str, list[dict[str, str]]]:
    """Classify generated exact-model agents for this launch snapshot."""
    active: list[dict[str, str]] = []
    unavailable: list[dict[str, str]] = []
    parent: list[dict[str, str]] = []
    for executor_id, executor in sorted(custom_claude_executors(cfg).items()):
        model = executor["model"]
        item = {"name": agent_slug(executor_id), "model": model}
        if model == brain_model:
            parent.append(item)
            continue
        (active if model in available else unavailable).append(item)
    return {"active": active, "unavailable": unavailable, "parent": parent}


def proxy_models_endpoint(base_url: str) -> str:
    return base_url.rstrip("/") + (
        "/models" if urlparse(base_url).path.rstrip("/").endswith("/v1") else "/v1/models"
    )


def fetch_proxy_models(base_url: str, token: str, timeout: float = 5.0) -> set[str]:
    request = Request(
        proxy_models_endpoint(base_url),
        headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            payload = json.load(response)
    except HTTPError as exc:
        raise RuntimeError(f"proxy model check returned HTTP {exc.code}") from exc
    except URLError as exc:
        raise RuntimeError(f"proxy model check failed: {exc.reason}") from exc
    except (OSError, json.JSONDecodeError) as exc:
        raise RuntimeError(f"proxy model check returned an invalid response: {exc}") from exc
    data = payload.get("data") if isinstance(payload, dict) else None
    if not isinstance(data, list):
        raise RuntimeError("proxy model check response is missing a data array")
    return {
        item["id"] for item in data
        if isinstance(item, dict) and isinstance(item.get("id"), str)
    }


def build_claude_launch(cfg: dict, forwarded: list[str], environ: dict[str, str] | None = None,
                        *, solo: bool = False, available: set[str] | None = None
                        ) -> tuple[list[str], dict[str, str]]:
    claude = cfg["claude"]
    # Separator ownership: parse_claude_launch_args consumes Parable's `--`
    # separators; every token here — including any `--` — belongs to Claude.
    # Claude's own terminator ends option scanning: everything after it is
    # literal prompt text and must pass through untouched, even option-shaped.
    args = list(forwarded)
    scan = args[:args.index("--")] if "--" in args else args
    option_names = {arg.split("=", 1)[0] for arg in scan if arg.startswith("-")}
    if "--model" in option_names:
        raise ValueError(
            "Parable owns the Claude parent model; remove --model and use --brain or --solo"
        )
    if solo:
        conflicts = sorted(option_names & {
            "--agent", "--agents",
            "--allowedTools", "--allowed-tools",
            "--disallowedTools", "--disallowed-tools",
            "--fallback-model",
        })
        if conflicts:
            raise ValueError(
                "solo mode owns model selection and agent isolation; remove Claude option(s): "
                + ", ".join(conflicts)
            )
    source_env = os.environ if environ is None else environ
    token_name = claude["auth_token_env"]
    token = source_env.get(token_name)
    if not token:
        raise ValueError(f"{token_name} is not set")
    launch_env = dict(source_env)
    launch_env["ANTHROPIC_BASE_URL"] = claude["base_url"]
    launch_env["ANTHROPIC_AUTH_TOKEN"] = token
    if token_name != "ANTHROPIC_AUTH_TOKEN":
        launch_env.pop(token_name, None)
    nested_parable = bool(
        source_env.get(PARABLE_AGENT_STATE_ENV)
        or source_env.get(PARABLE_WELCOME_ENV)
    )
    for inherited in (
        "ANTHROPIC_API_KEY",
        "CLAUDE_CODE_OAUTH_TOKEN",
        "CLAUDE_CODE_SUBAGENT_MODEL",
        PARABLE_WELCOME_ENV,
        PARABLE_AGENT_STATE_ENV,
    ):
        launch_env.pop(inherited, None)
    if nested_parable:
        # A new Parable launched from an existing Parable/Claude shell inherits
        # the old session's process-scoped context controls. They are launch
        # state, not a fresh user override; recompute them for the new brain.
        for inherited in (
            CLAUDE_CONTEXT_ENV,
            CLAUDE_AUTO_COMPACT_WINDOW_ENV,
            CLAUDE_AUTO_COMPACT_PCT_ENV,
        ):
            launch_env.pop(inherited, None)
    if solo:
        launch_env.pop("CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS", None)
    # MAX_CONTEXT teaches Claude Code the real ceiling of proxied non-Claude
    # models. AUTO_COMPACT_WINDOW and its percentage are process-wide, though:
    # setting them from a mixed cast also caps a Claude-family parent. Keep
    # those two controls scoped to a non-Claude parent. User values always win.
    ceiling = claude_context_ceiling(
        cfg, claude["brain_model"], solo=solo, available=available
    )
    if nested_parable or not source_env.get(CLAUDE_CONTEXT_ENV):
        if ceiling is not None:
            launch_env[CLAUDE_CONTEXT_ENV] = str(ceiling)
    effective_ceiling = launch_env.get(CLAUDE_CONTEXT_ENV)
    parent_is_claude = is_claude_family_model(claude["brain_model"])
    if (
        effective_ceiling
        and not parent_is_claude
        and (nested_parable or not source_env.get(CLAUDE_AUTO_COMPACT_WINDOW_ENV))
    ):
        launch_env[CLAUDE_AUTO_COMPACT_WINDOW_ENV] = effective_ceiling
    if (
        effective_ceiling
        and not parent_is_claude
        and (nested_parable or not source_env.get(CLAUDE_AUTO_COMPACT_PCT_ENV))
    ):
        launch_env[CLAUDE_AUTO_COMPACT_PCT_ENV] = str(
            MODEL_AUTO_COMPACT_PCT.get(
                claude["brain_model"], CLAUDE_AUTO_COMPACT_PCT
            )
        )
    isolation = ["--disallowedTools", "Agent"] if solo else []
    argv = [
        claude.get("binary", "claude"),
        "--model", claude_cli_model(cfg, claude["brain_model"]),
        *isolation,
        *args,
    ]
    return argv, launch_env


def parse_claude_launch_args(forwarded: list[str]) -> tuple[str, str | None, list[str]]:
    """Consume Parable's mutually-exclusive --brain/--solo launch options."""
    args = list(forwarded)
    if args and args[0] == "--":
        args.pop(0)
    mode = "config"
    brain_explicit = False
    solo: str | None = None
    if args and args[0].startswith("--solo="):
        solo = args.pop(0).split("=", 1)[1]
        if not solo:
            raise ValueError("--solo requires a configured alias or exact catalog model")
    elif args and args[0] == "--solo":
        args.pop(0)
        if not args or args[0] == "--" or args[0].startswith("-"):
            raise ValueError("--solo requires a configured alias or exact catalog model")
        solo = args.pop(0)
    elif args and args[0].startswith("--brain="):
        brain_explicit = True
        mode = args.pop(0).split("=", 1)[1]
    elif args and args[0] == "--brain":
        brain_explicit = True
        args.pop(0)
        if not args or args[0] == "--":
            raise ValueError("--brain requires auto, fable, sol, grok, or config")
        mode = args.pop(0)
    if mode not in CLAUDE_BRAIN_MODES:
        raise ValueError(
            f"--brain {mode!r} is invalid (use {', '.join(CLAUDE_BRAIN_MODES)})"
        )
    # A separator here closes Parable's option region only when a Parable
    # option was actually consumed; otherwise it is Claude's own terminator
    # and must be forwarded untouched.
    if (solo is not None or brain_explicit) and args and args[0] == "--":
        args.pop(0)
    # Scan only Claude's option region: a later standalone `--` is Claude's own
    # terminator, and option-shaped prompt text after it is literal, not misplaced.
    scan = args[:args.index("--")] if "--" in args else args
    misplaced_brain = any(arg == "--brain" or arg.startswith("--brain=") for arg in scan)
    misplaced_solo = any(arg == "--solo" or arg.startswith("--solo=") for arg in scan)
    if misplaced_brain or misplaced_solo:
        if (
            solo is not None and misplaced_brain
            or brain_explicit and misplaced_solo
            or misplaced_brain and misplaced_solo
        ):
            raise ValueError("--brain and --solo are mutually exclusive Parable launch options")
        option = "--brain" if misplaced_brain else "--solo"
        raise ValueError(f"place Parable's {option} option before the `--` Claude argument separator")
    return mode, solo, args


def parse_claude_brain_args(forwarded: list[str]) -> tuple[str, list[str]]:
    """Backward-compatible parser for callers that only support brain mode."""
    mode, solo, args = parse_claude_launch_args(forwarded)
    if solo is not None:
        raise ValueError("--solo is not a brain mode")
    return mode, args


def claude_resume_selector(forwarded: list[str]) -> tuple[int, int, list[str]] | None:
    """Locate a CLI resume selector before Claude's argument terminator.

    The returned slice can be replaced with an exact ``--resume <session-id>``
    after Claude resolves ``--continue``, a session name, or a PR selector.
    An empty selector list means the caller requested an interactive picker,
    whose eventual session id is not available to a pre-launch process.
    """
    limit = forwarded.index("--") if "--" in forwarded else len(forwarded)
    index = 0
    while index < limit:
        argument = forwarded[index]
        if argument in ("-c", "--continue"):
            return index, index + 1, ["--continue"]
        for option in ("--resume", "--from-pr"):
            prefix = option + "="
            if argument.startswith(prefix):
                value = argument[len(prefix):]
                return index, index + 1, [option, value] if value else []
        if argument in ("-r", "--resume", "--from-pr"):
            if index + 1 < limit and not forwarded[index + 1].startswith("-"):
                option = "--resume" if argument in ("-r", "--resume") else argument
                return index, index + 2, [option, forwarded[index + 1]]
            return index, index + 1, []
        index += 1
    return None


def _claude_result(stdout: str) -> dict:
    """Read Claude's final structured result while tolerating startup notices."""
    for line in reversed(stdout.splitlines()):
        try:
            payload = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(payload, dict) and payload.get("type") == "result":
            return payload
    raise RuntimeError("Claude resume preflight returned no structured result")


def _context_token_count(report: str) -> int:
    match = re.search(
        r"\*\*Tokens:\*\*\s*([0-9][0-9,.]*)\s*([kKmM]?)\s*/",
        report,
    )
    if not match:
        raise RuntimeError("Claude resume preflight could not read /context usage")
    value = float(match.group(1).replace(",", ""))
    scale = {"": 1, "k": 1_000, "m": 1_000_000}[match.group(2).lower()]
    return round(value * scale)


def prepare_claude_resume(
    forwarded: list[str],
    binary: str,
    launch_env: dict[str, str],
    available: set[str],
    *,
    run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
    report: Callable[[str], None] | None = None,
    target_ceiling: int | None = None,
) -> tuple[list[str], str | None]:
    """Compact an oversized resumed session before a smaller model loads it.

    Claude Code's SessionStart hook cannot block startup or invoke built-in
    commands. For an explicit CLI resume, use Sonnet 5's 1M window to query
    ``/context`` without model tokens, compact only above the target launch's
    safe threshold, then replace the selector with the resolved session id.
    """
    located = claude_resume_selector(forwarded)
    if located is None:
        return forwarded, None
    start, end, selector = located
    if not selector:
        if launch_env.get("PARABLE_CONTEXT_RESUME_PICKER") == "1":
            return forwarded, "picker selected; exact session will be checked after selection"
        return forwarded, (
            "picker selected; pre-compaction requires --continue or "
            "--resume <name-or-session-id>"
        )
    scan = forwarded[:forwarded.index("--")] if "--" in forwarded else forwarded
    if "--fork-session" in scan:
        return forwarded, "forked resume selected; pre-compaction skipped"
    if target_ceiling is None:
        raw_ceiling = launch_env.get(CLAUDE_CONTEXT_ENV)
        try:
            target_ceiling = int(raw_ceiling) if raw_ceiling else None
        except ValueError as exc:
            raise RuntimeError(f"invalid {CLAUDE_CONTEXT_ENV}={raw_ceiling!r}") from exc
    force_compact = launch_env.get(PARABLE_CONTEXT_FAILURE_RECOVERY_ENV) == "1"
    if target_ceiling is None or (
        target_ceiling >= CLAUDE_RESUME_COMPACT_WINDOW and not force_compact
    ):
        return forwarded, None
    if CLAUDE_RESUME_COMPACT_BASE_MODEL not in available:
        raise RuntimeError(
            f"resume needs {CLAUDE_RESUME_COMPACT_BASE_MODEL} to check a session before "
            f"loading it into a {target_ceiling:,}-token window"
        )

    preflight_env = dict(launch_env)
    for name in (
        CLAUDE_CONTEXT_ENV,
        CLAUDE_AUTO_COMPACT_PCT_ENV,
        CLAUDE_AUTO_COMPACT_WINDOW_ENV,
        PARABLE_CONTEXT_FAILURE_RECOVERY_ENV,
    ):
        preflight_env.pop(name, None)
    common = [
        binary,
        "--bare",
        "--model", CLAUDE_RESUME_COMPACT_MODEL,
        "--effort", "low",
        "--print",
        "--output-format", "json",
    ]

    def invoke(
        command: list[str],
        *,
        operation: str = "preflight",
        timeout: int = CLAUDE_RESUME_CHECK_TIMEOUT_SECONDS,
        heartbeat: str | None = None,
    ) -> dict:
        heartbeat_stop = threading.Event()
        heartbeat_thread = None
        if heartbeat and report:
            def report_heartbeat() -> None:
                elapsed = CLAUDE_RESUME_HEARTBEAT_SECONDS
                while not heartbeat_stop.wait(CLAUDE_RESUME_HEARTBEAT_SECONDS):
                    report(f"{heartbeat} ({elapsed}s elapsed)")
                    elapsed += CLAUDE_RESUME_HEARTBEAT_SECONDS

            heartbeat_thread = threading.Thread(
                target=report_heartbeat,
                name="parable-resume-heartbeat",
                daemon=True,
            )
            heartbeat_thread.start()
        try:
            completed = run(
                command,
                env=preflight_env,
                capture_output=True,
                text=True,
                timeout=timeout,
            )
        except subprocess.TimeoutExpired as exc:
            minutes = timeout // 60
            raise RuntimeError(
                f"Claude resume {operation} timed out after {minutes} minutes"
            ) from exc
        finally:
            heartbeat_stop.set()
            if heartbeat_thread:
                heartbeat_thread.join()
        if completed.returncode != 0:
            detail = (completed.stderr or completed.stdout or "unknown error").strip()
            raise RuntimeError(f"Claude resume preflight failed: {detail[:300]}")
        result = _claude_result(completed.stdout)
        if result.get("is_error"):
            detail = str(result.get("result") or result.get("api_error_status") or "unknown error")
            raise RuntimeError(f"Claude resume preflight failed: {detail[:300]}")
        return result

    if report:
        report("checking resumed session context with Sonnet 5")
    context = invoke([*common, *selector, "/context"])
    session_id = context.get("session_id")
    if not isinstance(session_id, str) or not session_id:
        raise RuntimeError("Claude resume preflight did not resolve a session id")
    exact = [*forwarded[:start], "--resume", session_id, *forwarded[end:]]
    used_tokens = _context_token_count(str(context.get("result") or ""))
    safe_tokens = target_ceiling * CLAUDE_AUTO_COMPACT_PCT // 100
    if used_tokens < safe_tokens and not force_compact:
        return exact, f"{used_tokens:,} tokens fit the {safe_tokens:,}-token safe start"

    if report:
        report(
            f"compacting {used_tokens:,} tokens with Sonnet 5; "
            "this can take several minutes"
        )
    compact = invoke(
        [*common, "--resume", session_id, CLAUDE_RESUME_COMPACT_PROMPT],
        operation="compaction",
        timeout=CLAUDE_RESUME_COMPACT_TIMEOUT_SECONDS,
        heartbeat="still compacting with Sonnet 5",
    )
    compact_result = str(compact.get("result") or "").strip()
    if report:
        report("compaction finished; verifying the reduced context")
    after = invoke([*common, "--resume", session_id, "/context"])
    remaining_tokens = _context_token_count(str(after.get("result") or ""))
    if remaining_tokens >= safe_tokens:
        detail = compact_result or "session remains above the safe start threshold"
        raise RuntimeError(
            "Claude resume compaction did not reduce the session below "
            f"{safe_tokens:,} tokens: {detail[:300]}"
        )
    return exact, (
        f"compacted {used_tokens:,} to {remaining_tokens:,} tokens with Sonnet 5 before the "
        f"{target_ceiling:,}-token launch"
    )


def _usage_percent(reports: list[dict], pool: str) -> float | None:
    report = next((item for item in reports if item.get("pool") == pool), None)
    if not report or report.get("status") != "ok":
        return None
    try:
        import parable_usage
    except ImportError:
        sys.path.insert(0, str(Path(__file__).resolve().parent))
        import parable_usage  # noqa: E402
    return parable_usage.worst_used_pct(report)


def _brain_model_state(model: str, configured: set[str], available: set[str]) -> str:
    if model not in configured:
        return "not configured"
    if model not in available:
        return "unavailable"
    return "eligible"


def resolve_claude_brain(cfg: dict, mode: str, available: set[str],
                         reports: list[dict] | None = None) -> tuple[str, str]:
    """Resolve an explicit or Fable-first automatic parent from configured models."""
    configured = set(claude_configured_models(cfg))
    if mode == "config":
        model = cfg["claude"]["brain_model"]
        if model not in available:
            raise ValueError(
                f"configured parent model {model!r} is unavailable in the proxy catalog"
            )
        return model, "configured parent"
    if mode in CLAUDE_BRAIN_MODELS:
        model = CLAUDE_BRAIN_MODELS[mode]
        if model not in configured:
            raise ValueError(
                f"--brain {mode} requires configured model {model!r}; "
                "rerun setup with its subscription selected"
            )
        if model not in available:
            raise ValueError(
                f"--brain {mode} model {model!r} is unavailable in the proxy catalog"
            )
        return model, f"explicit {mode} parent"

    fable = CLAUDE_BRAIN_MODELS["fable"]
    sol = CLAUDE_BRAIN_MODELS["sol"]
    grok = CLAUDE_BRAIN_MODELS["grok"]
    fable_state = _brain_model_state(fable, configured, available)
    sol_state = _brain_model_state(sol, configured, available)
    grok_state = _brain_model_state(grok, configured, available)

    if fable_state != "eligible":
        if sol_state == "eligible":
            return sol, f"Fable is {fable_state}; using Sol"
        if grok_state == "eligible":
            return grok, (
                f"Fable is {fable_state}; Sol is {sol_state}; using the configured Grok fallback; "
                "xAI usage telemetry is unavailable"
            )
        raise ValueError(
            "automatic brain selection found no available configured Fable, Sol, or Grok"
        )

    if sol_state != "eligible" and grok_state != "eligible":
        return fable, f"Sol is {sol_state}; Grok is {grok_state}; using Fable"

    live_probe = reports is None
    if live_probe:
        try:
            import parable_usage
        except ImportError:
            sys.path.insert(0, str(Path(__file__).resolve().parent))
            import parable_usage  # noqa: E402
        reports = parable_usage.probe_all(["claude"])
    fable_used = _usage_percent(reports, "claude")
    if fable_used is None:
        return fable, "Claude usage is unknown; keeping the preferred Fable parent"
    if fable_used < AUTO_BRAIN_TIGHT_PCT:
        return fable, f"Claude usage is {fable_used:.0f}%; keeping the preferred Fable parent"

    if sol_state != "eligible":
        if grok_state == "eligible":
            return grok, (
                f"Claude usage is tight at {fable_used:.0f}%; Sol is {sol_state}; "
                "using the configured Grok fallback; xAI usage telemetry is unavailable"
            )
        return fable, (
            f"Claude usage is tight at {fable_used:.0f}%; Sol is {sol_state} and Grok is "
            f"{grok_state}; keeping Fable"
        )

    if live_probe and not any(item.get("pool") == "codex" for item in reports):
        reports = [*reports, *parable_usage.probe_all(["codex"])]
    sol_used = _usage_percent(reports, "codex")
    if sol_used is None:
        return sol, f"Claude usage is tight at {fable_used:.0f}%; Sol pool is unknown"
    if sol_used < AUTO_BRAIN_TIGHT_PCT:
        return sol, f"Claude usage is tight at {fable_used:.0f}%; Sol pool is {sol_used:.0f}%"
    if grok_state == "eligible":
        return grok, (
            f"Claude and Sol pools are tight at {fable_used:.0f}% and {sol_used:.0f}%; "
            "using the configured Grok fallback; xAI usage telemetry is unavailable"
        )
    return fable, (
        f"Claude and Sol pools are tight at {fable_used:.0f}% and {sol_used:.0f}%; Grok is "
        f"{grok_state}; keeping Fable"
    )


def _solo_alias_key(value: str) -> str:
    return re.sub(r"[\s_]+", "-", value.strip().lower())


def solo_model_aliases(cfg: dict) -> dict[str, set[str]]:
    """Map friendly configured executor names to their exact proxy model ids."""
    aliases: dict[str, set[str]] = {}

    def add(alias: str, model: str) -> None:
        key = _solo_alias_key(alias)
        if key:
            aliases.setdefault(key, set()).add(model)

    for executor_id, executor in custom_claude_executors(cfg).items():
        model = executor["model"]
        add(executor_id, model)
        add(re.sub(r"_exact$", "", executor_id), model)
        add(_welcome_label(executor_id, model), model)
        add(model, model)
    for alias, model in CLAUDE_BRAIN_MODELS.items():
        add(alias, model)
    return aliases


def resolve_solo_model(cfg: dict, selector: str, available: set[str]) -> tuple[str, str]:
    """Resolve a configured friendly alias or exact proxy catalog id for solo mode."""
    if selector in available:
        return selector, f"exact catalog model {selector}"
    key = _solo_alias_key(selector)
    candidates = solo_model_aliases(cfg).get(key, set())
    if not candidates:
        known = ", ".join(sorted(solo_model_aliases(cfg)))
        raise ValueError(f"--solo {selector!r} is unknown (configured aliases: {known})")
    if len(candidates) != 1:
        models = ", ".join(sorted(candidates))
        raise ValueError(f"--solo {selector!r} is ambiguous ({models})")
    model = next(iter(candidates))
    if model not in available:
        raise ValueError(f"proxy model catalog is missing: {model}")
    return model, f"configured solo alias {selector}"


def model_context_window(cfg: dict, model: str) -> int | None:
    """Best-known context window for a proxied model, or None when unknown.

    A `context_ktok` on any enabled executor pinned to that exact model wins
    over the built-in table, so users can correct or extend it in parable.toml.
    """
    for executor in custom_claude_executors(cfg).values():
        if executor.get("model") == model:
            override = executor.get("context_ktok")
            if isinstance(override, int) and not isinstance(override, bool) and override > 0:
                return override * 1000
    return MODEL_CONTEXT_WINDOWS.get(model)


def is_claude_family_model(model: str) -> bool:
    """Whether Claude Code applies its native Claude-family window rules."""
    return model.lower().startswith("claude-")


def claude_cli_model(cfg: dict, model: str) -> str:
    """Return the model selector that makes Claude Code honor the real window.

    The proxy catalog and Parable configuration retain the exact bare model id.
    ``[1m]`` is Claude Code's own context selector and is stripped before the
    request reaches the provider. Without it, even Fable is treated as 200k.
    """
    if model.endswith(CLAUDE_LONG_CONTEXT_MARKER):
        return model
    window = model_context_window(cfg, model)
    if is_claude_family_model(model) and window == 1_000_000:
        return model + CLAUDE_LONG_CONTEXT_MARKER
    return model


def claude_context_ceiling(cfg: dict, brain_model: str, *, solo: bool = False,
                           available: set[str] | None = None) -> int | None:
    """The CLAUDE_CODE_MAX_CONTEXT_TOKENS value for a launch, or None to not set it.

    Claude Code honors this env var only for models whose id does not start
    with "claude-", and it is process-wide — one value covers the parent and
    every subagent. A non-Claude parent therefore gets its own window; taking
    the minimum across smaller cast models would silently shrink the active
    brain. A Claude-family parent uses its native window rules, while this env
    remains the minimum across available non-Claude cast models. Unknown windows
    use Claude Code's own 200k fallback so they never raise the assumed ceiling.
    """
    if solo:
        if is_claude_family_model(brain_model):
            return None
        return model_context_window(cfg, brain_model)
    if not is_claude_family_model(brain_model):
        return model_context_window(cfg, brain_model) or CLAUDE_DEFAULT_CONTEXT_WINDOW
    non_claude = [
        ex["model"] for ex in custom_claude_executors(cfg).values()
        if not is_claude_family_model(ex["model"])
        and (available is None or ex["model"] in available)
    ]
    if not non_claude:
        return None
    return min(
        model_context_window(cfg, model) or CLAUDE_DEFAULT_CONTEXT_WINDOW
        for model in non_claude
    )


def config_with_claude_brain(cfg: dict, model: str) -> dict:
    return {**cfg, "claude": {**cfg["claude"], "brain_model": model}}


def _welcome_label(executor_id: str, model: str) -> str:
    name = re.sub(r"_exact$", "", executor_id).replace("_", "-")
    return (name or model).upper()


def _welcome_summary(value: object, width: int) -> str:
    text = re.sub(r"\s+", " ", str(value or "General delegated work.")).strip()
    return textwrap.shorten(text, width=max(width, 20), placeholder="…")


def _welcome_animal(label: str) -> str:
    return PARABLE_ANIMALS.get(label, "◆")


def claude_welcome_cast(cfg: dict, brain_model: str, available: set[str]
                        ) -> tuple[str, list[tuple[str, str, str]]]:
    """Return the parent label and every usable native executor for display."""
    brain_label = brain_model.upper()
    rows: list[tuple[str, str, str]] = []
    for executor_id, executor in cfg.get("executors", {}).items():
        if executor.get("enabled", True) is False:
            continue
        provider = cfg.get("providers", {}).get(executor.get("provider"), {})
        if provider.get("type") != "subagent":
            continue
        model = executor.get("model")
        if not isinstance(model, str):
            continue
        if model.lower() not in CLAUDE_AGENT_ALIASES and model not in available:
            continue
        label = _welcome_label(executor_id, model)
        if model == brain_model:
            brain_label = label
            continue
        rows.append((label, model, str(executor.get("use_for") or "General delegated work.")))
    return brain_label, rows


def _welcome_window(cfg: dict, model: str) -> str:
    window = model_context_window(cfg, model)
    if window is None:
        return ""
    if window >= 1_000_000:
        return f" · {window // 1_000_000}M ctx"
    return f" · {window // 1_000}k ctx"


def render_claude_welcome(cfg: dict, brain_model: str, decision: str,
                          available: set[str], columns: int | None = None) -> str:
    """Render the zero-token launch card shown by the SessionStart hook."""
    terminal_width = columns or shutil.get_terminal_size((96, 24)).columns
    width = max(54, min(terminal_width - 10, 110))
    brain_label, rows = claude_welcome_cast(cfg, brain_model, available)
    model_width = min(30, max((len(model) for _label, model, _use in rows), default=12))
    summary_width = max(20, width - 4 - 8 - 1 - model_width - 2)
    procession = "  ".join(
        [_welcome_animal(brain_label), *(_welcome_animal(label) for label, _model, _use in rows)]
    )
    lines = [*PARABLE_ASCII, f"  {procession}   →  the road ahead"]
    lines.append(
        f"  {_welcome_animal(brain_label)} BRAIN   {brain_label} · {brain_model}"
        f"{_welcome_window(cfg, brain_model)}"
    )
    lines.append(f"          {_welcome_summary(decision, width - 10)}")
    lines.append(f"  CAST    {len(rows)} routed models ready")
    for label, model, use_for in rows:
        lines.append(
            f"  {_welcome_animal(label)} {label[:7]:<7} {model[:model_width]:<{model_width}}  "
            f"{_welcome_summary(use_for, summary_width)}"
        )
    unavailable = claude_cast_availability(cfg, available, brain_model)["unavailable"]
    if unavailable:
        labels = ", ".join(
            _welcome_label(
                re.sub(
                    r"-exact$", "",
                    item["name"].removeprefix(PARABLE_AGENT_PREFIX),
                ),
                item["model"],
            )
            for item in unavailable
        )
        lines.append(f"  DEGRADED {labels} unavailable for this session")
    return "\n".join(lines)


def render_claude_solo_welcome(cfg: dict, model: str, decision: str) -> str:
    """Render the single-model launch contract with no cast or delegation cues."""
    label = model.upper()
    for executor_id, executor in custom_claude_executors(cfg).items():
        if executor.get("model") == model:
            label = _welcome_label(executor_id, model)
            break
    lines = [*PARABLE_ASCII]
    lines.append(f"  {_welcome_animal(label)} SOLO    {label} · {model}{_welcome_window(cfg, model)}")
    lines.append(f"          {decision}")
    lines.append("  SOLO CONTRACT")
    lines.append("          You are the only agent. Work directly from request to verified result.")
    lines.append("          Do not invoke Agent, subagents, agent teams, delegation, or cast routing.")
    lines.append("          You own planning, implementation, testing, review, and final judgment.")
    return "\n".join(lines)


def add_claude_welcome(argv: list[str], launch_env: dict[str, str], cfg: dict,
                       brain_model: str, decision: str, available: set[str],
                       forwarded: list[str], *, solo: bool = False
                       ) -> tuple[list[str], dict[str, str]]:
    """Inject Parable's session plugin and its card for interactive sessions."""
    plugin_skip_flags = {
        "-h", "--help", "-v", "--version", "--bare", "--init-only",
    }
    scan = forwarded[:forwarded.index("--")] if "--" in forwarded else forwarded
    if any(argument.split("=", 1)[0] in plugin_skip_flags for argument in scan):
        return argv, launch_env
    print_flags = {"-p", "--print"}
    is_print = any(argument.split("=", 1)[0] in print_flags for argument in scan)
    if solo and is_print:
        return argv, launch_env
    manifest = PARABLE_WELCOME_PLUGIN / ".claude-plugin" / "plugin.json"
    hook = PARABLE_WELCOME_PLUGIN / "hooks" / "hooks.json"
    scripts = [
        PARABLE_WELCOME_PLUGIN / "scripts" / name
        for name in ("welcome.py", "model_guard.py", "context_recovery.py")
    ]
    if not all(
        path.is_file() and not path.is_symlink()
        for path in (manifest, hook, *scripts)
    ):
        raise ValueError("Parable welcome plugin is missing or unsafe; reinstall Parable")
    env = dict(launch_env)
    availability = (
        {"active": [], "unavailable": [], "parent": []}
        if solo else claude_cast_availability(cfg, available, brain_model)
    )
    agent_state = {
        status: [item["name"] for item in availability[status]]
        for status in ("active", "unavailable", "parent")
    }
    env[PARABLE_AGENT_STATE_ENV] = json.dumps(agent_state, separators=(",", ":"))
    if not is_print:
        env[PARABLE_WELCOME_ENV] = (
            render_claude_solo_welcome(cfg, brain_model, decision)
            if solo else render_claude_welcome(cfg, brain_model, decision, available)
        )
    return [argv[0], "--plugin-dir", str(PARABLE_WELCOME_PLUGIN), *argv[1:]], env


def checked_claude_config(root: Path, require_token: bool = True
                          ) -> tuple[dict, list[Path], str | None]:
    cfg, loaded = load_config(root)
    problems = validate_config(cfg)
    if problems:
        raise ValueError("; ".join(problems))
    if "claude" not in cfg:
        raise ValueError(
            "no [claude] session configured; add base_url, auth_token_env, and brain_model"
        )
    token_name = cfg["claude"]["auth_token_env"]
    token = os.environ.get(token_name)
    if require_token and not token:
        raise ValueError(f"{token_name} is not set")
    return cfg, loaded, token


def cmd_agents_sync(_args: argparse.Namespace) -> int:
    root = git_root()
    try:
        cfg, _loaded, _token = checked_claude_config(root, require_token=False)
        result = sync_claude_agents(root, cfg)
    except (OSError, ValueError) as exc:
        print(f"parable: {exc}", file=sys.stderr)
        return 1
    print(
        "agents: "
        f"{len(result['changed'])} changed, "
        f"{len(result['unchanged'])} unchanged, "
        f"{len(result['removed'])} removed"
    )
    return 0


def exact_named_cast(cfg: dict) -> list[dict[str, str]]:
    return [
        {"name": agent_slug(executor_id), "model": executor["model"]}
        for executor_id, executor in sorted(custom_claude_executors(cfg).items())
    ]


def cmd_finalize(args: argparse.Namespace) -> int:
    root = git_root()
    try:
        cfg, _loaded, token = checked_claude_config(root)
        assert token is not None
        available = fetch_proxy_models(cfg["claude"]["base_url"], token)
        brain_model, _decision = resolve_claude_brain(
            cfg, "auto", available, reports=[]
        )
        availability = claude_cast_availability(cfg, available, brain_model)
        result = sync_claude_agents(root, cfg)
    except (OSError, RuntimeError, ValueError) as exc:
        print(f"parable: {exc}", file=sys.stderr)
        return 1
    cast = exact_named_cast(cfg)
    configured_parent_unavailable = cfg["claude"]["brain_model"] not in available
    report = {
        "ready": True,
        "degraded": configured_parent_unavailable or bool(availability["unavailable"]),
        "configuredParentModel": cfg["claude"]["brain_model"],
        "parentModel": brain_model,
        "agents": cast,
        "activeAgents": availability["active"],
        "unavailableAgents": availability["unavailable"],
        "catalog": {
            "availableCount": len(available),
            "configuredCount": len(claude_configured_models(cfg)),
        },
        "sync": {key: len(value) for key, value in result.items()},
        "next": "parable",
    }
    if args.json:
        print(json.dumps(report, indent=2))
        return 0
    print(
        f"catalog: {'degraded' if report['degraded'] else 'ready'} "
        f"({len(report['activeAgents'])} active agents, "
        f"{len(report['unavailableAgents'])} unavailable, "
        f"{report['catalog']['availableCount']} models available)"
    )
    print(f"parent:  {report['parentModel']}")
    for agent in cast:
        print(f"agent:   {agent['name']} -> {agent['model']}")
    for agent in availability["unavailable"]:
        print(f"offline: {agent['name']} -> {agent['model']}")
    print(f"next:    {report['next']}")
    return 0


def cmd_claude(args: argparse.Namespace) -> int:
    root = git_root()
    try:
        cfg, _loaded, token = checked_claude_config(root)
        assert token is not None
        brain_mode, solo_selector, forwarded = parse_claude_launch_args(args.claude_args)
        if solo_selector is not None:
            available = fetch_proxy_models(cfg["claude"]["base_url"], token)
            brain_model, decision = resolve_solo_model(
                cfg, solo_selector, available
            )
            launch_cfg = config_with_claude_brain(cfg, brain_model)
            argv, launch_env = build_claude_launch(
                launch_cfg, forwarded, solo=True, available=available
            )
            forwarded, resume_note = prepare_claude_resume(
                forwarded, argv[0], launch_env, available,
                report=lambda message: print(f"resume: {message}", flush=True),
                target_ceiling=model_context_window(cfg, brain_model),
            )
            argv, launch_env = build_claude_launch(
                launch_cfg, forwarded, solo=True, available=available
            )
            argv, launch_env = add_claude_welcome(
                argv, launch_env, cfg, brain_model, decision, available, forwarded,
                solo=True,
            )
            result = None
        else:
            available = fetch_proxy_models(cfg["claude"]["base_url"], token)
            brain_model, decision = resolve_claude_brain(cfg, brain_mode, available)
            availability = claude_cast_availability(cfg, available, brain_model)
            result = sync_claude_agents(root, cfg)
            launch_cfg = config_with_claude_brain(cfg, brain_model)
            argv, launch_env = build_claude_launch(
                launch_cfg, forwarded, available=available
            )
            forwarded, resume_note = prepare_claude_resume(
                forwarded, argv[0], launch_env, available,
                report=lambda message: print(f"resume: {message}", flush=True),
                target_ceiling=model_context_window(cfg, brain_model),
            )
            argv, launch_env = build_claude_launch(
                launch_cfg, forwarded, available=available
            )
            argv, launch_env = add_claude_welcome(
                argv, launch_env, cfg, brain_model, decision, available, forwarded
            )
    except (OSError, RuntimeError, ValueError) as exc:
        print(f"parable: {exc}", file=sys.stderr)
        return 1
    ceiling = launch_env.get(CLAUDE_CONTEXT_ENV)
    parent_window = model_context_window(cfg, brain_model)
    compact_pct = launch_env.get(CLAUDE_AUTO_COMPACT_PCT_ENV)
    context_note = f"; parent context {parent_window:,} tokens" if parent_window else ""
    if ceiling and (parent_window is None or int(ceiling) != parent_window):
        context_note += f"; non-Claude cast ceiling {int(ceiling):,} tokens"
    if compact_pct:
        context_note += f"; auto-compact {compact_pct}%"
    if solo_selector is not None:
        print(f"proxy: ready ({len(available)} models); solo agent isolation enabled", flush=True)
        print(f"solo: {brain_model} ({decision}){context_note}", flush=True)
    else:
        assert result is not None
        print(
            f"proxy: ready ({len(available)} models); "
            f"agents: {len(result['changed'])} changed, "
            f"{len(result['unchanged'])} unchanged, {len(result['removed'])} removed",
            flush=True,
        )
        print(f"brain: {brain_model} ({decision}){context_note}", flush=True)
        if availability["unavailable"]:
            print(
                "degraded: "
                + ", ".join(item["name"] for item in availability["unavailable"])
                + " unavailable for this session",
                flush=True,
            )
    if resume_note:
        print(f"resume: {resume_note}", flush=True)
    try:
        os.execvpe(argv[0], argv, launch_env)
    except FileNotFoundError:
        print(f"parable: {argv[0]!r} not found on PATH", file=sys.stderr)
        return 1


def env_key_status(cfg: dict, executor_id: str) -> str:
    """Credential status for an executor. Returns exactly one of:
    'disabled', 'subagent', 'codex-native', 'ready', or 'missing <ENV_VAR_NAME>'."""
    ex = cfg["executors"][executor_id]
    if ex.get("enabled", True) is False:
        return "disabled"
    prov = cfg["providers"].get(ex.get("provider"), {})
    ptype = prov.get("type")
    if ptype == "subagent":
        return "subagent"
    if ptype == "codex-native":
        return "codex-native"
    # cursor authenticates via CURSOR_API_KEY unless env_key overrides the name
    if ptype == "cursor":
        key = prov.get("env_key", "CURSOR_API_KEY")
        return "ready" if os.environ.get(key) else f"missing {key}"
    # codex and pi providers both authenticate via a named env var
    key = prov.get("env_key", "")
    return "ready" if os.environ.get(key) else f"missing {key}"


# ---------------------------------------------------------------------------
# codex argv construction
# ---------------------------------------------------------------------------

def provider_overrides(provider_id: str, prov: dict) -> list[str]:
    """Inline -c overrides defining a custom provider for one invocation.
    The user's ~/.codex/config.toml is never modified."""
    pid = f"parable_{provider_id}"
    args: list[str] = []
    args += ["-c", f'model_providers.{pid}.name="{provider_id}"']
    args += ["-c", f'model_providers.{pid}.base_url="{prov["base_url"]}"']
    args += ["-c", f'model_providers.{pid}.env_key="{prov["env_key"]}"']
    args += ["-c", f'model_providers.{pid}.wire_api="responses"']
    for hk, hv in prov.get("http_headers", {}).items():
        args += ["-c", f'model_providers.{pid}.http_headers.{hk}="{hv}"']
    for qk, qv in prov.get("query_params", {}).items():
        args += ["-c", f'model_providers.{pid}.query_params.{qk}="{qv}"']
    args += ["-c", f'model_provider="{pid}"']
    return args


def build_run_argv(cfg: dict, executor_id: str, workdir: Path,
                   last_msg_path: Path) -> tuple[list[str], list[str]]:
    """Returns (argv, overrides). overrides is the flag list a resume must
    replay, built explicitly alongside argv."""
    ex = cfg["executors"][executor_id]
    prov = cfg["providers"][ex["provider"]]
    overrides: list[str] = []
    if prov.get("type") == "codex":
        overrides += provider_overrides(ex["provider"], prov)
    overrides += ["-c", f'model="{ex["model"]}"']
    # Always pin effort explicitly — otherwise runs silently inherit whatever
    # the user's personal codex config sets.
    overrides += ["-c", f'model_reasoning_effort="{ex.get("effort", "high")}"']
    for item in ex.get("extra_config", []):
        overrides += ["-c", item]
    argv = (["codex", "exec", "--yolo", "--json", "-C", str(workdir), "--skip-git-repo-check"]
            + overrides + ["--output-last-message", str(last_msg_path), "-"])
    return argv, overrides


# ---------------------------------------------------------------------------
# pi harness — provider generation, argv, env
# ---------------------------------------------------------------------------

def build_pi_models_json(provider_id: str, prov: dict, ex: dict, executor_id: str) -> dict:
    """Generated per-run pi provider config. The API key is an $ENV reference —
    resolved by pi at request time, never written to disk or argv."""
    cost = ex.get("cost", {})
    model: dict = {
        "id": ex["model"],
        "name": executor_id,
        "reasoning": bool(ex.get("reasoning", True)),
        "input": ["text"],
        "cost": {
            "input": cost.get("in", 0),
            "output": cost.get("out", 0),
            "cacheRead": cost.get("cache_in", 0),
            "cacheWrite": 0,
        },
    }
    if ex.get("context_ktok"):
        model["contextWindow"] = int(ex["context_ktok"]) * 1000
    model.update(ex.get("model_overrides", {}))
    provider: dict = {
        "name": f"parable_{provider_id}",
        "baseUrl": prov["base_url"],
        "apiKey": "$" + prov["env_key"],
        "api": prov.get("api", "openai-completions"),
        "models": [model],
    }
    for key in ("headers", "compat"):
        if prov.get(key):
            provider[key] = prov[key]
    return {"providers": {f"parable_{provider_id}": provider}}


def write_pi_agent_dir(base_dir: Path, models: dict) -> Path:
    """Hermetic pi agent dir for one run: pi reads models/settings/sessions from
    here (via PI_CODING_AGENT_DIR) and never touches the user's ~/.pi."""
    agent_dir = base_dir / "pi-agent"
    agent_dir.mkdir(parents=True, exist_ok=True)
    (agent_dir / "models.json").write_text(json.dumps(models, indent=1))
    user_bin = Path.home() / ".pi" / "agent" / "bin"
    link = agent_dir / "bin"
    if user_bin.is_dir() and not link.exists():
        link.symlink_to(user_bin)
    return agent_dir


PI_HERMETIC_FLAGS = ["--no-extensions", "--no-skills", "--no-prompt-templates", "--no-approve"]


def pi_model_flags(ex: dict, default_effort: str = "high") -> list[str]:
    """Provider and model go in as separate flags: model ids can contain
    slashes (e.g. accounts/fireworks/models/...), which the combined
    provider/id form mis-parses."""
    return ["--provider", f'parable_{ex["provider"]}',
            "--model", ex["model"],
            "--thinking", ex.get("effort", default_effort)]


def build_pi_argv(cfg: dict, executor_id: str, run_dir: Path, session_id: str,
                  plan_path: Path) -> tuple[list[str], list[str]]:
    """Returns (argv, overrides). overrides is the flag list a resume must
    replay (it carries the session flags that reopen this exact session)."""
    ex = cfg["executors"][executor_id]
    overrides = (["--mode", "json"] + pi_model_flags(ex)
                 + ["--session-dir", str(run_dir / "sessions"), "--session-id", session_id]
                 + PI_HERMETIC_FLAGS)
    # Plans go in as an @file: a single argv element caps out around 128KB.
    argv = ["pi", "-p"] + overrides + [f"@{plan_path}"]
    return argv, overrides


def pi_env(run_dir: Path) -> dict:
    return os.environ | {
        "PI_CODING_AGENT_DIR": str(run_dir / "pi-agent"),
        "PI_OFFLINE": "1",
        "PI_SKIP_VERSION_CHECK": "1",
    }


def parse_pi_events(jsonl_path: Path) -> dict:
    """Compact facts from a pi --mode json event stream, normalized to the same
    keys the codex parser emits so status/summaries work on either harness."""
    facts: dict = {
        "session_id": None, "turns": 0, "tool_calls": 0,
        "last_message": "", "errors": [], "usage": {}, "phase": "unknown",
    }
    if not jsonl_path.is_file():
        return facts
    with open(jsonl_path, errors="replace") as f:
        for line in f:
            line = line.strip()
            if not line.startswith("{"):
                continue
            try:
                d = json.loads(line)
            except json.JSONDecodeError:
                continue
            t = d.get("type")
            if t == "session":
                facts["session_id"] = d.get("id")
                facts["phase"] = "running"
            elif t == "turn_start":
                facts["turns"] += 1
            elif t == "tool_execution_start":
                facts["tool_calls"] += 1
            elif t == "message_end" and d.get("message", {}).get("role") == "assistant":
                m = d["message"]
                texts = [c.get("text", "") for c in m.get("content", []) if c.get("type") == "text"]
                if texts and texts[-1]:
                    facts["last_message"] = texts[-1]
                usage = m.get("usage", {})
                for src, dst in (("input", "input_tokens"), ("cacheRead", "cached_input_tokens"),
                                 ("output", "output_tokens")):
                    if isinstance(usage.get(src), (int, float)):
                        facts["usage"][dst] = facts["usage"].get(dst, 0) + usage[src]
                cost = usage.get("cost", {}).get("total")
                if isinstance(cost, (int, float)):
                    facts["usage"]["cost"] = facts["usage"].get("cost", 0) + cost
                if m.get("stopReason") in ("error", "aborted"):
                    facts["phase"] = "error"
                    facts["errors"].append(str(m.get("errorMessage"))[:200])
            elif t == "agent_end":
                if facts["phase"] != "error":
                    facts["phase"] = "complete"
    return facts


# ---------------------------------------------------------------------------
# cursor harness — cursor-agent -p --output-format stream-json
# ---------------------------------------------------------------------------

CURSOR_DEFAULT_ENV_KEY = "CURSOR_API_KEY"


def build_cursor_argv(cfg: dict, executor_id: str, workdir: Path,
                      resume_chat_id: str | None = None) -> tuple[list[str], list[str]]:
    """cursor-agent headless dispatch. effort is pinned inside the model slug
    (e.g. grok-4.5-high); composer has no effort variant so the bare slug is used.
    Returns (argv, overrides) — overrides carries the model+workspace flags a
    resume must replay to reopen the same chat."""
    ex = cfg["executors"][executor_id]
    overrides = ["--output-format", "stream-json", "--force",
                 "--model", ex["model"], "--workspace", str(workdir), "--trust"]
    argv = ["cursor-agent", "-p"] + overrides
    if resume_chat_id:
        argv += ["--resume", resume_chat_id]
    return argv, overrides


def cursor_env(cfg: dict, executor_id: str) -> dict:
    """cursor-agent reads CURSOR_API_KEY (or the provider's env_key alias) from
    the environment — passed through, never written to disk."""
    ex = cfg["executors"].get(executor_id, {})
    prov = cfg["providers"].get(ex.get("provider"), {})
    key_name = prov.get("env_key", CURSOR_DEFAULT_ENV_KEY)
    env = dict(os.environ)
    if key_name != CURSOR_DEFAULT_ENV_KEY and os.environ.get(key_name):
        env[CURSOR_DEFAULT_ENV_KEY] = os.environ[key_name]
    return env


def parse_cursor_events(jsonl_path: Path) -> dict:
    """Compact facts from a cursor-agent stream-json stream, normalized to the
    same keys codex/pi emit. Event shapes (verified live):
      {"type":"system","subtype":"init","session_id":...,"model":...}
      {"type":"tool_call","subtype":"started"|"completed",...}
      {"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":...}]}}
      {"type":"result","subtype":"success","is_error":bool,"session_id":...,
       "usage":{"inputTokens","outputTokens","cacheReadTokens","cacheWriteTokens"}}
    """
    facts: dict = {
        "session_id": None, "turns": 0, "tool_calls": 0,
        "last_message": "", "errors": [], "usage": {}, "phase": "unknown",
    }
    if not jsonl_path.is_file():
        return facts
    with open(jsonl_path, errors="replace") as f:
        for line in f:
            line = line.strip()
            if not line.startswith("{"):
                continue
            try:
                d = json.loads(line)
            except json.JSONDecodeError:
                continue
            t = d.get("type")
            if d.get("session_id") and not facts["session_id"]:
                facts["session_id"] = d["session_id"]
            if t == "system" and d.get("subtype") == "init":
                facts["phase"] = "running"
                facts["turns"] = 1  # cursor-agent runs one turn per -p invocation
            elif t == "tool_call" and d.get("subtype") == "started":
                facts["tool_calls"] += 1
            elif t == "assistant":
                texts = [c.get("text", "") for c in d.get("message", {}).get("content", [])
                         if c.get("type") == "text"]
                if texts and texts[-1]:
                    facts["last_message"] = texts[-1]
            elif t == "result":
                if d.get("is_error"):
                    facts["phase"] = "error"
                    facts["errors"].append(str(d.get("result") or d.get("subtype"))[:200])
                else:
                    facts["phase"] = "complete"
                    if d.get("result"):
                        facts["last_message"] = str(d["result"])
                usage = d.get("usage", {})
                for src, dst in (("inputTokens", "input_tokens"),
                                 ("cacheReadTokens", "cached_input_tokens"),
                                 ("outputTokens", "output_tokens")):
                    if isinstance(usage.get(src), (int, float)):
                        facts["usage"][dst] = facts["usage"].get(dst, 0) + usage[src]
    return facts


def parse_harness_events(harness: str, jsonl_path: Path) -> dict:
    if harness == "pi":
        return parse_pi_events(jsonl_path)
    if harness == "cursor":
        return parse_cursor_events(jsonl_path)
    return parse_events(jsonl_path)


def run_harness_process(argv: list[str], events_path: Path, harness: str,
                        max_minutes: float, cwd=None, env=None,
                        stdin=subprocess.DEVNULL) -> tuple[int, str]:
    """Launch a harness subprocess, streaming its output to events_path.
    Returns (exit_code, status) where status is OK | FAILED | TIMEOUT.
    Both harnesses read stdin even with a positional prompt; an
    open-but-silent stdin hangs the process forever, hence DEVNULL default."""
    with open(events_path, "w") as events:
        try:
            proc = subprocess.run(argv, cwd=cwd, env=env, stdin=stdin,
                                  stdout=events, stderr=subprocess.STDOUT,
                                  timeout=max_minutes * 60)
        except subprocess.TimeoutExpired:
            return 124, "TIMEOUT"
        except FileNotFoundError:
            if harness == "pi":
                sys.exit(f"parable: {PI_INSTALL_HINT}")
            if harness == "cursor":
                sys.exit("parable: 'cursor-agent' not found on PATH — "
                         "install with: curl https://cursor.com/install -fsS | bash")
            sys.exit("parable: 'codex' not found on PATH — install the codex CLI")
    return proc.returncode, ("OK" if proc.returncode == 0 else "FAILED")


# ---------------------------------------------------------------------------
# Event-stream parsing (codex --json)
# ---------------------------------------------------------------------------

def parse_events(jsonl_path: Path) -> dict:
    """Compact facts from a codex --json event stream."""
    facts: dict = {
        "session_id": None, "turns": 0, "tool_calls": 0,
        "last_message": "", "errors": [], "usage": {}, "phase": "unknown",
    }
    if not jsonl_path.is_file():
        return facts
    with open(jsonl_path, errors="replace") as f:
        for line in f:
            line = line.strip()
            if not line.startswith("{"):
                continue
            try:
                d = json.loads(line)
            except json.JSONDecodeError:
                continue
            t = d.get("type")
            if t == "thread.started":
                facts["session_id"] = d.get("thread_id")
                facts["phase"] = "running"
            elif t == "turn.started":
                facts["turns"] += 1
            elif t == "item.completed":
                item = d.get("item", {})
                it = item.get("type")
                if it == "agent_message":
                    facts["last_message"] = item.get("text", "")
                elif it in ("command_execution", "function_call", "custom_tool_call", "local_shell_call"):
                    facts["tool_calls"] += 1
                elif it == "error":
                    facts["errors"].append(item.get("message", "")[:200])
            elif t == "turn.completed":
                facts["phase"] = "complete"
                usage = d.get("usage", {})
                for k, v in usage.items():
                    if isinstance(v, (int, float)):
                        facts["usage"][k] = facts["usage"].get(k, 0) + v
            elif t == "turn.failed" or t == "error":
                facts["phase"] = "error"
                msg = d.get("message") or json.dumps(d)[:200]
                facts["errors"].append(str(msg)[:200])
    return facts


def summarize_run(meta: dict, facts: dict, run_dir: Path) -> str:
    usage = facts.get("usage", {})
    toks = f"in={usage.get('input_tokens', 0)} cached={usage.get('cached_input_tokens', 0)} out={usage.get('output_tokens', 0)}"
    if isinstance(usage.get("cost"), (int, float)):
        toks += f" cost=${usage['cost']:.4f}"
    last = (facts.get("last_message") or "").strip().replace("\n", " ")
    lines = [
        f"STATUS   {meta.get('status')}  exit={meta.get('exit_code')}  {meta.get('seconds')}s",
        f"EXECUTOR {meta.get('executor')}  model={meta.get('model')}  effort={meta.get('effort')}",
        f"SESSION  {meta.get('session_id')}",
        f"TURNS    {facts.get('turns')}  tool_calls={facts.get('tool_calls')}  tokens {toks}",
        f"LAST     {last[:220] or '(no agent message)'}",
        f"RUN_DIR  {run_dir}",
    ]
    if facts.get("errors"):
        lines.append(f"ERRORS   {len(facts['errors'])}: {facts['errors'][-1][:180]}")
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------

def cmd_config(args: argparse.Namespace) -> int:
    root = git_root()
    cfg, loaded = load_config(root)
    problems = validate_config(cfg)
    if args.json:
        print(json.dumps({"config": cfg, "loaded": [str(p) for p in loaded], "problems": problems}, indent=1))
        return 1 if problems else 0
    print(f"parable config — loaded: {', '.join(str(p) for p in loaded) or '(builtin Tier-0 defaults only)'}")
    par = cfg["parable"]
    print(f"defaults: executor={par.get('default_executor')} reviewer={par.get('default_reviewer')} log_dir={par.get('log_dir')}")
    print("executors:")
    for eid, ex in cfg["executors"].items():
        status = env_key_status(cfg, eid)
        cost = ex.get("cost", {})
        cost_s = f"${cost.get('in', '?')}/{cost.get('out', '?')}" if cost else "-"
        agent = ""
        if eid in custom_claude_executors(cfg):
            agent = f" agent={agent_slug(eid)}"
        print(
            f"  {eid:<10} {status:<14} {cost_s:<12} "
            f"tags={','.join(ex.get('tags', []))}{agent}"
        )
        if ex.get("use_for"):
            print(f"             use_for: {ex['use_for']}")
        if ex.get("avoid_for"):
            print(f"             avoid_for: {ex['avoid_for']}")
    research = cfg.get("research", {}).get("provider", "grep.ai")
    if research == "grep.ai":
        print("research: grep.ai — IN-DEPTH research + research-backed slides/docs/sheets route "
              "through the grep-research-skills package (invoke its skills by name using the "
              "current harness); quick lookups stay in-session; "
              "missing -> npx grep-research-skills; auth -> grep-login")
    else:
        print("research: claude — in-session with the best available model")
    print("routing:")
    for klass, chain in cfg.get("routing", {}).items():
        if isinstance(chain, list):
            print(f"  {klass:<14} -> {', '.join(chain)}")
    if cfg.get("checks"):
        print("checks: " + ", ".join(f"{cid}({','.join(c.get('when', []))})" for cid, c in cfg["checks"].items()))
    notes = (par.get("repo_notes") or "").strip()
    if notes:
        print("repo_notes:")
        for ln in notes.splitlines():
            print(f"  {ln}")
    if problems:
        print("PROBLEMS:")
        for p in problems:
            print(f"  ! {p}")
    if args.validate:
        print("VALID" if not problems else "INVALID")
        return 1 if problems else 0
    return 0


def cmd_list(_args: argparse.Namespace) -> int:
    cfg, _ = load_config(git_root())
    for eid, ex in cfg["executors"].items():
        cost = ex.get("cost", {})
        cost_s = f"${cost.get('in', '?')}/{cost.get('out', '?')}" if cost else "-"
        print(f"{eid:<10} {ex.get('model', ''):<50} {cost_s:<12} {env_key_status(cfg, eid):<14} {','.join(ex.get('tags', []))}")
    return 0


# Which usage probe backs each provider type — subscription-covered providers only.
# Metered providers (codex/pi with an API key) have no plan headroom to read.
POOL_FOR_PROVIDER_TYPE = {"subagent": "claude", "codex-native": "codex", "cursor": "cursor"}


def pools_in_config(cfg: dict) -> list[str]:
    """The subscription pools this cast actually routes to, in a stable order."""
    seen: set[str] = set()
    for ex in cfg.get("executors", {}).values():
        ptype = cfg.get("providers", {}).get(ex.get("provider"), {}).get("type")
        pool = POOL_FOR_PROVIDER_TYPE.get(ptype)
        if pool:
            seen.add(pool)
    return [p for p in ("claude", "codex", "cursor") if p in seen]


def cursor_env_key_in_config(cfg: dict) -> str:
    for ex in cfg.get("executors", {}).values():
        prov = cfg.get("providers", {}).get(ex.get("provider"), {})
        if prov.get("type") == "cursor":
            return prov.get("env_key", CURSOR_DEFAULT_ENV_KEY)
    return CURSOR_DEFAULT_ENV_KEY


def cmd_usage(args: argparse.Namespace) -> int:
    """Live subscription headroom across the pools this cast routes to, read
    from each harness's own usage endpoint — zero model tokens, no turn. The
    brain reads this BEFORE routing so load-balancing is measured, not guessed
    from throttle-after-the-fact."""
    try:
        import parable_usage
    except ImportError:
        sys.path.insert(0, str(Path(__file__).resolve().parent))
        import parable_usage  # noqa: E402
    cfg, _ = load_config(git_root())
    pools = ["claude", "codex", "cursor"] if args.all else pools_in_config(cfg)
    if not pools:
        print("parable usage — no subscription-covered pools in this cast (nothing to probe)")
        return 0
    reports = parable_usage.probe_all(pools, cursor_env_key=cursor_env_key_in_config(cfg))
    if args.json:
        print(json.dumps(reports, indent=1))
        return 0
    print("parable usage — live subscription headroom (zero model tokens)")
    print(parable_usage.format_report(reports))
    tight = [r["pool"] for r in reports
             if r.get("status") == "ok" and (parable_usage.worst_used_pct(r) or 0) >= 80]
    if tight:
        print(f"\nTIGHT: {', '.join(tight)} — route bulk work to a pool with more room.")
    return 0


def cmd_run(args: argparse.Namespace) -> int:
    root = git_root()
    cfg, _ = load_config(root)
    problems = validate_config(cfg)
    if problems:
        sys.exit("parable: config invalid:\n  " + "\n  ".join(problems))
    eid = args.executor
    if eid not in cfg["executors"]:
        sys.exit(f"parable: unknown executor '{eid}' (see: parable.py list)")
    ex = cfg["executors"][eid]
    if ex.get("enabled", True) is False:
        sys.exit(f"parable: executor '{eid}' is disabled (enabled = false in config)")
    if args.effort:
        prov_type = cfg["providers"].get(ex.get("provider"), {}).get("type")
        allowed = PI_THINKING_LEVELS if prov_type == "pi" else EFFORT_LEVELS
        if args.effort not in allowed:
            sys.exit(f"parable: --effort '{args.effort}' (allowed for {prov_type}: {', '.join(allowed)})")
        ex = {**ex, "effort": args.effort}
        cfg = {**cfg, "executors": {**cfg["executors"], eid: ex}}
    prov = cfg["providers"][ex["provider"]]
    if prov.get("type") == "subagent":
        sys.exit(f"parable: executor '{eid}' is a Claude subagent — dispatch it via the Agent tool, not parable.py run")
    if prov.get("type") in ("codex", "pi") and not os.environ.get(prov.get("env_key", "")):
        sys.exit(f"parable: env var {prov.get('env_key')} is not set (required by executor '{eid}')")
    if prov.get("type") == "cursor":
        key_name = prov.get("env_key", CURSOR_DEFAULT_ENV_KEY)
        if not os.environ.get(key_name):
            sys.exit(f"parable: env var {key_name} is not set (required by executor '{eid}')")
    plan_path = Path(args.plan)
    if not plan_path.is_file():
        sys.exit(f"parable: plan file not found: {plan_path}")
    workdir = Path(args.workdir).resolve() if args.workdir else root

    slug = args.slug or re.sub(r"[^a-z0-9-]+", "-", plan_path.parent.name.lower()).strip("-") or "task"
    log_root = root / cfg["parable"].get("log_dir", ".parable")
    run_dir = log_root / "runs" / f"{utc_stamp()}-{slug}-{eid}"
    run_dir.mkdir(parents=True, exist_ok=True)
    (run_dir / "plan.md").write_text(
        WORKSPACE_CONTRACT + "\n\n" + plan_path.read_text(errors="replace"))

    harness = {"pi": "pi", "cursor": "cursor"}.get(prov.get("type"), "codex")
    if harness == "pi":
        session_id = str(uuid.uuid4())
        write_pi_agent_dir(run_dir, build_pi_models_json(ex["provider"], prov, ex, eid))
        argv, overrides = build_pi_argv(cfg, eid, run_dir, session_id, run_dir / "plan.md")
    elif harness == "cursor":
        session_id = None  # cursor-agent mints the chat id; captured from the stream
        argv, overrides = build_cursor_argv(cfg, eid, workdir)
    else:
        session_id = None
        argv, overrides = build_run_argv(cfg, eid, workdir, run_dir / "last-message.txt")
    (run_dir / "cmd.txt").write_text(shlex.join(argv) + "\n")
    events_path = run_dir / "harness.jsonl"

    max_minutes = float(ex.get("max_minutes", 20))
    started = time.time()
    # meta.json exists from launch so `status` works on an in-progress run.
    meta = {
        "harness": harness, "executor": eid, "model": ex["model"],
        "effort": ex.get("effort", "high"), "provider": ex["provider"],
        "workdir": str(workdir), "session_id": session_id, "exit_code": None,
        "status": "RUNNING", "seconds": None,
        "overrides": overrides,
        "resumes": 0,
    }
    (run_dir / "meta.json").write_text(json.dumps(meta, indent=1))
    if harness == "pi":
        exit_code, status = run_harness_process(argv, events_path, harness, max_minutes,
                                                cwd=workdir, env=pi_env(run_dir))
    elif harness == "cursor":
        with open(run_dir / "plan.md", "rb") as plan_f:
            exit_code, status = run_harness_process(argv, events_path, harness, max_minutes,
                                                    cwd=workdir, env=cursor_env(cfg, eid), stdin=plan_f)
    else:
        with open(run_dir / "plan.md", "rb") as plan_f:
            exit_code, status = run_harness_process(argv, events_path, harness, max_minutes,
                                                    stdin=plan_f)
    seconds = round(time.time() - started, 1)

    facts = parse_harness_events(harness, events_path)
    meta.update(session_id=facts.get("session_id") or session_id, exit_code=exit_code,
                status=status, seconds=seconds)
    (run_dir / "meta.json").write_text(json.dumps(meta, indent=1))
    print(summarize_run(meta, facts, run_dir))
    return 0 if status == "OK" else 1


def find_run_dir(root: Path, cfg: dict, ref: str) -> Path:
    p = Path(ref)
    if p.is_dir() and (p / "meta.json").is_file():
        return p
    runs = root / cfg["parable"].get("log_dir", ".parable") / "runs"
    for run_dir in sorted(runs.glob("*"), reverse=True):
        meta_f = run_dir / "meta.json"
        if meta_f.is_file():
            try:
                if json.loads(meta_f.read_text()).get("session_id") == ref:
                    return run_dir
            except json.JSONDecodeError:
                continue
    sys.exit(f"parable: no run found for '{ref}'")


def cmd_resume(args: argparse.Namespace) -> int:
    root = git_root()
    cfg, _ = load_config(root)
    run_dir = find_run_dir(root, cfg, args.run)
    meta = json.loads((run_dir / "meta.json").read_text())
    if not meta.get("session_id"):
        sys.exit(f"parable: run {run_dir} has no session id — start a fresh run instead")
    n = meta.get("resumes", 0) + 1
    events_path = run_dir / f"resume-{n}.jsonl"
    harness = meta.get("harness", "codex")
    env = None
    resume_stdin = subprocess.DEVNULL
    if harness == "pi":
        # Replayed overrides carry --session-dir/--session-id: same id in the
        # same dir opens the existing session and continues it.
        argv = ["pi", "-p"] + meta.get("overrides", []) + [args.prompt]
        env = pi_env(run_dir)
    elif harness == "cursor":
        # Replay model/workspace overrides + --resume <chat id>; the delta prompt
        # goes in on stdin (cursor-agent reads a piped prompt), same as a fresh run.
        argv = (["cursor-agent", "-p"] + meta.get("overrides", [])
                + ["--resume", meta["session_id"]])
        env = cursor_env(cfg, meta.get("executor"))
        (run_dir / f"resume-{n}-prompt.txt").write_text(args.prompt)
        resume_stdin = open(run_dir / f"resume-{n}-prompt.txt", "rb")
    else:
        # `codex exec resume` does not accept -C; the session runs in the process cwd.
        argv = ["codex", "exec", "resume", meta["session_id"], "--yolo", "--json"]
        argv += meta.get("overrides", [])
        argv += [args.prompt]
    started = time.time()
    ex = cfg["executors"].get(meta.get("executor"), {})
    max_minutes = float(ex.get("max_minutes", 20))
    exit_code, status = run_harness_process(argv, events_path, harness, max_minutes,
                                            cwd=meta["workdir"], env=env, stdin=resume_stdin)
    meta["resumes"] = n
    meta["exit_code"] = exit_code
    meta["status"] = status
    meta["seconds"] = round(time.time() - started, 1)
    (run_dir / "meta.json").write_text(json.dumps(meta, indent=1))
    print(summarize_run(meta, parse_harness_events(harness, events_path), run_dir))
    return 0 if status == "OK" else 1


def merge_facts(facts_list: list[dict]) -> dict:
    """Aggregate facts across the original run and every resume, in order."""
    merged: dict = {"session_id": None, "turns": 0, "tool_calls": 0,
                    "last_message": "", "errors": [], "usage": {}, "phase": "unknown"}
    for f in facts_list:
        merged["session_id"] = merged["session_id"] or f.get("session_id")
        merged["turns"] += f.get("turns", 0)
        merged["tool_calls"] += f.get("tool_calls", 0)
        if f.get("last_message"):
            merged["last_message"] = f["last_message"]
        merged["errors"].extend(f.get("errors", []))
        for k, v in f.get("usage", {}).items():
            merged["usage"][k] = merged["usage"].get(k, 0) + v
        if f.get("phase") != "unknown":
            merged["phase"] = f["phase"]
    return merged


def cmd_status(args: argparse.Namespace) -> int:
    run_dir = Path(args.run_dir)
    meta_f = run_dir / "meta.json"
    if not meta_f.is_file():
        sys.exit(f"parable: {run_dir} has no meta.json")
    meta = json.loads(meta_f.read_text())
    harness = meta.get("harness", "codex")
    streams = [run_dir / "harness.jsonl"] + sorted(run_dir.glob("resume-*.jsonl"))
    print(summarize_run(meta, merge_facts([parse_harness_events(harness, s) for s in streams]), run_dir))
    if meta.get("status") == "RUNNING":
        live = [s for s in streams if s.is_file()]
        quiet = min((time.time() - s.stat().st_mtime for s in live), default=None)
        if quiet is not None and quiet > 180:
            print(f"WARNING  no events for {int(quiet // 60)}m — the dispatching session may have "
                  f"exited and killed this run (child processes die with it); if so, re-dispatch "
                  f"or resume rather than waiting")
    return 0


def run_check(root: Path, cid: str, check: dict, targets: str) -> dict:
    cmd = check["run"].replace("{targets}", targets)
    cwd = root / check.get("cwd", ".")
    timeout_s = float(check.get("timeout_minutes", 15)) * 60
    started = time.time()
    try:
        proc = subprocess.run(cmd, shell=True, cwd=cwd, capture_output=True,
                              text=True, timeout=timeout_s)
        rc = proc.returncode
        out = (proc.stdout or "") + (proc.stderr or "")
    except subprocess.TimeoutExpired as e:
        rc = 124
        out = ((e.stdout or b"").decode(errors="replace") if isinstance(e.stdout, bytes) else (e.stdout or "")) + "\n[TIMEOUT]"
    return {"id": cid, "rc": rc, "seconds": round(time.time() - started, 1), "output": out, "cmd": cmd}


def failure_lines(check: dict, result: dict) -> list[str]:
    tail_n = int(check.get("tail_lines", 8))
    lines = result["output"].splitlines()
    pattern = check.get("grep")
    if pattern:
        matched = [ln for ln in lines if re.search(pattern, ln)]
        if matched:
            return matched[:12]
    return lines[-tail_n:][:12]


def cmd_verify(args: argparse.Namespace) -> int:
    root = git_root(Path(args.workdir).resolve()) if args.workdir else git_root()
    cfg, _ = load_config(root)
    checks = cfg.get("checks", {})
    if args.only:
        wanted = [c.strip() for c in args.only.split(",")]
        unknown = [c for c in wanted if c not in checks]
        if unknown:
            sys.exit(f"parable: unknown checks: {', '.join(unknown)}")
        selected = {cid: checks[cid] for cid in wanted}
    else:
        selected = {cid: c for cid, c in checks.items() if args.when in c.get("when", [])}
    if not selected:
        print(f"PARABLE VERIFY  no checks configured for when={args.when}")
        return 0
    out_dir = root / cfg["parable"].get("log_dir", ".parable") / "verify" / utc_stamp()
    out_dir.mkdir(parents=True, exist_ok=True)
    # Targets reach the check command through the shell — quote each token.
    targets = " ".join(shlex.quote(t) for t in (args.targets or "").split())
    results = []
    for cid, check in selected.items():
        res = run_check(root, cid, check, targets)
        (out_dir / f"{cid}.log").write_text(res["output"])
        results.append((check, res))
    n_fail = sum(1 for _, r in results if r["rc"] != 0)
    total_s = round(sum(r["seconds"] for _, r in results), 1)
    print(f"PARABLE VERIFY  {len(results)} checks  {len(results) - n_fail} pass  {n_fail} fail  ({total_s}s)")
    for check, res in results:
        mark = "PASS" if res["rc"] == 0 else "FAIL"
        line = f"{mark} {res['id']:<14} {res['seconds']}s"
        if res["rc"] != 0:
            line += f"  exit {res['rc']}  log: {out_dir / (res['id'] + '.log')}"
        print(line)
        if res["rc"] != 0:
            for ln in failure_lines(check, res):
                print(f"    {ln[:200]}")
    return 1 if n_fail else 0


SECRETISH_RE = re.compile(r"(^|/)\.env(\.|$)|\.pem$|\.key$|credential|secret|token", re.IGNORECASE)


def partition_untracked(untracked: list[str], log_dir: str, paths: list[str]) -> tuple[list[str], list[str]]:
    """Split untracked files into (content_included, listed_by_name_only).

    Untracked content goes to an external reviewer model, so it is included
    only when the caller scoped the review with --paths, and never for
    secret-looking filenames. Everything else is listed by name so omissions
    are visible instead of silent."""
    include, listed = [], []
    for uf in filter(None, untracked):
        if uf.startswith(log_dir.rstrip("/") + "/"):
            continue
        in_scope = bool(paths) and any(uf == p or uf.startswith(p.rstrip("/") + "/") for p in paths)
        if in_scope and not SECRETISH_RE.search(uf):
            include.append(uf)
        else:
            listed.append(uf)
    return include, listed


REVIEW_FALLBACK_PROMPT = """You are reviewing a code diff — read and pronounce, do not investigate.
Deterministic verification (typecheck, tests) has already run; do NOT run commands, builds, or tests,
and do NOT modify any file. You may briefly read a file for context, but your job is judgment on the
diff below, delivered fast.

Report every issue you find, including ones you are uncertain about or consider low-severity. Do not
filter for importance or confidence — a separate verification step will do that. Your goal is coverage:
it is better to surface a finding that later gets filtered out than to silently drop a real bug. For
each finding give: file:line, what is wrong, why it matters, confidence (high/medium/low), severity
(P0/P1/P2). Also check explicitly: acceptance criteria met (when a plan is provided); out-of-scope
hunks (list each); tests for the changed behavior that test behavior, not implementation. Findings
only — no praise, no summary of correct code. End with the complete findings list as your final
message. The diff:

"""


def review_rubric() -> str:
    """The rubric lives in references/review-prompt.md (single source, below the `---` rule);
    the embedded constant serves installs where the reference file is absent."""
    ref = Path(__file__).resolve().parent.parent / "references" / "review-prompt.md"
    try:
        rubric = ref.read_text().split("---", 1)[1].strip()
        return rubric + "\n\nThe diff:\n\n"
    except (OSError, IndexError):
        return REVIEW_FALLBACK_PROMPT

WORKSPACE_CONTRACT = """\
WORKSPACE CONTRACT: You may be one of several agents working in this repository \
concurrently. Modified or untracked files you did not create belong to another \
agent's in-flight work. Never revert, checkout, stash, clean, or delete changes \
outside the files your task assigns to you — if the workspace state looks wrong, \
report it in your final message instead of fixing it. Scope limits like "only \
touch X" bound what YOU edit; they are never an instruction to remove someone \
else's changes."""

REVIEW_DEFAULT_EFFORT = "medium"   # reviews run at their own effort, independent of the implement dispatch
REVIEW_MAX_MINUTES = 8
PI_REVIEW_TOOLS = "read,grep,find,ls"  # reviewers are read-only: judgment on the diff, not investigation


def cmd_review(args: argparse.Namespace) -> int:
    root = git_root()
    cfg, _ = load_config(root)
    eid = args.executor
    if eid not in cfg["executors"]:
        sys.exit(f"parable: unknown executor '{eid}'")
    ex = cfg["executors"][eid]
    prov = cfg["providers"][ex["provider"]]
    if ex.get("enabled", True) is False:
        sys.exit(f"parable: reviewer '{eid}' is disabled (enabled = false in config)")
    if prov.get("type") == "subagent":
        sys.exit(f"parable: executor '{eid}' is a Claude subagent — review via the Agent tool")
    if prov.get("type") in ("codex", "pi") and not os.environ.get(prov.get("env_key", "")):
        sys.exit(f"parable: env var {prov.get('env_key')} is not set (required by reviewer '{eid}')")
    if prov.get("type") == "cursor" and not os.environ.get(prov.get("env_key", CURSOR_DEFAULT_ENV_KEY)):
        sys.exit(f"parable: env var {prov.get('env_key', CURSOR_DEFAULT_ENV_KEY)} is not set (required by reviewer '{eid}')")
    if args.author:
        author = cfg["executors"].get(args.author, {})
        if author.get("model") == ex.get("model"):
            sys.exit(f"parable: reviewer '{eid}' uses the same model as author '{args.author}' — pick a different reviewer")
    workdir = Path(args.workdir).resolve() if args.workdir else root
    paths = [p.strip() for p in (args.paths or "").split(",") if p.strip()]
    diff_args = ["git", "diff"]
    if args.base:
        diff_args += [f"{args.base}...HEAD"]
    else:
        diff_args += ["HEAD"]
    if paths:
        diff_args += ["--"] + paths
    diff = subprocess.run(diff_args, capture_output=True, text=True, cwd=workdir).stdout
    # New files are usually untracked — but their content ships to an external
    # reviewer model, so content is included only inside an explicit --paths
    # scope (and never for secret-looking names); the rest is listed by name.
    log_dir = cfg["parable"].get("log_dir", ".parable").strip("/")
    max_bytes = 200_000
    untracked = subprocess.run(["git", "ls-files", "--others", "--exclude-standard"],
                               capture_output=True, text=True, cwd=workdir).stdout.split("\n")
    include, listed = partition_untracked(untracked, log_dir, paths)
    for uf in include:
        fpath = workdir / uf
        if not fpath.is_file() or fpath.stat().st_size > max_bytes or b"\x00" in fpath.read_bytes()[:8000]:
            listed.append(uf)
            continue
        diff += subprocess.run(["git", "diff", "--no-index", "--", "/dev/null", uf],
                               capture_output=True, text=True, cwd=workdir).stdout
    if listed:
        diff += ("\n[untracked files present, content NOT included (scope with --paths to review them): "
                 + ", ".join(listed[:40]) + (" …" if len(listed) > 40 else "") + "]\n")
    if not diff.strip():
        print("PARABLE REVIEW  empty diff — nothing to review")
        return 0
    if len(diff) > 400_000:
        diff = diff[:400_000] + "\n[diff truncated at 400KB — review what is shown]\n"
    prompt = review_rubric()
    if getattr(args, "plan", None) and Path(args.plan).is_file():
        plan_block = "The plan:\n\n" + Path(args.plan).read_text(errors="replace") + "\n\n"
        prompt = prompt.replace("The diff:", plan_block + "The diff:", 1)
    prompt += diff
    max_minutes = float(args.max_minutes or REVIEW_MAX_MINUTES)
    effort = args.effort or REVIEW_DEFAULT_EFFORT
    review_dir = root / cfg["parable"].get("log_dir", ".parable") / "reviews" / f"{utc_stamp()}-{eid}"
    review_dir.mkdir(parents=True, exist_ok=True)
    if prov.get("type") == "pi":
        rc, out = review_via_pi(cfg, eid, ex, prov, review_dir, workdir, prompt, max_minutes, effort)
    elif prov.get("type") == "cursor":
        rc, out = review_via_cursor(cfg, eid, ex, review_dir, workdir, prompt, max_minutes)
    else:
        rc, out = review_via_codex(ex, prov, workdir, prompt, max_minutes, effort)
    # The full review always lands in an artifact file: stdout tails are lossy
    # and a lost finding triggers wasteful re-reviews.
    review_file = review_dir / "review.md"
    review_file.write_text(out or "")
    print(out)
    print(f"REVIEW_FILE {review_file}")
    if rc != 0 or not out:
        print(f"PARABLE REVIEW  reviewer exited {rc} with {'no' if not out else 'this'} output — "
              f"use the next configured reviewer or an available native subagent; do not retry it",
              file=sys.stderr)
    return rc


def review_via_pi(cfg: dict, eid: str, ex: dict, prov: dict, review_dir: Path,
                  workdir: Path, prompt: str, max_minutes: float,
                  effort: str) -> tuple[int, str]:
    """One-shot pi review: rubric+diff go in as an @file (argv size limits),
    a throwaway hermetic agent dir carries the provider, read-only tools,
    no session kept. Returns (exit_code, review_text)."""
    prompt_file = review_dir / "prompt.md"
    prompt_file.write_text(prompt)
    write_pi_agent_dir(review_dir, build_pi_models_json(ex["provider"], prov, ex, eid))
    argv = (["pi", "-p", "--mode", "json"]
            + pi_model_flags({**ex, "effort": effort})
            + ["--no-session", "--tools", PI_REVIEW_TOOLS]
            + PI_HERMETIC_FLAGS + [f"@{prompt_file}"])
    events_path = review_dir / "harness.jsonl"
    exit_code, status = run_harness_process(argv, events_path, "pi", max_minutes,
                                            cwd=workdir, env=pi_env(review_dir))
    facts = parse_pi_events(events_path)
    out = facts.get("last_message", "").strip()
    if status == "TIMEOUT" and not out:
        return 124, ""
    return exit_code, out


def review_via_cursor(cfg: dict, eid: str, ex: dict, review_dir: Path,
                      workdir: Path, prompt: str, max_minutes: float) -> tuple[int, str]:
    """One-shot cursor-agent review in read-only plan mode (--plan: analyze, no
    edits), the rubric+diff piped on stdin. Effort rides the model slug, so no
    effort flag here. Returns (exit_code, review_text)."""
    prompt_file = review_dir / "prompt.md"
    prompt_file.write_text(prompt)
    argv = ["cursor-agent", "-p", "--plan", "--output-format", "stream-json",
            "--model", ex["model"], "--workspace", str(workdir), "--trust"]
    events_path = review_dir / "harness.jsonl"
    with open(prompt_file, "rb") as pf:
        exit_code, status = run_harness_process(argv, events_path, "cursor", max_minutes,
                                                cwd=workdir, env=cursor_env(cfg, eid), stdin=pf)
    out = parse_cursor_events(events_path).get("last_message", "").strip()
    if status == "TIMEOUT" and not out:
        return 124, ""
    return exit_code, out


def review_via_codex(ex: dict, prov: dict, workdir: Path, prompt: str,
                     max_minutes: float, effort: str) -> tuple[int, str]:
    """Codex review reads the rubric+diff from stdin and answers in prose.
    Returns (exit_code, review_text)."""
    argv = ["codex", "exec", "--yolo", "-C", str(workdir), "--skip-git-repo-check"]
    if prov.get("type") == "codex":
        argv += provider_overrides(ex["provider"], prov)
    argv += ["-c", f'model="{ex["model"]}"']
    argv += ["-c", f'model_reasoning_effort="{effort}"']
    argv += ["-"]
    try:
        proc = subprocess.run(argv, input=prompt, text=True, capture_output=True,
                              timeout=max_minutes * 60)
    except subprocess.TimeoutExpired:
        return 124, ""
    except FileNotFoundError:
        sys.exit("parable: 'codex' not found on PATH — install the codex CLI")
    return proc.returncode, proc.stdout.strip()


def main() -> int:
    ap = argparse.ArgumentParser(prog="parable.py", description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("config", help="show merged config summary")
    p.add_argument("--validate", action="store_true")
    p.add_argument("--json", action="store_true")
    p.set_defaults(fn=cmd_config)

    p = sub.add_parser("list", help="list executors")
    p.set_defaults(fn=cmd_list)

    p = sub.add_parser("usage", help="live subscription headroom per pool (zero model tokens)")
    p.add_argument("--all", action="store_true", help="probe all pools, not just those in the cast")
    p.add_argument("--json", action="store_true")
    p.set_defaults(fn=cmd_usage)

    p = sub.add_parser("claude", help="launch Claude Code through the configured local proxy")
    p.add_argument("claude_args", nargs=argparse.REMAINDER,
                   help="arguments forwarded to Claude Code (use -- before Claude flags)")
    p.set_defaults(fn=cmd_claude)

    p = sub.add_parser("finalize", help="verify exact catalog ids and synchronize named agents")
    p.add_argument("--json", action="store_true")
    p.set_defaults(fn=cmd_finalize)

    p = sub.add_parser("agents", help="manage Parable's project-local Claude agents")
    agent_sub = p.add_subparsers(dest="agents_cmd", required=True)
    p_sync = agent_sub.add_parser("sync", help="synchronize namespaced custom agents")
    p_sync.set_defaults(fn=cmd_agents_sync)

    p = sub.add_parser("run", help="dispatch a plan to an executor")
    p.add_argument("executor")
    p.add_argument("plan")
    p.add_argument("workdir", nargs="?",
                   help="git root the executor operates in (default: current repo root)")
    p.add_argument("--slug")
    p.add_argument("--effort", help="override the executor's configured effort for this dispatch")
    p.set_defaults(fn=cmd_run)

    p = sub.add_parser("resume", help="send a delta prompt to a prior run's session")
    p.add_argument("run", help="run dir or session uuid")
    p.add_argument("prompt")
    p.set_defaults(fn=cmd_resume)

    p = sub.add_parser("status", help="compact status of a run from its event stream")
    p.add_argument("run_dir")
    p.set_defaults(fn=cmd_status)

    p = sub.add_parser("verify", help="run configured deterministic checks")
    p.add_argument("--when", default="post-implement", choices=list(CHECK_WHEN_VALUES))
    p.add_argument("--only")
    p.add_argument("--targets", default="")
    p.add_argument("workdir", nargs="?")
    p.set_defaults(fn=cmd_verify)

    p = sub.add_parser("review", help="model code-review of the current diff")
    p.add_argument("executor")
    p.add_argument("workdir", nargs="?")
    p.add_argument("--author", help="executor id that authored the diff (enforces different reviewer model)")
    p.add_argument("--base", help="review changes against this base branch instead of HEAD")
    p.add_argument("--paths", help="comma-separated paths limiting the review to the task's files")
    p.add_argument("--plan", help="path to the task's plan.md — included so the reviewer can check acceptance criteria and scope")
    p.add_argument("--effort", help=f"reviewer effort (default {REVIEW_DEFAULT_EFFORT}; never inherits the executor's implement effort)")
    p.add_argument("--max-minutes", type=float, help=f"review timeout (default {REVIEW_MAX_MINUTES})")
    p.set_defaults(fn=cmd_review)

    args = ap.parse_args()
    return args.fn(args)


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