#!/usr/bin/env -S uv run --no-project --python >=3.12 python
"""scripts/fleet-wave — the coordinator kit for a builder wave in tmux.

A wave is a set of builder sessions, one tmux window each: ``claude-NAME``
(``claude -w NAME``) or ``devin-NAME`` (``scripts/devin-tree NAME``). This
tool launches them, reads their panes, types into them and closes them.
The thread-session fleet (one window per roadmap thread) is
``scripts/fleet``; this is the per-wave kit beside it. Runbook: the
``gripe-fleet`` skill (.claude/skills/gripe-fleet/SKILL.md).

    fleet-wave launch NAME --brief FILE [--tier R|W] [--devin]
    fleet-wave peek [WIN...] [-n N]     last N non-empty lines per window
    fleet-wave ctx                      context % per Claude pane
    fleet-wave prompts                  panes at a dialog: window, type, command
    fleet-wave send-line WIN TEXT       type TEXT + Enter; refuses unsent input
    fleet-wave watch-done NAME...       emit when <tree>/.claude/purpose finishes
    fleet-wave stuck-watch [NAME...]    emit on prompt / quiet / done / gone
    fleet-wave close NAME...            kill window, remove tree, drop branch;
                                        refuses a tree with unlanded commits
    fleet-wave approve TREE SHA[,SHA...] [--note ".."]   grant landing these shas
    fleet-wave approve --list [TREE]    print the approval ledger
    fleet-wave approve --revoke TREE SHA[,SHA...]        append a revoke line
    fleet-wave approved TREE SHA[,SHA...]  exit 0 iff all approved, none revoked

WIN is a window name (``claude-x``) or a wave NAME (resolved to its
``claude-``/``devin-`` window). tmux is always called with an argument
list, never a shell string. Nothing is stored in the repo.

Launch tiers are authority v2's R and W (.claude/fleet/authority.md);
``readonly`` and ``qland`` are aliases for R and W. Launch creates the
worktree first (as ``claude -w``/devin-tree would; both reuse it), runs
``scripts/fleet-authority apply --tier T --to TREE --yes`` so the tree's
.claude/settings.local.json and .devin/config.json carry the tier's
permissions, and prints the tier's Codex flags. The prompt footer is
.claude/fleet/generated/footer-<T>.txt plus .claude/fleet/builder-workflow.txt
with ``{coordinator}`` and ``{release_owner}`` filled. FLEET_LEGACY_FOOTERS=1
restores the pre-v2 launch for one release: footers/<readonly|qland>.txt
(which also needs ``{approval}``) and no apply.

The approval ledger lives in ``$FLEET_STATE/approvals/<tree>.tsv`` (never
the repo: its lines carry user and session ids, and the repo is public).
It is append-only TSV — stamp, action, shas, note, approver — and revoke
is a later line, never an edit. ``approve``/``--revoke`` run only from
the coordinator tree (``FLEET_COORDINATOR_TREE``; refused when unset):
the grant is typed by the user in the coordinator window, never relayed
through a session message. This is a convention-level guard — anything
with shell access could append to the file directly. It holds because
the coordinator session's own rules forbid it writing a grant the user
did not type, and the coordinator transcript shows which lines the user
typed; .claude/fleet/authority.md still governs.

Configuration (env, or the matching flag before the subcommand):
  FLEET_TMUX_SESSION  --session   tmux session (default: the current one)
  FLEET_TMUX_SOCKET   --socket    private tmux server (tmux -L), tests
  FLEET_WORKTREES     --worktrees worktree root (default <main>/.claude/worktrees)
  FLEET_MODEL         --model     builder model (default opus)
  FLEET_STATE                     watcher state + close log + approvals/
                                  (default ~/.claude/fleet-wave)
  FLEET_COORDINATOR_TREE          absolute path of the coordinator tree;
                                  approve/--revoke refuse anywhere else
  FLEET_COORDINATOR, FLEET_RELEASE_OWNER, FLEET_APPROVAL   footer fills
  FLEET_LEGACY_FOOTERS=1          pre-v2 footers, no apply (one release)
"""

from __future__ import annotations

import argparse
import hashlib
import itertools
import os
import re
import subprocess
import sys
import time
from collections.abc import Iterable, Sequence
from datetime import UTC, datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
# The pane parsers live in scripts/fleet_panes.py (shared with
# scripts/fleet-panes); the names stay re-exported here for the tests and
# for the small fleet-wave-only helpers built on them.
from fleet_panes import (
    ANSI_RE,
    LAUNCH_PERMISSION_MODE,
    ctx_percent,
    dialog_command,
    dialog_type,
    nonempty_tail,
    unsent_input,
)

#: Above this length, or with a newline, text goes in as one bracketed
#: paste (typed text loses its head in a Claude pane; see scripts/fleet).
PASTE_MIN = 200
PREFIXES = ("claude-", "devin-")
SCRIPTS = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPTS))
try:
    from branch_audit import Auditor
except ImportError:  # a staged copy without scripts/lib — close falls back
    Auditor = None  # type: ignore[assignment, misc]
#: Authority v2 tiers; readonly/qland are the pre-v2 names, kept as aliases.
TIER_ALIASES = {"readonly": "R", "qland": "W"}
LEGACY_FOOTER = {"R": "readonly", "W": "qland"}
PLACEHOLDER_RE = re.compile(r"\{(coordinator|release_owner|approval)\}")


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


class Fleet:
    def __init__(self, args: argparse.Namespace) -> None:
        self.session: str = args.session or os.environ.get("FLEET_TMUX_SESSION", "")
        self.socket: str = args.socket or os.environ.get("FLEET_TMUX_SOCKET", "")
        # Explicit choice only: the "opus" default is a claude alias, which
        # devin-tree would reject, so Devin launches pass --model only when set.
        self.model_set: str = args.model or os.environ.get("FLEET_MODEL", "")
        self.model: str = self.model_set or "opus"
        self.repo = Path(
            git("rev-parse", "--show-toplevel", cwd=Path.cwd()).strip() or "."
        )
        common = git("rev-parse", "--path-format=absolute", "--git-common-dir")
        self.main = Path(common.strip()).parent if common.strip() else self.repo
        wt = args.worktrees or os.environ.get("FLEET_WORKTREES", "")
        self.worktrees = Path(wt) if wt else self.main / ".claude" / "worktrees"
        st = os.environ.get("FLEET_STATE", "")
        self.state = Path(st) if st else Path.home() / ".claude" / "fleet-wave"

    # --- tmux ---------------------------------------------------------------

    def tmux(self, *argv: str, check: bool = True) -> str:
        cmd = ["tmux", *(["-L", self.socket] if self.socket else []), *argv]
        proc = subprocess.run(
            cmd, capture_output=True, text=True, encoding="utf-8", check=False
        )
        if check and proc.returncode:
            raise SystemExit(f"fleet-wave: {' '.join(cmd[:4])}…: {proc.stderr.strip()}")
        return proc.stdout

    def target(self, win: str) -> str:
        return f"={self.session}:{win}" if self.session else win

    def windows(self) -> list[str]:
        argv = ["list-windows", "-F", "#{window_name}"]
        if self.session:
            argv[1:1] = ["-t", f"={self.session}"]
        return [w for w in self.tmux(*argv).splitlines() if w]

    def resolve(self, name: str) -> str | None:
        wins = self.windows()
        if name in wins:
            return name
        for p in PREFIXES:
            if p + name in wins:
                return p + name
        return None

    def capture(self, win: str, *, escapes: bool = False) -> str | None:
        argv = ["capture-pane", "-p", "-J", "-t", self.target(win)]
        if escapes:
            argv.insert(2, "-e")
        cmd = ["tmux", *(["-L", self.socket] if self.socket else []), *argv]
        proc = subprocess.run(
            cmd, capture_output=True, text=True, encoding="utf-8", check=False
        )
        return None if proc.returncode else proc.stdout

    def wave_windows(self) -> list[str]:
        return [w for w in self.windows() if w.startswith(PREFIXES)]


def git(*argv: str, cwd: Path | None = None) -> str:
    proc = subprocess.run(
        ["git", *argv],
        capture_output=True,
        text=True,
        encoding="utf-8",
        cwd=cwd,
        check=False,
    )
    return proc.stdout if proc.returncode == 0 else ""


# --- approval ledger -----------------------------------------------------------


def _tsv(text: str) -> str:
    """One field: tabs and newlines would break the TSV line."""
    return re.sub(r"[\t\r\n]+", " ", text).strip()


def _ledger_path(f: Fleet, tree: str) -> Path:
    return f.state / "approvals" / f"{tree}.tsv"


def _ledger_entries(path: Path) -> list[tuple[str, str, list[str], str, str]]:
    """(stamp, action, shas, note, approver) per line; malformed lines skip."""
    try:
        text = path.read_text(encoding="utf-8")
    except OSError:
        return []
    out = []
    for ln in text.splitlines():
        fields = ln.split("\t")
        if len(fields) < 4 or fields[1] not in ("approve", "revoke"):
            continue
        out.append(
            (
                fields[0],
                fields[1],
                [s for s in fields[2].split(",") if s],
                fields[3],
                fields[4] if len(fields) > 4 else "",
            )
        )
    return out


def _coordinator_refusal() -> str | None:
    """Refusal reason for a ledger write, or None. Convention-level only:
    shell access beats the check — it holds because the coordinator
    session's rules forbid it writing a grant the user did not type, and
    .claude/fleet/authority.md governs."""
    coord = os.environ.get("FLEET_COORDINATOR_TREE", "")
    if not coord or not Path(coord).is_absolute():
        return "FLEET_COORDINATOR_TREE is unset or not an absolute path"
    if Path.cwd().resolve() != Path(coord).resolve():
        return f"not the coordinator tree ({coord})"
    return None


def _full_sha(want: str) -> str:
    """The full commit sha for ``want``, or "" when it does not resolve."""
    return git("rev-parse", "--verify", f"{want}^{{commit}}").strip()


def _bad_tree(tree: str) -> bool:
    return not re.fullmatch(r"[A-Za-z0-9._-]+", tree) or ".." in tree


def cmd_approve(f: Fleet, a: argparse.Namespace) -> int:
    if a.list:
        paths = (
            sorted((f.state / "approvals").glob("*.tsv"))
            if a.tree is None
            else [_ledger_path(f, a.tree)]
        )
        rc = 0
        for p in paths:
            if a.tree is None:
                print(f"--- {p.stem}")
            try:
                print(p.read_text(encoding="utf-8"), end="")
            except OSError:
                print(f"(no ledger for {p.stem})")
                rc = 1
        return rc
    if not a.tree or not a.shas:
        print(
            "fleet-wave: approve needs <tree> <sha>[,<sha>...] (or --list)",
            file=sys.stderr,
        )
        return 2
    if _bad_tree(a.tree):
        print(f"fleet-wave: bad tree name {a.tree!r}", file=sys.stderr)
        return 2
    if why := _coordinator_refusal():
        print(f"fleet-wave: REFUSED: {why}", file=sys.stderr)
        return 3
    full = []
    for s in a.shas.split(","):
        s = s.strip()
        if not (sha := _full_sha(s)):
            print(f"fleet-wave: unknown sha {s!r}", file=sys.stderr)
            return 2
        full.append(sha)
    action = "revoke" if a.revoke else "approve"
    path = _ledger_path(f, a.tree)
    path.parent.mkdir(parents=True, exist_ok=True)
    approver = _tsv(
        f"{os.environ.get('USER', '?')}+{os.environ.get('CLAUDE_SESSION_ID', 'tty')}"
    )
    line = "\t".join(
        [utc_stamp(), action, ",".join(full), _tsv(a.note or ""), approver]
    )
    with path.open("a", encoding="utf-8") as fh:
        fh.write(line + "\n")
    print(f"{action}d {a.tree} {','.join(full)}")
    return 0


def _sha_match(stored: str, want: str, full: str) -> bool:
    """A ledger sha covers ``want``: exact, the rev-parse'd form, or a
    7+-char short sha."""
    return stored in (want, full) or (len(want) >= 7 and stored.startswith(want))


def cmd_approved(f: Fleet, a: argparse.Namespace) -> int:
    entries = _ledger_entries(_ledger_path(f, a.tree))
    rc = 0
    for want in a.shas.split(","):
        want = want.strip()
        if not want:
            continue
        full = _full_sha(want)
        last = ""
        for _stamp, action, shas, _note, _who in entries:
            if any(_sha_match(s, want, full) for s in shas):
                last = action
        if last == "approve":
            continue
        print(f"{'revoked' if last else 'missing'}: {want}")
        rc = 1
    if rc == 0:
        print(f"approved {a.tree} {a.shas}")
    return rc


# --- subcommands --------------------------------------------------------------


def cmd_peek(f: Fleet, a: argparse.Namespace) -> int:
    rc = 0
    for name in a.win or f.wave_windows():
        win = f.resolve(name)
        pane = f.capture(win) if win else None
        print(f"--- {win or name}")
        if pane is None:
            print("(no such window)")
            rc = 1
            continue
        for ln in nonempty_tail(pane, a.n):
            print(ln)
    return rc


def cmd_ctx(f: Fleet, _a: argparse.Namespace) -> int:
    out = f.tmux("list-panes", "-a", "-F", "#{pane_id}\t#{session_name}:#{window_name}")
    for line in out.splitlines():
        pane_id, _, label = line.partition("\t")
        if f.session and not label.startswith(f.session + ":"):
            continue
        pane = f.capture(pane_id)
        pct = ctx_percent(pane or "")
        if pct is not None:
            print(f"{label}\t{pct}%")
    return 0


def cmd_prompts(f: Fleet, _a: argparse.Namespace) -> int:
    for win in f.windows():
        pane = f.capture(win)
        kind = dialog_type(pane or "")
        if kind:
            print(f"{win}\t{kind}\t{dialog_command(pane or '')}")
    return 0


def cmd_send_line(f: Fleet, a: argparse.Namespace) -> int:
    win = f.resolve(a.win)
    if not win:
        print(f"fleet-wave: no window {a.win}", file=sys.stderr)
        return 2
    pane = f.capture(win, escapes=True)
    if pane is None:
        print(f"fleet-wave: cannot read {win}", file=sys.stderr)
        return 2
    plain = ANSI_RE.sub("", pane)
    if dialog_type(plain):
        print(f"REFUSED {win}: a dialog is open ({dialog_command(plain)})")
        return 3
    if pending := unsent_input(pane):
        print(f"REFUSED {win}: input box holds unsent text: {pending[:60]}")
        return 3
    text = a.text
    t = f.target(win)
    if len(text) > PASTE_MIN or "\n" in text:
        buf = f"fleet-wave-{os.getpid()}"
        f.tmux("set-buffer", "-b", buf, "--", text)
        f.tmux("paste-buffer", "-p", "-d", "-b", buf, "-t", t)
    else:
        f.tmux("send-keys", "-t", t, "-l", "--", text)
    # An Enter arriving with the text is swallowed by the paste: pause first.
    time.sleep(a.pause)
    f.tmux("send-keys", "-t", t, "Enter")
    print(f"sent: {win}")
    return 0


def _fill_footer(text: str, fills: dict[str, str]) -> str:
    missing = sorted({m for m in PLACEHOLDER_RE.findall(text) if not fills.get(m)})
    if missing:
        raise SystemExit(
            "fleet-wave: footer placeholders unfilled: "
            + ", ".join(missing)
            + " (flags --coordinator/--release-owner/--approval or FLEET_* env)"
        )
    return PLACEHOLDER_RE.sub(lambda m: fills[m.group(1)], text)


def _legacy_footers() -> bool:
    return os.environ.get("FLEET_LEGACY_FOOTERS", "") == "1"


def _footer_template(f: Fleet, tier: str) -> str:
    """Generated tier footer plus the builder workflow; the pre-v2
    footers/<readonly|qland>.txt under FLEET_LEGACY_FOOTERS=1."""
    fleet = f.repo / ".claude" / "fleet"
    if _legacy_footers():
        legacy = fleet / "footers" / f"{LEGACY_FOOTER[tier]}.txt"
        return legacy.read_text(encoding="utf-8")
    gen = (fleet / "generated" / f"footer-{tier}.txt").read_text(encoding="utf-8")
    body = [ln for ln in gen.splitlines() if not ln.startswith("# GENERATED")]
    workflow = (fleet / "builder-workflow.txt").read_text(encoding="utf-8")
    return "\n".join(body) + "\n" + workflow


def _ensure_tree(f: Fleet, name: str) -> Path:
    """Create the builder worktree before the agent starts, the way
    ``claude -w`` and scripts/devin-tree would (both reuse an existing
    tree), so its permission files are in place at session start."""
    tree = f.worktrees / name
    if tree.is_dir():
        return tree
    branch = f"worktree-{name}"
    git("fetch", "-q", "origin", "main", cwd=f.main)
    has_branch = subprocess.run(
        ["git", "show-ref", "-q", "--verify", f"refs/heads/{branch}"],
        cwd=f.main,
        check=False,
    )
    add = ["git", "worktree", "add", "-q"]
    add += (
        [str(tree), branch]
        if has_branch.returncode == 0
        else ["-b", branch, str(tree), "origin/main"]
    )
    proc = subprocess.run(
        add, cwd=f.main, capture_output=True, text=True, encoding="utf-8", check=False
    )
    if proc.returncode:
        raise SystemExit(f"fleet-wave: worktree add {name}: {proc.stderr.strip()}")
    return tree


def _apply_authority(tier: str, tree: Path) -> str:
    """Run scripts/fleet-authority apply on ``tree``; return its Codex flags."""
    argv = [str(SCRIPTS / "fleet-authority"), "apply", "--tier", tier]
    argv += ["--to", str(tree), "--yes"]
    proc = subprocess.run(
        argv,
        cwd=SCRIPTS.parent,
        capture_output=True,
        text=True,
        encoding="utf-8",
        check=False,
    )
    if proc.returncode:
        raise SystemExit(f"fleet-wave: authority apply refused: {proc.stderr.strip()}")
    flags = [ln for ln in proc.stdout.splitlines() if ln.startswith("codex-flags: ")]
    return flags[-1].removeprefix("codex-flags: ") if flags else ""


def cmd_launch(f: Fleet, a: argparse.Namespace) -> int:
    name = a.name
    if not re.fullmatch(r"[A-Za-z0-9._-]+", name):
        print(f"fleet-wave: bad tree name {name!r}", file=sys.stderr)
        return 2
    win = ("devin-" if a.devin else "claude-") + name
    if win in f.windows():
        print(f"exists: {win}")
        return 0
    tier = TIER_ALIASES.get(a.tier, a.tier)
    fills = {
        "coordinator": a.coordinator or os.environ.get("FLEET_COORDINATOR", ""),
        "release_owner": a.release_owner or os.environ.get("FLEET_RELEASE_OWNER", ""),
        "approval": a.approval or os.environ.get("FLEET_APPROVAL", ""),
    }
    footer = _fill_footer(_footer_template(f, tier), fills)
    brief = Path(a.brief).read_text(encoding="utf-8").rstrip("\n")
    text = f"{brief}\n{footer}"
    f.state.mkdir(parents=True, exist_ok=True)
    prompt_file = f.state / f"{win}.prompt"
    prompt_file.write_text(text, encoding="utf-8")
    codex = ""
    if a.dry_run:
        print(f"fleet-authority apply --tier {tier} --to {f.worktrees / name} --yes")
    elif not _legacy_footers():
        codex = _apply_authority(tier, _ensure_tree(f, name))
    if a.devin:
        argv = [
            str(f.repo / "scripts" / "devin-tree"),
            name,
            "--prompt-file",
            str(prompt_file),
            # Reto ruling me499092: wave Devins always launch in bypass.
            "--permission-mode",
            LAUNCH_PERMISSION_MODE["devin"],
        ]
        if f.session:
            argv += ["--session", f.session]
        if f.model_set:
            argv += ["--model", f.model_set]
        if a.dry_run:
            print(" ".join(argv))
            return 0
        return subprocess.run(argv, check=False).returncode
    tmux_argv = ["new-window", "-d", "-n", win, "-c", str(f.main)]
    if f.session:
        tmux_argv[2:2] = ["-t", f"={f.session}:"]
    claude = [
        "claude",
        "-w",
        name,
        "--model",
        f.model,
        "--permission-mode",
        LAUNCH_PERMISSION_MODE["claude"],
        text,
    ]
    if a.dry_run:
        print(" ".join([*tmux_argv, "--", *claude[:-1], f"<{len(text)} chars>"]))
        return 0
    f.tmux(*tmux_argv, "--", *claude)
    print(f"launched {win} ({f.model}, tier {tier}, prompt {prompt_file})")
    if codex:
        print(f"codex flags for tier {tier}: {codex}")
    return 0


def _finish_line(f: Fleet, name: str) -> str:
    purpose = f.worktrees / name / ".claude" / "purpose"
    try:
        lines = purpose.read_text(encoding="utf-8").splitlines()
    except OSError:
        return ""
    hits = [ln for ln in lines if ln.startswith(f"{name} at ")]
    return hits[-1] if hits else ""


def _ticks(a: argparse.Namespace) -> Iterable[int]:
    return range(a.ticks) if a.ticks else itertools.count()


def cmd_watch_done(f: Fleet, a: argparse.Namespace) -> int:
    seen_dir = f.state / "watch-seen"
    seen_dir.mkdir(parents=True, exist_ok=True)
    for k, _ in enumerate(_ticks(a)):
        if k:
            time.sleep(a.interval)
        for name in a.names:
            line = _finish_line(f, name)
            seen = seen_dir / name
            old = seen.read_text(encoding="utf-8") if seen.exists() else ""
            if line and line != old:
                seen.write_text(line, encoding="utf-8")
                print(f"DONE {utc_stamp()} {line[:160]}", flush=True)
    return 0


def _stuck_state(f: Fleet, name: str, win: str | None, quiet_s: int) -> str:
    pane = f.capture(win) if win else None
    if pane is None:
        return "gone"
    st_dir = f.state / "stuck-seen"
    digest = hashlib.sha256(pane.encode("utf-8")).hexdigest()[:12]
    hfile, tfile = st_dir / f"{name}.hash", st_dir / f"{name}.t"
    now = int(time.time())
    if not hfile.exists() or hfile.read_text(encoding="utf-8") != digest:
        hfile.write_text(digest, encoding="utf-8")
        tfile.write_text(str(now), encoding="utf-8")
    since = int(tfile.read_text(encoding="utf-8") or now) if tfile.exists() else now
    age = now - since
    if dialog_type(pane):
        return "prompt: " + dialog_command(pane)[:80]
    if _finish_line(f, name):
        return "done"
    if age > quiet_s:
        return f"quiet{age // 600 * 10}m"
    return "ok"


def cmd_stuck_watch(f: Fleet, a: argparse.Namespace) -> int:
    st_dir = f.state / "stuck-seen"
    st_dir.mkdir(parents=True, exist_ok=True)
    for k, _ in enumerate(_ticks(a)):
        if k:
            time.sleep(a.interval)
        names = a.names or [w.split("-", 1)[1] for w in f.wave_windows()]
        for name in names:
            win = f.resolve(name)
            st = _stuck_state(f, name, win, a.quiet_min * 60)
            sfile = st_dir / f"{name}.st"
            old = sfile.read_text(encoding="utf-8") if sfile.exists() else ""
            if st != old:
                sfile.write_text(st, encoding="utf-8")
                if st != "ok":
                    print(f"{utc_stamp()} {win or name} {st}", flush=True)
    return 0


def cmd_close(f: Fleet, a: argparse.Namespace) -> int:
    log = f.state / "dropped-branches.log"
    f.state.mkdir(parents=True, exist_ok=True)
    rc = 0
    wins = f.windows()
    # Best effort: a stale origin/main only errs towards refusing.
    git("fetch", "origin", cwd=f.main)
    for name in a.names:
        tree = f.worktrees / name
        if not tree.is_dir():
            for p in PREFIXES:
                if p + name in wins:
                    f.tmux("kill-window", "-t", f.target(p + name), check=False)
                    print(f"window closed: {p}{name}")
            print(f"no tree: {name}")
            continue
        if not git("rev-parse", "--verify", "origin/main", cwd=tree).strip():
            print(f"REFUSED {name}: origin/main does not resolve; kept")
            rc = 1
            continue
        # Landed = the scripts/branch-audit test (gr497896): ancestry, an
        # empty tree diff against origin/main, per-commit patch equivalence,
        # a squash-subject match, or a clean reverse-apply. A multi-commit
        # squash-land trips the plain `git cherry` check that used to gate
        # this, so a landed tree looked unlanded.
        if Auditor is not None:
            audit = Auditor(tree).audit("HEAD")
            landed = audit.verdict == "landed"
        else:
            cherry = git("cherry", "origin/main", "HEAD", cwd=tree).splitlines()
            landed = all(ln.startswith("-") for ln in cherry)
        if not landed:
            print(f"REFUSED {name}: unlanded commits; window, tree, branch kept")
            print(git("log", "--oneline", "origin/main..HEAD", cwd=tree), end="")
            rc = 1
            continue
        for p in PREFIXES:
            if p + name in wins:
                f.tmux("kill-window", "-t", f.target(p + name), check=False)
                print(f"window closed: {p}{name}")
        branch = git("rev-parse", "--abbrev-ref", "HEAD", cwd=tree).strip()
        sha = git("rev-parse", "--short", "HEAD", cwd=tree).strip()
        git("worktree", "unlock", str(tree), cwd=f.main)
        proc = subprocess.run(
            ["git", "worktree", "remove", str(tree)],
            cwd=f.main,
            capture_output=True,
            text=True,
            encoding="utf-8",
            check=False,
        )
        if proc.returncode:
            print(f"KEPT (dirty or busy): {name}: {proc.stderr.strip()[:120]}")
            rc = 1
            continue
        dropped = ""
        if branch and branch != "HEAD":
            d = subprocess.run(
                ["git", "branch", "-d", branch],
                cwd=f.main,
                capture_output=True,
                text=True,
                encoding="utf-8",
                check=False,
            )
            # -d proves ancestry only; a squash- or patch-landed branch still
            # fails it, and the audit above already proved the work is on main.
            if d.returncode:
                d = subprocess.run(
                    ["git", "branch", "-D", branch],
                    cwd=f.main,
                    capture_output=True,
                    text=True,
                    encoding="utf-8",
                    check=False,
                )
            dropped = branch if d.returncode == 0 else ""
        with log.open("a", encoding="utf-8") as fh:
            fh.write(f"{utc_stamp()} {name} {branch} {sha} removed\n")
        print(
            f"removed: {name} at {sha}"
            + (f"; branch deleted: {dropped}" if dropped else "")
        )
    return rc


def parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="fleet-wave",
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    p.add_argument("--session", default="")
    p.add_argument("--socket", default="")
    p.add_argument("--worktrees", default="")
    p.add_argument("--model", default="")
    sub = p.add_subparsers(dest="cmd", required=True)

    s = sub.add_parser("launch", help="start a builder window")
    s.add_argument("name")
    s.add_argument("--brief", required=True)
    s.add_argument("--tier", choices=("R", "W", *TIER_ALIASES), default="R")
    s.add_argument("--devin", action="store_true")
    s.add_argument("--coordinator", default="")
    s.add_argument("--release-owner", default="")
    s.add_argument("--approval", default="")
    s.add_argument("--dry-run", action="store_true")
    s.set_defaults(fn=cmd_launch)

    s = sub.add_parser("peek", help="last lines of each window")
    s.add_argument("win", nargs="*")
    s.add_argument("-n", type=int, default=int(os.environ.get("PEEK_N", "6")))
    s.set_defaults(fn=cmd_peek)

    sub.add_parser("ctx", help="context %% per Claude pane").set_defaults(fn=cmd_ctx)
    sub.add_parser("prompts", help="panes at a dialog").set_defaults(fn=cmd_prompts)

    s = sub.add_parser("send-line", help="type TEXT and press Enter")
    s.add_argument("win")
    s.add_argument("text")
    s.add_argument("--pause", type=float, default=1.0, help=argparse.SUPPRESS)
    s.set_defaults(fn=cmd_send_line)

    for cmd, fn, nargs in (
        ("watch-done", cmd_watch_done, "+"),
        ("stuck-watch", cmd_stuck_watch, "*"),
    ):
        s = sub.add_parser(cmd)
        s.add_argument("names", nargs=nargs)
        s.add_argument("--interval", type=float, default=60.0)
        s.add_argument("--ticks", type=int, default=0, help="stop after N polls")
        if cmd == "stuck-watch":
            s.add_argument("--quiet-min", type=int, default=20)
        s.set_defaults(fn=fn)

    s = sub.add_parser(
        "close",
        help="kill window, remove tree, drop branch; refuses unlanded work",
    )
    s.add_argument("names", nargs="+")
    s.set_defaults(fn=cmd_close)

    s = sub.add_parser(
        "approve",
        help="append a grant to the approval ledger (coordinator tree only)",
        description="Append a grant (or, with --revoke, a revoke) to "
        "$FLEET_STATE/approvals/<tree>.tsv. Refuses unless run from "
        "FLEET_COORDINATOR_TREE — a convention-level guard; anyone with "
        "shell access could append directly, so it holds because the "
        "coordinator session's own rules forbid it writing a grant the "
        "user did not type (the coordinator transcript shows which lines "
        "the user typed). A grant relayed in a session message is not an "
        "approval; .claude/fleet/authority.md still governs.",
    )
    s.add_argument("tree", nargs="?")
    s.add_argument("shas", nargs="?")
    s.add_argument("--note", default="")
    s.add_argument("--list", action="store_true", help="print the ledger")
    s.add_argument("--revoke", action="store_true", help="append a revoke line")
    s.set_defaults(fn=cmd_approve)

    s = sub.add_parser(
        "approved",
        help="exit 0 iff every sha has a grant and no later revoke",
    )
    s.add_argument("tree")
    s.add_argument("shas")
    s.set_defaults(fn=cmd_approved)
    return p


def main(argv: Sequence[str] | None = None) -> int:
    a = parser().parse_args(argv)
    return int(a.fn(Fleet(a), a))


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