#!/usr/bin/env python3
"""rein -- CLI for the Rein agentic kit.

Subcommands:
  rein route <assessment.json> --agent <id> [--model <m>] [--limits <orca.json>]
  rein discover-check <report.json> [--root <path>] [--repo <name=path>]
                      evidence-backed advice; no dispatch or account query
  rein doctor         environment probe (plugin wiring, CLIs, config, stack);
                      --json returns {"version", "pluginRoot", "project", "plan",
                      "verifyState", "ledger"}
  rein detect         resolved stack + commands as JSON (config > runner > auto)
  rein verify         actually RUN each resolved command and report the truth --
                      "could these be INVOKED", exit 0 when they all could, even
                      if a suite ran and failed
  rein gate           the other question: did they PASS. Runs the gate slots (lint,
                      typecheck, build, test; never testOne or custom slots), cheap
                      first, stopping at the first that fails (the rest: skipped).
                      Exits 0 passed / 1 the code is wrong / 126 the environment
                      could not run the checks. --require-review also demands a
                      current APPROVED review episode. --unit-only runs the unit slots
                      only (no testIntegration) with no browser-render demand.
                      gate.env from flow.config.json is applied ([--repo] [--env-root]);
                      gate.prepare runs first when a slot is not cached (failure: 126).
                      A slot that passed on the same tree state, command and env is
                      reused and printed "cached" (--no-cache runs it anyway). Each
                      slot's timeout is 2x its slowest recent pass (2-30 min); without
                      history, --timeout or ship.gateTimeoutSeconds.
                      --tier task --base <ref>: lint, typecheck and only the tests the
                      change since <ref> reaches (gate.affected); --tier full: every
                      gate slot with testIntegration. Both without the render demand
  rein gate-log <change> [--root <plan repo>] | --path <checkout>  [--json]
                      every gate a ship run (task and repo gates, next to its state) or a
                      checkout (`rein gate`, `rein quick done`, under
                      ~/.claude/rein/verify/log) ran: tier, slots, time, cached, budget,
                      gate.env names. Exit 1 when there is no log
  rein env [root]     `export` lines for the worktree's gate.env (flow.config.json):
                      the per-worktree environment Rein's gates use there.
                      [--repo <name>] [--env-root <plan repo>]
  rein role <name>    one role's operating profile (planner|implementer|reviewer)
                      as plain markdown -- the same section Claude Code reads via
                      /rein:rein-role, for any agent that cannot invoke a skill
  rein tasks          parsed plan as JSON (tasks.md or openspec)
  rein plan-check     mechanical findings on a drafted plan's own text, as JSON. A task's
                      Verification is looked up in its Repo: the plan root (--root, else
                      the plan file's checkout), plan.repos in its flow.config.json, the
                      plan root's sibling directory of that name, or else the repo of
                      that name registered in Orca (`orca repo list`).
                      Always exits 0 (D5: never a silent skip, never a hard stop) --
                      it runs inside the planning skill, where the judgement is the
                      planner's. `--gate` is for a RUNNER instead: exits 1 on a
                      BLOCKING finding, 126 when the file cannot be read. `--tasks
                      T001,T002` scopes it -- a finding pinned to a task nobody will
                      execute cannot stop the run, while one with no task id is
                      plan-level and always counts
  rein context        detect + tasks in ONE call -- what the loop's first agent runs
  rein close <id>     tick a task's checkbox in the plan
  rein next           the DETERMINISTIC gate: is there a task to claim, and may it be
  rein review         record / check a review episode (approval is state-bound)
  rein token-report   real per-model token accounting for the last agent run
  rein ledger         summary of recorded runs (per project / per session);
                      --json returns {"runs": [...]}
  rein baseline       mark/show/clear the reference run compared against
  rein ship <change>  drive a plan through Orca workers: dispatch, gate, audit.
                      [--root <plan repo>] [--agent <id> [--model <m>]] [--dry-run | --detach]
                      [--allow-temp-root]
                      --agent alone uses that agent's default model.
                      A plan root in a temporary directory (/tmp, /var/folders, $TMPDIR) loses
                      its .rein/ state on reboot: ship refuses it unless --allow-temp-root;
                      quick and plan-check warn.
                      --detach opens a new Orca tab that runs the loop, then returns
                      exit 0 shipped and approved / 1 failed tasks or rejected audit /
                      2 usage / 126 preflight or environment. Never pushes or merges
  rein answer [<change>] ["<text>"] [--gate <id>] [--root <plan repo>]
                      list (no text) or resolve the owner gates a running `rein ship`
                      is waiting on; option gates take one of their options. Exit 0 /
                      1 failure / 2 usage (ambiguous change or gate, invalid option)
  rein watch <change> [--root <plan repo>] [--once | --timeout <seconds>] [--skip <id,id>]
                      block until `rein ship <change>` needs the owner (a new question,
                      the run finished, or the driver died), print it, exit 0. --once
                      polls a single time and exits 1 when nothing is pending (126 when
                      Orca cannot be asked); --skip hides gates the owner deferred. Exit 2
                      usage
  rein quick "<sentence>" [--repo <name>] [--base <branch>] [--here [--dirty-ok]]
                      [--agent <id> [--model <m>]] [--linear <id>] [--allow-config <path,...>]
                      (--allow-config: the gate config, e.g. flow.config.json, this task may
                      change; Rein's guard refuses any other change to it)
                      hand a small change to a cheap agent chosen by quota, in a tab
                      you talk to. Default: its own worktree; --here uses the checkout
                      you are in (main or a linked worktree) on its current branch, even
                      a local-only one. Exit 0 started / 1 failed / 2 usage / 126 environment
  rein quick ask "<question>" [--replaces q1]
  rein quick resolve [<path or quick-<slug>>] --id q1   (close a question answered in the agent's terminal)
                      run by that agent when it needs the owner: the question goes to
                      .rein/quick.json for the orchestrator; it then waits for the answer
  rein quick answer [<path or quick-<slug>>] "<text>" [--id q1]
                      record the owner's answer and type it into the agent's terminal.
                      Exit 0 sent / 1 no open question / 2 usage / 126 recorded, not sent
  rein quick watch [<path or quick-<slug>>] [--once]
                      block until a quick task needs the owner -- a question, a REPORT, or STALLED:
                      no sign of work ship.stallMinutes after its launch (a startup
                      dialog?) -- print it, exit 0; --once exits 1 when nothing is pending
  rein quick report "<text or file>"
                      run by that agent when it finishes: its report as text, kept in
                      .rein/quick.json (and .rein/quick-report.md); `rein quick watch`
                      prints it (REPORT), so nobody reads the agent's screen
  rein quick allow-config <path or quick-<slug>> <path,...>
                      grant --allow-config to a quick already running: updates its record
                      and tells the agent in its terminal. Exit 0 / 1 no task / 2 usage /
                      126 recorded, agent not reached
  rein quick resend [<path or quick-<slug>>] [--terminal <handle>]
                      type the task's brief (kept in .rein/quick.json) into the agent's
                      terminal again, e.g. once a startup dialog that swallowed it is gone.
                      Exit 0 sent / 1 no quick task / 2 ambiguous / 126 Orca
  rein quick done [--accept-outside] [--accept-path <abs path>]... [--accept-commit <sha>]... [--accept-hook <sha>]...
                      run by that agent when you say "listo": light gate on the
                      changed files, plus an audit only for risky paths. Never commits
                      or pushes. Exit 0 green / 1 red or unresolved outside activity (the owner
                      checks it, then --accept-outside; or names a real change of another
                      session's -- --accept-path / --accept-commit, verified and logged; or a commit
                      whose pre-commit hook Rein could not re-run and you ran by hand --
                      --accept-hook <sha>) /
                      2 no quick task / 126 setup
  rein accept-outside [--root <plan repo>] (--path <abs path> | --commit <sha> | --hook <sha>)...
                      ship: record that a change in a watched checkout is another session's
                      work, so no task of that plan is held for it. Only what really changed
                      there is accepted when a task is judged, and each use is logged
  rein done <change> [--root <plan repo>] [--native <manifest.json>] [--dry-run] [--yes] [--push]
                      close a finished change: remove its worktrees, close its tab,
                      archive the openspec plan (local commit on the plan repo's current
                      branch; its folder must be committed). --push also pushes that one
                      commit to the branch of the same name; it refuses (before removing
                      anything) another upstream name or other local commits ahead. It
                      never pushes main, master or qa: there it prints the exact git push
                      for the owner or orchestrator to run. Removes
                      NOTHING if any worktree is dirty, a main checkout or not integrated
                      (HEAD and branch) into origin/<base>; gitignored files that would be
                      lost (.env ...) are listed and need --yes, except gate.regenerable ones.
                      Each removed worktree then runs gate.cleanup (also in quick close). Exit 0 closed / 1 blockers / 2 usage / 126 Orca error
  rein quick close [<workspace path or quick-<slug> name>] [--dry-run] [--yes]
                      remove the worktrees of finished quick tasks (all of them when
                      no argument) and their records, only when clean and integrated
                      into origin/<base>; blocked ones are listed and kept (also one
                      whose gitignored files would be lost, until --yes). A --here
                      task only loses its record. The record is deleted before Orca
                      removes the worktree (only that file, never .rein/). Run it from
                      outside the worktree. Exit 0 closed / 1 blockers / 126 Orca
  rein benchmark startup [--profile default|lean|both] [--source <dir>] [--model <m>]
                      [--max-tokens <n>] [--json]
                      the first-turn context a fresh Claude worker loads under each launch
                      profile (ship.launchProfile): one `claude -p` reply per profile in a
                      scratch repo with --source's CLAUDE.md/AGENTS.md (default: here), on
                      your subscription (default model haiku). --max-tokens (default 30000)
                      applies to lean, and to default only when it is the profile asked for.
                      Exit 0 within / 1 over the limit / 2 usage / 126 claude failed
  rein migrate-config [root] [--write]
                      v1 -> v2 flow.config.json: drop models/limits/worktree, add the
                      ship block. Prints a diff; writes only with --write

Kept dependency-free on purpose: a plugin that drags a toolchain along at install
time is a plugin nobody installs. Python 3 stdlib only.
"""

from __future__ import annotations

import json
import os
import shutil
import signal
import sys
import time

VERSION = "2.1.2"

HERE = os.path.dirname(os.path.abspath(os.path.realpath(__file__)))
PLUGIN_ROOT = os.path.dirname(HERE)
sys.path.insert(0, os.path.join(PLUGIN_ROOT, "lib"))

import detect as _detect  # noqa: E402
import gate as _gate  # noqa: E402
import gate_cache as _gate_cache  # noqa: E402
import gate_env as _gate_env  # noqa: E402
import gate_log as _gate_log  # noqa: E402
import gate_run as _gate_run  # noqa: E402
import guard as _guard  # noqa: E402
import launch_profile as _profile  # noqa: E402
import migrate_config as _migrate_config  # noqa: E402
import plan as _plan  # noqa: E402
import plan_check as _plan_check  # noqa: E402
import quick_driver as _quick  # noqa: E402
import quick_inbox as _inbox  # noqa: E402
import quota as _quota  # noqa: E402
import roles as _roles  # noqa: E402
import serve as _serve  # noqa: E402
import ship as _ship  # noqa: E402
import ship_answer as _answer  # noqa: E402
import ship_done as _done  # noqa: E402
import ship_watch as _watch  # noqa: E402
import tmproot as _tmproot  # noqa: E402
import token_report as _tr  # noqa: E402
import verify as _verify  # noqa: E402

USAGE = __doc__


def _opt(argv: list[str], name: str, default: str = "") -> str:
    """Read `--name value` or `--name=value` without pulling in argparse."""
    for i, a in enumerate(argv):
        if a == f"--{name}" and i + 1 < len(argv):
            return argv[i + 1]
        if a.startswith(f"--{name}="):
            return a.split("=", 1)[1]
    return default


def _positional(argv: list[str]) -> str:
    for i, a in enumerate(argv):
        if a.startswith("-"):
            continue
        if i > 0 and argv[i - 1].startswith("--") and "=" not in argv[i - 1]:
            continue  # value of a preceding flag
        return a
    return ""


def cmd_detect(argv: list[str]) -> int:
    print(json.dumps(_detect.resolve(_positional(argv) or "."), indent=2))
    return 0


def cmd_verify(argv: list[str]) -> int:
    """Run every resolved command for real -- an inference is not a fact (D2).

    This command itself never repairs, installs, or writes anything -- it
    only runs and reports. But it runs the PROJECT'S OWN real test/lint/
    typecheck commands, in the operator's actual checkout, and those may
    write whatever they normally write (caches, coverage, build info): `rein
    verify` can leave a working tree dirtier than it found it, as a property
    of the commands being verified, not of this command.

    Exits non-zero only when a command is NOT INVOCABLE (missing binary, a
    shell resolution failure): a setup problem. A command that ran and
    reported failure, or one that timed out, is the ordinary state of a repo
    mid-change and does not fail this exit code -- conflating the two is
    exactly the misdirection this command exists to remove.

    `--only slot1,slot2` restricts execution to those slots -- the loop's
    Prepare precheck reads only test/lint/typecheck and has no use for
    running `build` (or the whole `test` suite a second time) in the
    operator's main checkout before Isolate.
    """
    root = _positional(argv) or "."
    timeout_raw = _opt(argv, "timeout", str(_verify.DEFAULT_TIMEOUT))

    # --plan: the per-task Verification commands, not the project's slots.
    # Nobody ran these before implementers were paid, and two tasks in a real
    # plan named a test module that did not exist -- they "passed" against
    # `no tests collected` and the defect surfaced two hours later at review.
    if "--plan" in argv:
        try:
            timeout = float(timeout_raw)
        except ValueError:
            timeout = _verify.DEFAULT_TIMEOUT
        resolved_root = os.path.abspath(root)
        # Select the plan exactly as `rein tasks --change` does (source and
        # path from the resolved config). `--change` used to be passed as the
        # SOURCE, so it selected nothing and an empty verdict read as green.
        cfg = _detect.resolve(resolved_root)
        plan_doc = _plan.read_plan(
            resolved_root,
            source=_opt(argv, "source") or cfg["plan"]["source"],
            change=_opt(argv, "change"),
            configured=_opt(argv, "path") or cfg["plan"].get("path", ""),
        )
        if not plan_doc.get("exists"):
            err = plan_doc.get("error") or f"no plan at {plan_doc.get('path')}"
            if "--json" in argv:
                print(json.dumps({"error": err, "root": resolved_root, "results": {}}, indent=2))
            else:
                print(f"rein verify --plan: {err}")
            return 2
        report = _verify.verify_plan(resolved_root, plan_doc.get("tasks") or [], timeout=timeout,
                                     env=_gate_env.for_worktree(resolved_root, repo=_opt(argv, "repo")))
        if "--json" in argv:
            print(json.dumps(report, indent=2))
        else:
            print(f"plan verifications for {report['root']}")
            if not report["results"]:
                print("  (the plan declares no tasks)")
            for tid, r in report["results"].items():
                mark = "✗" if r["outcome"] in (_verify.OUTCOME_PROVES_NOTHING,
                                               _verify.OUTCOME_NOT_INVOCABLE) else "·"
                print(f"  {mark} {tid:<6} {r['outcome']:<15} {r['command'][:56]}")
                if r.get("reason"):
                    print(f"           {r['reason'][:90]}")
            if report["unusable"]:
                print(f"\n  UNUSABLE: {', '.join(report['unusable'])} — these cannot confirm their "
                      f"own criteria, so an implementer would 'pass' without proving anything")
            else:
                print("\n  every verification can prove something (failing is fine; proving nothing is not)")
        # Non-zero ONLY for unusable. A verification that fails before the work
        # exists is the normal state of a plan and must not stop anything.
        return 0 if report["allUsable"] else 1
    try:
        timeout = float(timeout_raw)
    except ValueError:
        print(f"invalid --timeout: {timeout_raw!r}")
        return 2
    only_raw = _opt(argv, "only")
    only = {s.strip() for s in only_raw.split(",") if s.strip()} if only_raw else None

    resolved = _detect.resolve(root)
    env = _gate_env.for_worktree(resolved["root"], repo=_opt(argv, "repo"), resolved=resolved)
    report = _verify.verify_commands(resolved, timeout, only=only, env=env or None)
    _verify.write_state(resolved["root"], report)

    if "--json" in argv:
        print(json.dumps(report, indent=2))
        return 0 if report["allInvocable"] else 1

    print(f"verify: {resolved['root']}")
    if not report["results"]:
        print("  (no commands resolved -- nothing to verify)")
        return 0
    for slot, res in sorted(report["results"].items()):
        code = res["exitCode"] if res["exitCode"] is not None else "-"
        print(f"  {slot:<10} [{res['outcome']:<13}] exit={code}  {res['elapsedMs']}ms  $ {res['command']}")
        if res["outcome"] in (_verify.OUTCOME_NOT_INVOCABLE, _verify.OUTCOME_TIMEOUT) and res["error"]:
            print(f"    {res['error']}")
        for line in res["outputHead"][:5]:
            print(f"    | {line}")
    if not report["allInvocable"]:
        not_invocable = [s for s, r in report["results"].items() if not r["invocable"]]
        print(f"\n  NOT INVOCABLE: {', '.join(sorted(not_invocable))} -- a setup problem, fix the environment")
    return 0 if report["allInvocable"] else 1


def cmd_gate(argv: list[str]) -> int:
    """Did the configured commands PASS -- the question `verify` does not ask.

    `rein verify` reports invocability and exits 0 when everything could be
    invoked, so a suite that ran and failed exits 0 there. That is right for a
    precheck (nobody should be paid to work toward a gate that cannot pass)
    and unusable as a gate. This runs the same commands and applies
    `gate.decide_gate` to the same report.

    Three exit codes, and the third is the point: 0 passed, 1 the code is
    wrong, 126 the environment could not run the checks. A caller that gets
    126 has been told to fix its machine, not its code.
    """
    # Root is POSITIONAL here because that is this CLI's convention -- `rein
    # verify <root>`, `rein next <root>`, `rein doctor <root>`. Accepting
    # `--root` instead would have failed silently: `_positional` skips
    # anything starting with `-`, so `rein gate --root=/elsewhere` would have
    # resolved `.` and reported a confident verdict about the wrong repo.
    # Stopped by the driver (a gate nobody needs, or one past its bound): take the running check
    # down with it -- it runs in its own process group, which killing this process does not reach.
    def _stop(signum, _frame):
        _verify.kill_running()
        os._exit(128 + signum)
    for _sig in (signal.SIGTERM, signal.SIGINT):
        signal.signal(_sig, _stop)

    root = _positional(argv) or "."
    resolved = _detect.resolve(root)
    # No --timeout: the project's own per-command budget (ship.gateTimeoutSeconds, 900 s by
    # default), never verify's 120 s -- that made every auditor's run of a 6-10 minute suite
    # "time out" into a SETUP finding.
    default_timeout = _ship.sd.ShipConfig.from_flow_config(resolved).gate_timeout_seconds
    timeout_raw = _opt(argv, "timeout", "%g" % default_timeout)
    try:
        timeout = float(timeout_raw)
    except ValueError:
        print(f"gate: --timeout must be a number, got {timeout_raw!r}")
        return 2

    only_raw = _opt(argv, "only")
    only = {s.strip() for s in only_raw.split(",") if s.strip()} if only_raw else None
    require_review = "--require-review" in argv
    tier = _opt(argv, "tier")
    if tier not in _gate_run.TIERS:
        print(f"gate: --tier must be task or full, got {tier!r}")
        return 2

    env = _gate_env.for_worktree(resolved["root"], repo=_opt(argv, "repo"),
                                 env_root=_opt(argv, "env-root"), resolved=resolved)
    # Only the gate slots, cheap first, stopping at the first that does not pass (gate_run).
    # --tier task: lint, typecheck and the tests the change since --base reaches.
    planned = _gate_run.plan(resolved["root"], resolved, tier=tier, only=only,
                             unit_only="--unit-only" in argv, base=_opt(argv, "base"))
    # Each slot's budget: 2x its slowest recent pass in this repo (within [2, 30] min), else
    # the timeout above. History needs a repo identity, so only a git checkout keeps one.
    history_repo = (_opt(argv, "repo") or _gate_env.repo_name_of(resolved["root"])) \
        if _gate_cache.is_git(resolved["root"]) else ""

    def budget(slot, configured):
        return _gate_run.step_budget(history_repo, slot, configured, resolved.get("commands") or {}, timeout)
    prepare = _gate_env.prepare_command(resolved, resolved["root"], history_repo or _opt(argv, "repo"))
    report = _gate_run.run_steps(resolved["root"], planned["steps"], budget, env=env,
                                 cache="--no-cache" not in argv, history_repo=history_repo,
                                 skipped=planned["skipped"], prepare=prepare)
    report["env"] = sorted(env)  # names only: the values are the repo's business
    report["tier"] = tier or "default"
    if planned["affected"] is not None:
        info = planned["affected"]
        report["affected"] = {"strategy": info.get("strategy", ""), "mode": info["mode"],
                              "reason": info.get("reason", ""), "count": len(info.get("targets") or []),
                              "targets": list(info.get("targets") or [])[:50]}
    if tier != "task":  # a task's partial run is not the repo's verification state
        _verify.write_state(resolved["root"], report)

    review = _gate.check_review(resolved["root"], _opt(argv, "change")) if require_review else None

    # Render evidence arrives as a FILE, written by whatever actually drove a
    # browser -- never as a flag an agent sets. `--render` is the same rule as
    # everywhere else in this kit: a process writes the facts, and this reads
    # them. A frontend repo with no evidence and no way to get it stays at 126,
    # which is what "nobody has looked at the UI" honestly is.
    render = None
    render_path = _opt(argv, "render")
    if render_path:
        try:
            with open(render_path, encoding="utf-8") as fh:
                render = json.load(fh)
        except (OSError, json.JSONDecodeError) as exc:
            print(f"gate: SETUP -- could not read --render {render_path}: {type(exc).__name__}: {exc}")
            return _gate.EXIT_SETUP

    tools_raw = _opt(argv, "browser-tools")
    policy = dict(resolved.get("verifyPolicy") or {})
    if "--unit-only" in argv or tier:
        # A per-task gate asks "do the commands pass", not "was the UI
        # rendered": render evidence belongs to `Type: verification` tasks.
        # It also narrows the run to the unit slots: no `testIntegration`.
        # The driver's tiers (task, full) are command gates too.
        policy["mode"] = "unit"
    if tools_raw:
        policy["tools"] = [t.strip() for t in tools_raw.split(",") if t.strip()]
    serve = {"command": (resolved.get("commands") or {}).get("serve") or "",
             "url": _opt(argv, "url")}

    decision = _gate.decide_gate(report, review=review, require_review=require_review,
                                 verify_policy=policy, serve=serve, render=render,
                                 allow_empty=tier == "task")
    unprepared = next((r.get("error") for r in report["results"].values()
                       if str(r.get("error") or "").startswith("gate.prepare failed")), "")
    if unprepared and decision["decision"] == _gate.GATE_SETUP:
        decision["reason"] = "%s -- %s" % (unprepared, decision["reason"])
    # Kept per checkout, readable afterwards with `rein gate-log --path <root>`.
    _gate_log.append(_gate_log.root_log_path(resolved["root"]), dict(
        _gate_log.summarize({"decision": decision, "report": report}), kind="gate", repo=history_repo,
        base=_opt(argv, "base"), exit=decision["exit"]))

    if "--json" in argv:
        print(json.dumps({"decision": decision, "report": report}, indent=2))
        return decision["exit"]

    mark = {_gate.GATE_GREEN: "PASS", _gate.GATE_RED: "FAIL", _gate.GATE_SETUP: "SETUP"}[decision["decision"]]
    print(f"gate: {resolved['root']}")
    for slot, res in report["results"].items():  # run order
        code = res["exitCode"] if res["exitCode"] is not None else "-"
        cached = " cached" if res.get("cached") else ""
        print(f"  {slot:<10} [{res['outcome']:<13}] exit={code}  {res['elapsedMs']}ms{cached}  $ {res['command']}")
        for line in res["outputHead"][:5]:
            print(f"    | {line}")
    if "affected" in report:
        a = report["affected"]
        print(f"  affected tests: {a['mode']} ({a['count']} file(s))" + (f" -- {a['reason']}" if a["reason"] else ""))
    print(f"\n  {mark}: {decision['reason']}")
    if decision["decision"] == _gate.GATE_SETUP:
        print("  fix the environment and run again -- this is not a verdict on the code")
    return decision["exit"]


def cmd_gate_log(argv: list[str]) -> int:
    """The gates a ship run (`<change>`) or a checkout (`--path`) ran, oldest first."""
    usage = "usage: rein gate-log <change> [--root <plan repo>] [--json]  |  rein gate-log --path <checkout> [--json]"
    change, path = _positional(argv), _opt(argv, "path")
    if bool(change) == bool(path):
        print(usage)
        return 2
    where = (_gate_log.root_log_path(path) if path else
             _gate_log.ship_log_path(_ship.ss.state_path(os.path.abspath(_opt(argv, "root", ".")), change)))
    entries = _gate_log.read(where)
    if "--json" in argv:
        print(json.dumps({"path": where, "gates": entries}, indent=2))
        return 0 if entries else 1
    if not entries:
        print(f"no gate log at {where}")
        return 1
    print(f"gates ({where}):")
    print(_gate_log.render(entries))
    return 0


def cmd_env(argv: list[str]) -> int:
    """`export` lines for a worktree's `gate.env` (flow.config.json), placeholders filled.

    `eval "$(rein env)"` in a worktree gives a shell the environment Rein's gates
    use there -- the same test database, never another worktree's.
    """
    root = _positional(argv) or "."
    env = _gate_env.for_worktree(root, repo=_opt(argv, "repo"), env_root=_opt(argv, "env-root"))
    for line in _gate_env.export_lines(env):
        print(line)
    return 0


def cmd_role(argv: list[str]) -> int:
    """One role's operating profile, for an agent that is not Claude Code.

    The profiles live in the `rein-role` SKILL, which Claude Code invokes as
    `/rein:rein-role`. Nothing else can. This prints the SAME section of the
    SAME file, so any agent -- codex, opencode, a shell -- can be handed the
    rules it is expected to be bound by.

    Deliberately not a second copy: this kit already proves one definition has
    two consumers by matching both shipped files against each other, because
    two copies of one rule drift apart with nothing failing. Reading the one
    file makes drift impossible rather than merely detectable.

    Prints and nothing else -- no event is recorded. A command whose job is to
    emit text should be safe to pipe.
    """
    if "--list" in argv or not _positional(argv):
        found = _roles.available(PLUGIN_ROOT)
        if not found:
            print(f"role: SETUP -- no role profiles found at {_roles.skill_path(PLUGIN_ROOT)}")
            return _gate.EXIT_SETUP
        print("usage: rein role <" + "|".join(found) + ">")
        return 0 if "--list" in argv else 2

    result = _roles.profile(_positional(argv), PLUGIN_ROOT)
    if not result["ok"]:
        print(f"role: {result['error']}")
        # An unknown role is the caller's mistake; an unreadable skill file is
        # this installation's. Different exit codes, because a caller that gets
        # 126 has been told to fix its machine, not its argument.
        return 2 if "unknown role" in result["error"] else _gate.EXIT_SETUP

    if "--json" in argv:
        print(json.dumps(result, indent=2))
        return 0
    print(f"# role: {result['role']}\n")
    print(result["text"])
    return 0


def cmd_tasks(argv: list[str]) -> int:
    root = _positional(argv) or "."
    cfg = _detect.resolve(root)
    print(
        json.dumps(
            _plan.read_plan(
                root,
                source=_opt(argv, "source") or cfg["plan"]["source"],
                change=_opt(argv, "change"),
                configured=_opt(argv, "path") or cfg["plan"].get("path", ""),
            ),
            indent=2,
        )
    )
    return 0


def _plan_root_of(path: str) -> str:
    """The checkout holding a plan file (its git toplevel), else the file's directory."""
    folder = os.path.dirname(os.path.abspath(path)) or "."
    code, top = _quick.run_git(folder, "rev-parse", "--show-toplevel")
    return top.strip() if code == 0 and top.strip() else folder


def _orca_repo_paths() -> dict:
    """{display name: path} of the repos registered in Orca; raises when Orca cannot say."""
    repos = _ship.OrcaClient().repo_list().get("repos") or []
    return {str(r.get("displayName")): str(r.get("path") or "") for r in repos
            if isinstance(r, dict) and r.get("displayName")}


def cmd_plan_check(argv: list[str]) -> int:
    """Mechanical, pre-agent findings on a drafted plan's own text (T001:
    the-plan-checks-itself) -- the shipped entry point for `plugins/rein/lib/
    plan_check.py`, which is otherwise unreachable in production.

    `rein plan-check <plan-file>` reads PLAN-FILE as plan markdown -- the
    drafted text before it is written to `plan.path`, or an already-written
    tasks.md -- and prints its BLOCKING/IMPORTANT findings as JSON. It never
    parses argv beyond the one positional path: no root resolution, no
    `flow.config.json` lookup, because the drafted text this exists to check
    usually is not on disk at `plan.path` yet.

    D5: unavailable is never a stop. A missing path, an unreadable file, or
    text that fails to parse each yield an empty `findings` list with the
    problem named in `error` -- this command always exits 0, so a caller can
    never be blocked by the check itself, only by what it finds.
    """
    # --gate is boolean: it must not consume a following draft path.
    path = _positional([arg for arg in argv if arg != "--gate"])
    error = ""
    findings: list[dict] = []
    if not path:
        error = "usage: rein plan-check <plan-file>"
    elif not os.path.exists(path):
        error = f"no file at {path}"
    else:
        try:
            with open(path, encoding="utf-8", errors="replace") as fh:
                text = fh.read()
        except OSError as exc:
            error = str(exc)
        else:
            findings = _plan_check.mechanical_findings(text)

            # The half that needs the repository: a verification naming a test

            # module that neither exists nor is promised. Decidable, so BLOCKING.

            # Each task's verification in ITS repo: the plan root (`--root`, else the plan
            # file's git checkout), `plan.repos`, or the plan root's sibling of that name.
            # Last, and only when needed: Orca's registered repos (a plan in an Orca worktree has
            # no sibling checkouts).
            findings += _plan_check.unbacked_findings(text, _opt(argv, "root") or _plan_root_of(path),
                                                      repo_paths=_plan_check.lazy(_orca_repo_paths))
    # Scoped to the tasks this run will actually execute: a BLOCKING finding
    # pinned to a task nobody will touch cannot waste an implementer, and
    # stopping for it is the false stop the scoping exists to prevent. A
    # finding with NO taskId is plan-level and always counts.
    only_raw = _opt(argv, "tasks")
    run_ids = [t.strip() for t in only_raw.split(",") if t.strip()] if only_raw else []
    decision = _plan_check.decide_plan_check(findings, run_ids)

    # Only the root the caller names: the draft itself usually sits in a scratch file, which is temporary.
    warnings = [w for w in (_tmproot.warning(_opt(argv, "root")),) if w]
    print(json.dumps({"path": path, "error": error, "findings": findings, "warnings": warnings,
                      "decision": decision["decision"], "reason": decision["reason"]}, indent=2))

    # A file that could not be read is a setup problem, not a clean plan. This
    # returned 0 for a missing path, which is the same silent pass the rest of
    # this CLI has been losing.
    # D5 -- "never a silent skip, never a hard stop". This command runs INSIDE
    # /rein:rein-plan, before the plan is written, where a hard stop would
    # replace the planner's judgement with a regex's. So by default it reports
    # and exits 0, missing file included, exactly as it always has.
    #
    # A RUNNER is a different consumer of the same findings: loop.js called
    # decidePlanCheck and stopped before Isolate, because a BLOCKING finding
    # found there costs nothing and the same one found at Review costs a whole
    # run. `--gate` is that second consumer, opt-in, so D5 keeps its default.
    if "--gate" not in argv:
        return 0
    if error:
        return _gate.EXIT_SETUP
    return 1 if decision["decision"] == _plan_check.DECISION_STOP else 0


def cmd_context(argv: list[str]) -> int:
    """Everything the loop needs to start, in one bash round-trip.

    The whole thesis is that turns cost context re-reads. Two commands is two
    round-trips; one is one.
    """
    root = _positional(argv) or "."
    cfg = _detect.resolve(root)
    plan = _plan.read_plan(
        root,
        source=_opt(argv, "source") or cfg["plan"]["source"],
        change=_opt(argv, "change"),
        configured=_opt(argv, "path") or cfg["plan"].get("path", ""),
    )
    ordered, stuck = _plan.order_by_dependencies(plan["pending"])
    print(
        json.dumps(
            {
                "config": cfg,
                "plan": {**plan, "tasks": plan["tasks"], "pending": ordered, "unresolvableDeps": stuck},
            },
            indent=2,
        )
    )
    return 0


def cmd_close(argv: list[str]) -> int:
    task_id = _positional(argv)
    if not task_id:
        print("usage: rein close <task-id> [root] [--change X]")
        return 2
    root = _opt(argv, "root", ".")
    cfg = _detect.resolve(root)
    plan = _plan.read_plan(root, source=cfg["plan"]["source"], change=_opt(argv, "change"),
                           configured=cfg["plan"].get("path", ""))
    if not plan["exists"]:
        print(plan.get("error") or f"no plan at {plan['path']}")
        return 1
    if _plan.close_task(plan["path"], task_id):
        print(f"closed {task_id} in {plan['path']}")
        return 0
    print(f"{task_id}: not found or already closed in {plan['path']}")
    return 1


def cmd_next(argv: list[str]) -> int:
    """The signal a bounded loop stops on. JSON by default: it is read by scripts."""
    root = _positional(argv) or "."
    cfg = _detect.resolve(root)
    result = _gate.next_task(
        root,
        source=_opt(argv, "source") or cfg["plan"]["source"],
        change=_opt(argv, "change"),
        configured=_opt(argv, "path") or cfg["plan"].get("path", ""),
    )
    # The plan says what may be worked on; it cannot say whether the gate that
    # work will be judged against can even run. That check existed only inside
    # loop.js's Prepare phase, so every other caller of `rein next` was blind
    # to it. Reads the last persisted `rein verify` report -- runs nothing.
    result = _gate.decide_claimable(
        result,
        _verify.read_state(cfg["root"]),
        cfg.get("commands") or {},
        monorepo_unconfigured=_gate.monorepo_unconfigured(cfg),
    )
    print(json.dumps(result, indent=2))
    # Exit code carries the same signal, so `rein next && ...` works in a shell
    # without parsing anything.
    return 0 if result["ready"] else 1


def cmd_review(argv: list[str]) -> int:
    if not argv:
        print("usage: rein review <record|check> [...]")
        return 2
    sub, rest = argv[0], argv[1:]
    root = _opt(rest, "root", ".")
    change = _opt(rest, "change")

    if sub == "record":
        files = [f for f in _opt(rest, "files").split(",") if f.strip()]
        findings = [f for f in _opt(rest, "findings").split("|") if f.strip()]
        vague = _gate.unconfirmed_blocking(findings, files)
        if vague:
            print("refused to record: a BLOCKING is a confirmed defect, written as `BLOCKING: <file:line> "
                  "expected: ... observed: ... evidence: <a test, a command, or the interleaving that "
                  "breaks it> breaks: <the criterion, contract or invariant>`. A suspicion is IMPORTANT, and "
                  "a red gate alone is not a finding. " + " ; ".join(
                      "missing %s in: %s" % (", ".join(m), t[:100]) for t, m in vague))
            return 1
        try:
            episode = _gate.record_review(
                root, change, _opt(rest, "verdict"), files, findings,
                _opt(rest, "reviewer"), _opt(rest, "agent"),
            )
        except (ValueError, OSError) as exc:
            print(f"refused to record: {exc}")
            return 1
        print(f"recorded {episode['verdict']} for {episode['change'] or '(change)'} -> {episode['path']}")
        print(f"  reviewed {len(episode['reviewed_files'])} file(s), state hash {episode['reviewed_state_hash'][:16]}...")
        return 0

    if sub == "check":
        result = _gate.check_review(root, change)
        if result["ok"]:
            print(f"APPROVED and current -- reviewed by {result['reviewer']} ({result['episode']})")
            for f in result.get("findings", [])[:10]:
                print(f"  finding [{f['severity']}]: {f['text']}")
            return 0
        print(f"gate NOT satisfied: {result['reason']}")
        if result["changed"]:
            print("  changed since approval: " + ", ".join(result["changed"][:10]))
        for f in result.get("findings", [])[:10]:
            print(f"  finding [{f['severity']}]: {f['text']}")
        return 1

    print(f"unknown review subcommand: {sub}\n\nusage: rein review <record|check> [...]")
    return 2


def cmd_token_report(argv: list[str]) -> int:
    return _tr.main(argv)


def cmd_ledger(argv: list[str]) -> int:
    rows = _tr.read_ledger()

    if not rows and "--json" in argv:
        print(json.dumps({"runs": []}, indent=2))
        return 0
    if not rows:
        print(f"ledger is empty ({_tr.LEDGER_PATH})")
        print("run `rein token-report` on a run's transcripts to record one")
        return 0

    as_json = "--json" in argv
    if as_json:
        print(json.dumps({"runs": rows}, indent=2, ensure_ascii=False))
        return 0

    try:
        baseline = _tr.read_baseline()
    except _tr.BaselineCorruptError as exc:
        print(str(exc))
        return 1
    print(_tr.render_ledger(rows, _tr.LEDGER_PATH, baseline))
    return 0


def cmd_baseline(argv: list[str]) -> int:
    if not argv:
        print("usage: rein baseline <mark|show|clear> [wf_id]")
        return 2
    sub, rest = argv[0], argv[1:]

    if sub == "mark":
        try:
            record = _tr.mark_baseline(_positional(rest) or None)
        except (ValueError, OSError) as exc:
            print(str(exc))
            return 1
        project = record.get("project") or "unknown project"
        print(f"baseline marked: {record['wf_id']} ({project})")
        return 0

    if sub == "show":
        try:
            baseline = _tr.read_baseline()
        except _tr.BaselineCorruptError as exc:
            print(str(exc))
            return 1
        print(_tr.render_baseline(baseline))
        return 0

    if sub == "clear":
        print("baseline cleared" if _tr.clear_baseline() else "no baseline marked")
        return 0

    print(f"unknown baseline subcommand: {sub}\n")
    print("usage: rein baseline <mark|show|clear> [wf_id]")
    return 2


def cmd_serve_probe(argv: list[str]) -> int:
    """Start `--command`, poll `--url` until it accepts a TCP connection, tear down.

    Default (single-shot) form: start, poll, ALWAYS tear down, report. Prints
    exactly one JSON object and exits 0 only when ready is true, so an agent
    gets the signal from the exit code without parsing (D3-adjacent: a boolean
    the loop can check, not a sentence). Good for a throwaway boot check.

    `--start --pidfile <path>` starts the same way but leaves the process
    group running past this call (its pgid recorded in `pidfile`) so a render
    can happen against it. `--stop --pidfile <path>` is the matching teardown
    -- the SAME CLI, invoked a second time, still the one thing that owns the
    server's lifecycle (D2).
    """
    if "--stop" in argv:
        pidfile = _opt(argv, "pidfile")
        if not pidfile:
            print("usage: rein serve-probe --stop --pidfile <path>")
            return 2
        stopped, error = _serve.stop(pidfile)
        print(json.dumps({"stopped": stopped, "error": error}))
        return 0 if stopped else 1

    command = _opt(argv, "command")
    url = _opt(argv, "url")
    if not command or not url:
        print(
            "usage: rein serve-probe --command <c> --url <u> [--timeout N] [--cwd <dir>] "
            "[--start --pidfile <path>] | rein serve-probe --stop --pidfile <path>"
        )
        return 2
    timeout_raw = _opt(argv, "timeout", str(_serve.DEFAULT_TIMEOUT))
    try:
        timeout = float(timeout_raw)
    except ValueError:
        print(f"invalid --timeout: {timeout_raw!r}")
        return 2
    cwd = _opt(argv, "cwd", ".")

    if "--start" in argv:
        pidfile = _opt(argv, "pidfile")
        if not pidfile:
            print("usage: rein serve-probe --command <c> --url <u> --start --pidfile <path>")
            return 2
        result = _serve.start(command, cwd, url, timeout, pidfile)
        print(json.dumps(result.to_dict()))
        return 0 if result.ready else 1

    result = _serve.probe(command, cwd, url, timeout)
    print(json.dumps(result.to_dict()))
    return 0 if result.ready else 1


def _annotated_verify_state(verify_state, resolved):
    """`verify_state` with each slot's freshness resolved, so `--json` says
    what the text output says.

    A persisted outcome is only about the command it was recorded against.
    flow.config.json, a lockfile-driven autodetect change or a newly-set
    `subproject` key can all rewrite the configured command out from under a
    report that already ran, and the text path has annotated that since it
    shipped. Emitting the raw report under --json made the two surfaces of
    one report disagree.
    """
    if not verify_state:
        return verify_state
    out = dict(verify_state)
    results = {}
    for slot, vr in (verify_state.get("results") or {}).items():
        cmd = (resolved.get("commands") or {}).get(slot)
        fresh = bool(vr) and vr.get("command") == cmd
        results[slot] = {
            **vr,
            "fresh": fresh,
            "freshness": "fresh" if fresh else "stale -- the command changed since `rein verify` last ran",
        }
    out["results"] = results
    return out


def cmd_doctor(argv: list[str]) -> int:
    root = argv[0] if argv and not argv[0].startswith("-") else "."
    resolved = _detect.resolve(root)
    ok = "ok"
    miss = "MISSING"

    plan = _plan.read_plan(
        root,
        source=resolved["plan"]["source"],
        change=_opt(argv, "change"),
        configured=resolved["plan"].get("path", ""),
    )
    if plan["exists"]:
        plan_state = f"{len(plan['pending'])} pending / {len(plan['tasks'])} tasks"
    elif plan["path"]:
        plan_state = f"NOT FOUND at {plan['path']}"
    else:
        # D3: no location to report a NOT FOUND at (an openspec source with
        # no change named) -- `error` names the changes that exist instead.
        plan_state = plan.get("error", "no plan")
    # Last-known verification state, read-only -- doctor never runs a command
    # itself (that is `rein verify`'s job). "known" means a prior `rein
    # verify` ran and persisted a report; nothing here re-derives it.
    verify_state = _verify.read_state(resolved["root"])
    verify_results = (verify_state or {}).get("results", {})
    runs = _tr.read_ledger()
    latest = _tr.latest_workflow_dir()

    if "--json" in argv:
        report = {
            "version": VERSION,
            "pluginRoot": PLUGIN_ROOT,
            "project": resolved,
            "plan": {**plan, "state": plan_state},
            # Annotated, not raw: the text path below compares each persisted
            # command against the currently resolved one and says "stale" when
            # they differ. Emitting the raw report here made the two surfaces
            # of the same data disagree -- a machine consumer read `ok` for a
            # command that had changed under it, the exact defect
            # TestChangedCommandIsStale exists to prevent for the text output.
            "verifyState": _annotated_verify_state(verify_state, resolved),
            "ledger": {
                "path": _tr.LEDGER_PATH,
                "runCount": len(runs),
                "latestWorkflowRun": latest,
            },
        }
        print(json.dumps(report, indent=2))
        return 0

    print(f"rein {VERSION}")
    print(f"  plugin root : {PLUGIN_ROOT}")
    print(f"  CLAUDE_PLUGIN_ROOT env : {os.environ.get('CLAUDE_PLUGIN_ROOT') or '(not set)'}")
    print(f"  `rein` on PATH : {shutil.which('rein') or '(not on PATH -- call it via $CLAUDE_PLUGIN_ROOT/bin/rein)'}")
    print()
    print(f"  project root : {resolved['root']}")
    print(f"  flow.config.json : {'found' if resolved['configFound'] else 'not found (using autodetect)'}")
    print(f"  stack : {resolved['stack']}" + (f"  subtypes: {', '.join(resolved['subtypes'])}" if resolved["subtypes"] else ""))
    print(f"  package manager : {resolved['packageManager'] or '-'}")
    print(f"  task runner : {resolved['taskRunner'] or '-'}")
    print(f"  plan source : {resolved['plan']['source']}   tracker: {resolved['tracker']['kind']}")
    print(f"  plan        : {plan_state}")
    print()
    print("  resolved commands:")
    if resolved["commands"]:
        # Aligns on the LONGEST slot name actually present, not a fixed
        # width -- a long slot like `harnessTest` must not push the cmd/source
        # columns out of alignment for every other row (finding 4). Never
        # narrower than the historical width of 10, so a fixture with only
        # short names prints byte-identically to before this change.
        slot_width = max([10] + [len(s) for s in resolved["commands"]])
        # The command column was hardcoded at 40 while the single-subproject path
        # now emits `cd <dir> && <cmd>`, which is systematically longer -- so the
        # [source] column broke on exactly the repos this change added support for.
        cmd_width = max([40] + [len(c) for c in (resolved["commands"] or {}).values()])
        for slot, cmd in sorted(resolved["commands"].items()):
            vr = verify_results.get(slot)
            if vr is None:
                status = "  verified: unknown -- run `rein verify`"
            elif vr.get("command") == cmd:
                # Only trust the persisted outcome when it was recorded
                # against the SAME configured command we are printing --
                # flow.config.json, a lockfile-driven autodetect change, or a
                # newly-set `subproject` key can all rewrite `cmd` out from
                # under a report that already ran (finding 1).
                status = f"  verified: {vr['outcome']} (exit={vr['exitCode']})"
            else:
                status = "  verified: stale -- the command changed since `rein verify` last ran"
            source = f"[{resolved['commandSources'][slot]}]"
            print(f"    {slot:<{slot_width}} {cmd:<{cmd_width}} {source}{status}")
    else:
        print("    (none)")
    if resolved["missingCommands"]:
        print(f"    {miss}: {', '.join(resolved['missingCommands'])} -- set them in flow.config.json")
    if verify_state:
        import datetime as _dt

        checked = _dt.datetime.fromtimestamp(verify_state["checkedAt"]).isoformat(timespec="seconds")
        print(f"    last verified: {checked}")
    print()
    vp = resolved["verifyPolicy"]
    tools_str = ", ".join(vp["tools"]) if vp["tools"] else "(none reachable)"
    print(f"  verify : mode={vp['mode']}  tools={tools_str}")
    if resolved.get("verifyWarnings"):
        for w in resolved["verifyWarnings"]:
            print(f"    WARNING: {w}")
    serve = resolved.get("serve")
    if serve is not None:
        serve_cmd = serve["command"] or miss
        print(f"  serve  : {serve_cmd} @ {serve['url']}")
    print()
    print(f"  models : aux={resolved['models']['aux']}  impl={resolved['models']['impl']}  review={resolved['models']['review']}")
    print(f"  limits : maxTaskSteps={resolved['limits']['maxTaskSteps']}  maxReviewRounds={resolved['limits']['maxReviewRounds']}")
    print()
    print(f"  capabilities : {', '.join(resolved['capabilities']) or '(none)'}")
    for optional in ("graphify", "openspec", "serena", "codegraph"):
        if optional in resolved["capabilities"]:
            state = ok
        else:
            state = "absent (optional -- flow degrades, never breaks)"
        print(f"    {optional:<10} {state}")
    print()
    print(f"  ledger : {_tr.LEDGER_PATH} ({len(runs)} run(s) recorded)")
    print(f"  latest workflow run : {latest or '(none found)'}")
    return 0


def _flag_missing_value(argv: list[str], names: tuple) -> bool:
    """True when a value flag in `names` has no usable value: last, followed by a
    `-`-led token (`--agent --model x`), or empty (`--agent=`)."""
    for i, a in enumerate(argv):
        name, eq, inline = a.partition("=")
        if name in names:
            if eq and not inline:
                return True
            if not eq and (i + 1 >= len(argv) or argv[i + 1].startswith("-")):
                return True
    return False


def cmd_ship(argv: list[str]) -> int:
    """Run a planned change through Orca workers, resuming from .rein/ship/<change>.json.

    Put the change name first: `_positional` treats the word after a bare flag
    as that flag's value, so `--dry-run <change>` is filtered out below.
    """
    change = _positional([a for a in argv if a not in ("--dry-run", "--detach", "--allow-temp-root")])
    agent, model = _opt(argv, "agent"), _opt(argv, "model")
    detach = "--detach" in argv
    usage = ("usage: rein ship <change> [--root <plan repo>] [--agent <id> [--model <m>]] "
             "[--dry-run | --detach] [--allow-temp-root]  (--detach and --dry-run cannot be combined)")
    if _flag_missing_value(argv, ("--agent", "--model")):
        print(usage)
        return 2
    if agent and not model:
        model = _quota.default_model(agent)
        if not model:
            print("%s\nunknown agent %r without --model; known agents: %s"
                  % (usage, agent, ", ".join(_quota.DEFAULT_MODELS)))
            return 2
    if not change or bool(agent) != bool(model) or (detach and "--dry-run" in argv):
        print(usage)
        return 2
    root = _opt(argv, "root", ".")
    temp = _tmproot.warning(root)
    if temp:
        if "--dry-run" not in argv and "--allow-temp-root" not in argv:
            print("rein ship refuses a plan root in a temporary directory: %s.\nMove the plan to a checkout "
                  "that survives a restart, or pass --allow-temp-root to run anyway." % temp)
            return 2
        print("warning: %s" % temp)
    if detach:
        return _ship.detach_ship(root, change, _ship.OrcaClient(), override=(agent, model) if agent else None,
                                 allow_temp_root="--allow-temp-root" in argv)
    env = dict(os.environ)
    dry_run = "--dry-run" in argv
    if dry_run:
        env.setdefault("ORCA_TERMINAL_HANDLE", "dry-run")
        client = _ship.dry_run_client(root, change, lambda line: print(f"dry-run: {line}"))
    else:
        client = _ship.OrcaClient()
    return _ship.run_ship(
        root, change, client, _ship.default_gate_runner, env=env, clock=time.time,
        dry_run=dry_run, override=(agent, model) if agent else None,
    )


def cmd_answer(argv: list[str]) -> int:
    """List or resolve the owner gates of a `rein ship` run, in plain language.

    `rein answer [<change>]` lists; `rein answer <change> "<text>"` resolves. With the
    change omitted, `--change` is the way to answer: `rein answer "<text>" --change <c>`.
    """
    usage = 'usage: rein answer [<change>] ["<text>"] [--gate <id>] [--root <plan repo>]'
    known = ("--root", "--gate", "--change")
    words, skip = [], False
    for a in argv:
        if skip:
            skip = False
        elif a in known:
            skip = True
        elif a.startswith("-") and a.split("=")[0] in known:
            continue
        elif a.startswith("-"):
            print(usage)
            return 2
        else:
            words.append(a)
    flag_change = _opt(argv, "change")
    if skip or len(words) > (1 if flag_change else 2):
        print(usage)
        return 2
    if flag_change:
        change, text = flag_change, (words[0] if words else "")
    else:
        change, text = (words[0] if words else ""), (words[1] if len(words) == 2 else "")
    return _answer.run_answer(_opt(argv, "root", "."), change, text, _opt(argv, "gate"), _ship.OrcaClient())


def cmd_watch(argv: list[str]) -> int:
    """`rein watch <change>`: return owner decisions to an actively waiting caller."""
    usage = "usage: rein watch <change> [--root <plan repo>] [--once | --timeout <seconds>] [--skip <id,id>]"
    timeouts, argv, bad_timeout = _take_values(argv, "--timeout")
    if bad_timeout or len(timeouts) > 1:
        print(usage)
        return 2
    timeout = None
    if timeouts:
        try:
            timeout = float(timeouts[0])
        except ValueError:
            print(usage)
            return 2
        if not 0 < timeout <= 300 or "--once" in argv:
            print(usage)
            return 2
    words, skip, once = [], False, False
    for a in argv:
        if skip:
            skip = False
        elif a in ("--root", "--skip"):
            skip = True
        elif a.startswith(("--root=", "--skip=")):
            continue
        elif a == "--once":
            once = True
        elif a.startswith("-"):
            print(usage)
            return 2
        else:
            words.append(a)
    if skip or len(words) != 1:
        print(usage)
        return 2
    skipped = frozenset(i.strip() for i in _opt(argv, "skip").split(",") if i.strip())
    if ("--skip" in argv or any(a.startswith("--skip=") for a in argv)) and not skipped:
        print(usage)
        return 2
    options = {"timeout": timeout} if timeout is not None else {}
    return _watch.run_watch(_opt(argv, "root", "."), words[0], _ship.OrcaClient(), once=once,
                            skip=skipped, **options)


def cmd_done(argv: list[str]) -> int:
    """`rein done <change>`: remove what a finished change left behind, safely."""
    usage = "usage: rein done <change> [--root <plan repo>] [--native <manifest.json>] [--dry-run] [--yes] [--push]"
    native, argv, bad_native = _take_values(argv, "--native")
    if bad_native or len(native) > 1 or (native and "--push" in argv):
        print(usage)
        return 2
    words, skip, dry, yes, push = [], False, False, False, False
    for a in argv:
        if skip:
            skip = False
        elif a == "--root":
            skip = True
        elif a.startswith("--root="):
            continue
        elif a == "--dry-run":
            dry = True
        elif a == "--yes":
            yes = True
        elif a == "--push":
            push = True
        elif a.startswith("-"):
            print(usage)
            return 2
        else:
            words.append(a)
    if skip or len(words) != 1 or not _opt(argv, "root", "."):
        print(usage)
        return 2
    options = {"native": native[0]} if native else {}
    return _done.run_done(_opt(argv, "root", "."), words[0], _ship.OrcaClient(), dry_run=dry, yes=yes,
                          push=push, **options)


_INBOX_USAGE = ('usage: rein quick ask "<question>" [--replaces q1]  |  rein quick resolve [<path or quick-<slug>>] --id q1  |  rein quick answer [<path or quick-<slug>>] "<text>" '
                '[--id q1]  |  rein quick watch [<path or quick-<slug>>] [--once]  |  '
                'rein quick resend [<path or quick-<slug>>] [--terminal <handle>]  |  '
                'rein quick report "<text or file>"  |  rein quick allow-config <path or quick-<slug>> <path,...>')


def _quick_inbox(argv: list[str]):
    """`rein quick ask|answer|watch ...`, or None when argv is not one of them."""
    if not argv or argv[0] not in ("ask", "answer", "watch", "resend", "report", "allow-config", "resolve"):
        return None
    sub, rest = argv[0], argv[1:]
    words = [a for i, a in enumerate(rest)
             if not a.startswith("-") and not (i and rest[i - 1] in ("--id", "--terminal"))]
    flags = [a for a in rest if a.startswith("-")]
    if sub == "ask":
        replaces, rest2, bad_replaces = _take_values(rest, "--replaces")
        words2 = [a for a in rest2 if not a.startswith("-")]
        if bad_replaces or len(words2) != 1 or any(a.startswith("-") for a in rest2):
            print(_INBOX_USAGE)
            return 2
        return _inbox.ask(".", words2[0], replaces=replaces)
    if sub == "resolve":
        ids, rest2, bad_ids = _take_values(rest, "--id")
        words2 = [a for a in rest2 if not a.startswith("-")]
        if bad_ids or len(words2) > 1 or any(a.startswith("-") for a in rest2):
            print(_INBOX_USAGE)
            return 2
        return _inbox.resolve(words2[0] if words2 else "", ids, _quick.OrcaClient(), cwd=".")
    if sub == "report":
        if len(words) != 1 or flags:
            print(_INBOX_USAGE)
            return 2
        return _inbox.report(".", words[0])
    if sub == "allow-config":
        if len(words) != 2 or flags:
            print(_INBOX_USAGE)
            return 2
        return _inbox.allow_config(words[0], words[1], _quick.OrcaClient(), cwd=".")
    if sub == "answer":
        if len(words) not in (1, 2) or any(f != "--id" and not f.startswith("--id=") for f in flags):
            print(_INBOX_USAGE)
            return 2
        target, text = (words[0], words[1]) if len(words) == 2 else ("", words[0])
        return _inbox.answer(target, text, _quick.OrcaClient(), cwd=".", qid=_opt(rest, "id"))
    if sub == "resend":
        if len(words) > 1 or any(f != "--terminal" and not f.startswith("--terminal=") for f in flags) \
                or _flag_missing_value(rest, ("--terminal",)):
            print(_INBOX_USAGE)
            return 2
        return _inbox.resend(words[0] if words else "", _quick.OrcaClient(), cwd=".",
                             terminal=_opt(rest, "terminal"))
    if len(words) > 1 or any(f != "--once" for f in flags):
        print(_INBOX_USAGE)
        return 2
    return _inbox.watch(words[0] if words else "", _quick.OrcaClient(), cwd=".", once="--once" in flags)


def _take_values(argv: list, flag: str) -> tuple:
    """`(values of the repeatable flag, the rest of argv, bad)`."""
    first, _second, rest, bad = _take_accepts(argv, (flag, "--\0"))
    return first, rest, bad


def _take_accepts(argv: list, flags: tuple) -> tuple:
    """`(values of flags[0], values of flags[1], the rest of argv, bad)`: both flags repeat, so they
    are lifted out before the one-value flags are read. A flag with no value is `bad`."""
    out, rest, bad, i = ([], []), [], False, 0
    while i < len(argv):
        a = argv[i]
        i += 1
        name, eq, inline = a.partition("=")
        if name in flags:
            if eq:
                value = inline
            elif i < len(argv) and not argv[i].startswith("-"):
                value, i = argv[i], i + 1
            else:
                value = ""
            if not value:
                bad = True
            else:
                out[flags.index(name)].append(value)
        else:
            rest.append(a)
    return out[0], out[1], rest, bad


def cmd_accept_outside(argv: list[str]) -> int:
    """`rein accept-outside [--root DIR] (--path P | --commit SHA)...`: the ship owner's registered
    exit for a real change in a watched checkout that is another session's, not a worker's."""
    paths, commits, rest, bad = _take_accepts(argv, ("--path", "--commit"))
    hooks, _unused, rest, bad_hook = _take_accepts(rest, ("--hook", "--hook"))
    bad = bad or bad_hook
    root = "."
    while rest and not bad:
        if rest[0] == "--root" and len(rest) >= 2:
            root, rest = rest[1], rest[2:]
        else:
            bad = True
    short = [c for c in commits + hooks if len(c) < 7]
    if bad or not (paths or commits or hooks) or short or [p for p in paths if not os.path.isabs(p)]:
        print("usage: rein accept-outside [--root <plan repo>] (--path <absolute path> | --commit <sha, 7+ chars> | --hook <sha, 7+ chars>)...")
        return 2
    recorded = _guard.record_accepts(os.path.abspath(root), paths, commits, hooks)
    print("recorded in %s: %d path(s), %d commit(s), %d hook-checked commit(s); a task is only released for what really changed "
          "there when it is judged, and each use goes to the gate log"
          % (_guard.accept_file(os.path.abspath(root)), len(recorded["paths"]), len(recorded["commits"]),
             len(recorded["hooks"])))
    return 0


def cmd_quick(argv: list[str]) -> int:
    """Start a quick task (`rein quick "<sentence>"`) or close it (`rein quick done`).

    `--here` and `--dirty-ok` are booleans, so they are filtered out before the bare words
    are counted -- otherwise the sentence after them would read as their value. A value flag
    with no value (`--model` last, `--agent=`) or whose next token is another flag
    (`--agent --model x`) is a usage error, judged here rather than by `_opt`.
    """
    usage = ('usage: rein quick "<sentence>" [--repo <name>] [--base <branch>] [--here [--dirty-ok]] '
             "[--agent <id> [--model <m>]] [--linear <id>] [--allow-config <path,...>]  |  "
             "rein quick done [--accept-outside] [--accept-path <path>] [--accept-commit <sha>] [--accept-hook <sha>]  |  "
             "rein quick close [<path or quick-<slug>>] [--dry-run] [--yes]")
    # One pass over argv. A value flag takes the next token as its value (or `--flag=value`);
    # a missing, empty or `-`-led value is a usage error, and the token that was not consumed
    # stays visible as its own flag. Everything else that is not a flag is a bare word.
    # `close` is the subcommand only when what follows is flags, one path, one quick-* name or
    # nothing; anything else ("close the modal") is a task sentence and falls through.
    # `close` followed by any other flag is a malformed close, never a task called "close".
    inbox = _quick_inbox(argv)
    if inbox is not None:
        return inbox
    tail = [a for a in argv[1:] if a not in ("--dry-run", "--yes")]
    if argv and argv[0] == "close":
        if len(tail) <= 1 and (not tail or (not tail[0].startswith("-") and (
                tail[0].startswith("quick-") or os.sep in tail[0] or os.path.isdir(tail[0])))):
            return _done.run_quick_close(tail[0] if tail else "", _ship.OrcaClient(), cwd=".",
                                         dry_run="--dry-run" in argv, yes="--yes" in argv)
        if any(a.startswith("-") for a in tail):
            print(usage)
            return 2
    accept_paths, accept_commits, argv, bad_accept = _take_accepts(argv, ("--accept-path", "--accept-commit"))
    accept_hooks, _unused, argv, bad_hook = _take_accepts(argv, ("--accept-hook", "--accept-hook"))
    bad_accept = bad_accept or bad_hook or any(len(h) < 7 for h in accept_hooks)
    if bad_accept or ((accept_paths or accept_commits or accept_hooks) and argv != ["done"] + [
            a for a in argv[1:] if a == "--accept-outside"]):
        print(usage)
        return 2
    value_flags = ("repo", "base", "agent", "model", "linear", "allow-config")
    values: dict = {}
    words: list = []
    here = dirty_ok = bad = False
    i = 0
    while i < len(argv):
        a = argv[i]
        i += 1
        name, eq, inline = a.partition("=")
        if a == "--here":
            here = True
        elif a == "--dirty-ok":
            dirty_ok = True
        elif name.startswith("--") and name[2:] in value_flags:
            if eq:
                value = inline
            elif i < len(argv) and not argv[i].startswith("-"):
                value, i = argv[i], i + 1
            else:
                value = ""
            if not value or name[2:] in values:
                bad = True
            values[name[2:]] = value
        elif not a.startswith("-"):
            words.append(a)
    sentence = words[0] if words else ""
    temp = _tmproot.warning(".")
    if temp and sentence != "done":
        print("warning: %s" % temp)
    agent, model = values.get("agent", ""), values.get("model", "")
    # A lone --agent is fine (the driver supplies its default model); a lone --model is not.
    if bad or len(words) != 1 or ("model" in values and "agent" not in values) or (dirty_ok and not here) \
            or sentence == "close":
        print(usage)
        return 2
    if sentence == "done":
        return _quick.done_quick(cwd=".", client=_quick.OrcaClient(), accept_outside="--accept-outside" in argv,
                                 accept_paths=accept_paths, accept_commits=accept_commits,
                                 accept_hooks=accept_hooks)
    return _quick.start_quick(
        sentence, repo=values.get("repo", ""), base=values.get("base", ""), here=here, dirty_ok=dirty_ok,
        agent=agent, model=model, linear=values.get("linear", ""), cwd=".", client=_quick.OrcaClient(),
        allow_config=values.get("allow-config", ""),
    )


LEAN_STARTUP_LIMIT = 30000  # tokens; the lean profile measured 21.2k, the default 59.0k (claude -p)


def cmd_benchmark(argv: list[str]) -> int:
    """`rein benchmark startup`: measure what a fresh worker loads before its brief."""
    usage = ("usage: rein benchmark startup [--profile default|lean|both] [--source <dir>] "
             "[--model <m>] [--max-tokens <n>] [--json]")
    if not argv or argv[0] != "startup":
        print(usage)
        return 2
    opts = {"--profile": "both", "--source": ".", "--model": "haiku", "--max-tokens": ""}
    rest, as_json = argv[1:], False
    while rest:
        flag = rest.pop(0)
        if flag == "--json":
            as_json = True
        elif flag in opts and rest and not rest[0].startswith("--"):
            opts[flag] = rest.pop(0)
        else:
            print(usage)
            return 2
    profiles = list(_profile.PROFILES) if opts["--profile"] == "both" else [opts["--profile"]]
    if any(p not in _profile.PROFILES for p in profiles):
        print(usage)
        return 2
    try:
        limit = int(opts["--max-tokens"]) if opts["--max-tokens"] else LEAN_STARTUP_LIMIT
    except ValueError:
        print(usage)
        return 2
    if not shutil.which("claude"):
        print("claude is not on PATH")
        return 126
    source = os.path.abspath(opts["--source"])
    results = [_profile.measure(p, source=source, model=opts["--model"]) for p in profiles]
    for r in results:
        bounded = r["profile"] == "lean" or opts["--profile"] == r["profile"]
        r["limit"] = limit if bounded else None
        r["over"] = bool(bounded and r["tokens"] is not None and r["tokens"] > limit)
    code = 126 if any(r["error"] for r in results) else (1 if any(r["over"] for r in results) else 0)
    if as_json:
        print(json.dumps({"model": opts["--model"], "source": source, "results": results}, indent=2))
        return code
    for r in results:
        if r["error"]:
            print(f"{r['profile']:8} error: {r['error']}")
            continue
        verdict = "" if r["limit"] is None else (" OVER the limit of %d" % r["limit"] if r["over"]
                                                 else " (limit %d)" % r["limit"])
        print(f"{r['profile']:8} {r['tokens']:>8,} tokens in the first turn{verdict}")
    known = {r["profile"]: r["tokens"] for r in results if r["tokens"]}
    if len(known) == 2:
        print(f"lean saves {1 - known['lean'] / known['default']:.0%} of the startup context on every turn")
    return code


def cmd_migrate_config(argv: list[str]) -> int:
    """Show (and with --write, apply) the v1 -> v2 rewrite of flow.config.json."""
    usage = "usage: rein migrate-config [root] [--write]"
    flags = [a for a in argv if a.startswith("-")]
    paths = [a for a in argv if not a.startswith("-")]
    if any(f != "--write" for f in flags) or len(paths) > 1:
        print(usage)
        return 2
    path = os.path.join(paths[0] if paths else ".", "flow.config.json")
    try:
        with open(path, encoding="utf-8") as fh:
            before = fh.read()
        cfg = json.loads(before)
        if not isinstance(cfg, dict):
            raise ValueError("top level is not an object")
    except (OSError, ValueError) as exc:
        print(f"cannot read {path}: {exc}")
        return 1
    after = _migrate_config.render(_migrate_config.migrate(cfg))
    if _migrate_config.render(cfg) == after:
        print(f"{path}: nothing to change")
        return 0
    print(_migrate_config.diff(_migrate_config.render(cfg), after), end="")
    if "--write" not in flags:
        print("(dry run: pass --write to apply)")
        return 0
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(after)
    print(f"wrote {path}")
    return 0



def _evidence_roots(root, repos):
    roots = {".": root}
    for entry in repos:
        name, sep, path = entry.partition("=")
        if not sep or not name or not path or name in roots:
            raise ValueError("--repo requires a unique name=path")
        roots[name] = path
    return roots


def cmd_discover_check(argv: list[str]) -> int:
    """Check an agent's report; never investigate or execute report-supplied commands."""
    import argparse
    import impact
    parser = argparse.ArgumentParser(prog="rein discover-check")
    parser.add_argument("report")
    parser.add_argument("--root", default=".")
    parser.add_argument("--repo", action="append", default=[])
    args = parser.parse_args(argv)
    try:
        with open(args.report, encoding="utf-8") as fh:
            report = json.load(fh)
        result = impact.assess(report, _evidence_roots(args.root, args.repo))
    except (OSError, ValueError, TypeError) as exc:
        print("rein discover-check: %s" % exc, file=sys.stderr)
        return 2
    print(json.dumps(result, indent=2))
    return 0 if result["discovery"] == "sufficient" else 1


def cmd_route(argv: list[str]) -> int:
    """Read evidence, return advice; never launches a worker or queries an account."""
    import routing
    import argparse
    parser = argparse.ArgumentParser(prog="rein route")
    parser.add_argument("assessment")
    parser.add_argument("--agent", required=True)
    parser.add_argument("--model", default="")
    parser.add_argument("--limits", help="saved Orca account-list JSON; absent means unknown")
    parser.add_argument("--plans", help="JSON provider-to-plan mapping; metadata only")
    parser.add_argument("--root", default=".", help="root for v2 file evidence")
    parser.add_argument("--repo", action="append", default=[])
    parser.add_argument("--execution", choices=("native", "quick", "ship"))
    plan_flags = parser.add_mutually_exclusive_group()
    plan_flags.add_argument("--plan", dest="plan", action="store_true", default=None)
    plan_flags.add_argument("--no-plan", dest="plan", action="store_false")
    audit_flags = parser.add_mutually_exclusive_group()
    audit_flags.add_argument("--audit", dest="audit", action="store_true", default=None)
    audit_flags.add_argument("--no-audit", dest="audit", action="store_false")
    args = parser.parse_args(argv)
    try:
        with open(args.assessment, encoding="utf-8") as fh:
            report = json.load(fh)
        limits = {}
        account = {}
        plans = {}
        if args.plans:
            with open(args.plans, encoding="utf-8") as fh:
                plans = json.load(fh)
            if not isinstance(plans, dict) or any(not isinstance(v, str) for v in plans.values()):
                raise ValueError("plans must map provider names to plan strings")
        if args.limits:
            import quota
            with open(args.limits, encoding="utf-8") as fh:
                account = json.load(fh)
                limits = quota.checked_limits(account)
        override = {k: v for k, v in {"execution": args.execution,
                    "plan": args.plan, "audit": args.audit}.items() if v is not None}
        decision = routing.recommend(report, args.agent, args.model, limits, override,
                                     _evidence_roots(args.root, args.repo))
        import quota
        decision["quota_snapshot"] = quota.snapshot(account, plans)
    except (OSError, ValueError, TypeError) as exc:
        print("rein route: %s" % exc, file=sys.stderr)
        return 2
    print(json.dumps(decision, indent=2))
    return 0

COMMANDS = {
    "discover-check": cmd_discover_check,
    "route": cmd_route,
    "accept-outside": cmd_accept_outside,
    "doctor": cmd_doctor,
    "detect": cmd_detect,
    "verify": cmd_verify,
    "gate": cmd_gate,
    "gate-log": cmd_gate_log,
    "env": cmd_env,
    "role": cmd_role,
    "tasks": cmd_tasks,
    "plan-check": cmd_plan_check,
    "context": cmd_context,
    "close": cmd_close,
    "next": cmd_next,
    "review": cmd_review,
    "token-report": cmd_token_report,
    "ledger": cmd_ledger,
    "baseline": cmd_baseline,
    "serve-probe": cmd_serve_probe,
    "ship": cmd_ship,
    "answer": cmd_answer,
    "watch": cmd_watch,
    "done": cmd_done,
    "quick": cmd_quick,
    "migrate-config": cmd_migrate_config,
    "benchmark": cmd_benchmark,
}


def main(argv: list[str]) -> int:
    if not argv or argv[0] in ("-h", "--help", "help"):
        print(USAGE)
        return 0
    if argv[0] in ("-V", "--version"):
        print(f"rein {VERSION}")
        return 0
    handler = COMMANDS.get(argv[0])
    if not handler:
        print(f"unknown subcommand: {argv[0]}\n")
        print(USAGE)
        return 2
    return handler(argv[1:])


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))
