#!/usr/bin/env -S uv run --no-project --python >=3.12 python
"""scripts/round — the shared tally for a coordinated peer round.

A round is one pass of: peers dogfood on prod, fix what they find, qland;
the coordinator gates the integrated main, deploys, and opens the next round.
The expensive part used to be the bookkeeping: every "have you landed?" and
every "qlanded <sha>" was a cross-session message, and a message wakes a
whole session on both ends. This file-backed tally replaces those messages.
A peer marks itself from its own tree; the coordinator reads the table.

State lives under the repo's COMMON git dir (`<git-common-dir>/precis-round/`),
next to the ship lock, so every worktree of this checkout sees the same round
and nothing is tracked or pushed.

  scripts/round open [--base <sha>]   coordinator: start the next round
  scripts/round in <sha> [note]       peer: I landed <sha> this round (repeatable)
  scripts/round none [note]           peer: nothing to land this round
  scripts/round eta <text>            peer: still working, here is when/what
  scripts/round status [--json]       the table: every worktree and its mark
  scripts/round pending               worktrees with no final mark, one per line
  scripts/round close                 coordinator: end the round
  scripts/round gate [--json]         coordinator: exact release head verdict, else newest green main
  scripts/round deploy [--dry-run]    coordinator: deploy pinned, tag/forward/retire a release
  scripts/round cut [--sha S] [--dry-run]   coordinator: cut release/r<N> at the candidate (or S)
  scripts/round cut --abandon         coordinator: delete the open release branch (every commit on main)

A peer is named by its worktree directory (`PRECIS_ROUND_PEER` overrides).
Marks carry the round number, so a mark left from an earlier round reads as
"no mark" — `open` never has to clean up after anyone. `in` and `none` are
final for the round; `eta` is not, so an `eta` peer stays in `pending`.

There is no local gate and no ship lock in a round any more. Every main push
gets a full GitHub CI verdict (check.yml no longer cancels main runs), so the
candidate is simply the newest main sha whose shards all went green
(`scripts/last-gated-main-sha`). `gate` reports it and how much unverified
main sits above it; `deploy` refuses unless it is still an ancestor of
origin/main, fresh enough, and a fast-forward of `gated` AND `prod`, then deploys exactly
that sha with `--pinned`. An exact certified release may instead retain a
descendant `gated` already contained in main: the newer gate remains valid,
while the frozen payload keeps its own exact certificate and prod guards.
Neither verb takes the ship lock or runs a test, so qlands never stall behind
a round. Peers do not deploy — the coordinator
runs `deploy`.

`cut` freezes what the round ships: it pushes `release/r<N>` at the candidate
(the sha `gate` names, or `--sha`), records it in round.json and prints the
peers' marked shas the cut does not contain (`late: <peer> <sha>`). check.yml
runs the full gate shape on every `release/**` push. It refuses while any
release branch exists on origin, off main's first-parent line, or when the cut
tree carries a new duplicate migration number. While recorded, gate/deploy
read only its exact remote head's full CI certificate and age, bound to one
check.yml run/attempt with distinct required jobs: no earlier green SHA or
main fallback. The head need not yet be
on main (an interrupted release fix can still need its forward merge).

A separate advisory lifecycle lock serializes round mutations and release
fix final sections; main lands never take it. Kernel ownership avoids age
steals during slow rollout, and the deploy child inherits ownership if the
controller dies. The ship lock remains narrow and independent. A pinned
deployment journal survives partial failure. Release deploy requires the
existing deploy's health/convergence path, zero exit, fresh success marker
and exact origin/prod, then awaits the coordinator's actual live runtime
observation (exit 3).
`--confirm-runtime SHA --runtime-evidence TEXT` records that observation and
publishes immutable deployed/r<N>, forward-merges
with a CAS when needed and lease-deletes only the inspected head. Clear state
last: retry cannot substitute newer main, overwrite a tag or lose a late fix.
Completed rounds are terminal; immutable receipts/r<N>.json retain CI/runtime
evidence across round open. Successful archive publication commits completion;
an interrupted final state write reconciles the original journal before any
later receipt refresh, preserving evidence without a new runtime claim.
Invalid lifecycle state refuses instead of using
main fallback. Runtime confirmation binds the exact receipt generation:
a later same-SHA install needs a new live observation before closeout.
Close/abandon refuse a partial deployment. The coordinator still verifies the
live product after rollout; installed metadata alone cannot prove a running
process reloaded. Main gates retain local pins but defer global gated
publication while the release owns it. No live health result is inferred
from these tooling tests.

Round numbers are global: `open` takes max(local n, highest deployed/r<N> tag
or release/r<N> branch on origin) + 1, and `cut` refuses if deployed/r<N> exists.

Exit codes: 0 ok; 3 rollout returned success, runtime confirmation pending;
1 refused (no round open, unknown sha, bad usage). `pending`
exits 0 with no output when everyone is in.
"""

from __future__ import annotations

import argparse
import fcntl
import json
import math
import os
import re
import runpy
import subprocess
import sys
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, cast

FINAL = ("in", "none")
# check.yml `plan`'s docs-lane classifier: a commit touching only paths that
# match this is docs-only. A test pins that the workflow still says the same.
DOCS_ONLY_RE = r"^(docs/|[^/]+\.md$)"
# How old the candidate's CI verdict may be before `deploy` refuses it. Reto's
# ruling on the right window is pending; only this default may change.
DEFAULT_MAX_CANDIDATE_HOURS = 6.0
_SHA = re.compile(r"^[0-9a-f]{7,40}$")


def _git(*args: str) -> str | None:
    try:
        cp = subprocess.run(
            ["git", *args],
            capture_output=True,
            text=True,
            encoding="utf-8",
            check=False,
            timeout=60,
        )
    except (OSError, subprocess.SubprocessError):
        return None
    return cp.stdout.strip() if cp.returncode == 0 else None


def _state_dir() -> Path:
    common = _git("rev-parse", "--path-format=absolute", "--git-common-dir")
    if not common:
        sys.exit("scripts/round: not inside a git checkout")
    return Path(common) / "precis-round"


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


def _read(path: Path) -> dict[str, Any] | None:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    return data if isinstance(data, dict) else None


def _write(path: Path, data: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    tmp.write_text(json.dumps(data, indent=1) + "\n", encoding="utf-8")
    os.replace(tmp, path)


def _read_round(state: Path) -> dict[str, Any]:
    """Only absence permits legacy main fallback; broken lifecycle state refuses."""
    reader = cast(
        Callable[[Path], dict[str, Any]],
        runpy.run_path(str(Path(__file__).resolve().parent / "lib/round_state.py"))[
            "read_round"
        ],
    )
    try:
        return reader(state / "round.json")
    except (OSError, ValueError) as exc:
        sys.exit(f"scripts/round: unreadable/invalid round state — refusing: {exc}")


def _archive_receipt(state: Path, cur: dict[str, Any]) -> None:
    """Exclusive atomic publication retains immutable evidence across round open."""
    if not cur.get("deployed"):
        return
    receipt = {"n": cur["n"], "base": cur.get("base", ""), "deployed": cur["deployed"]}
    path = state / "receipts" / f"r{cur['n']}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    try:
        tmp.write_text(json.dumps(receipt, indent=1) + "\n", encoding="utf-8")
        try:
            os.link(tmp, path)
        except FileExistsError:
            if _read(path) != receipt:
                sys.exit(
                    f"scripts/round: immutable receipt {path.name} differs or is unreadable"
                )
    finally:
        tmp.unlink(missing_ok=True)


def _completed_deployment(cur: dict[str, Any]) -> dict[str, Any]:
    rel = cur["release"]
    return {
        **rel["deployment"],
        "tag": f"deployed/r{cur['n']}",
        "branch": rel["branch"],
        "at": rel["deployment"]["runtime_confirmed_at"],
    }


def _reconcile_completion_archive(
    state: Path, cur: dict[str, Any], *, dry_run: bool
) -> int | None:
    """Archive publication commits completion, before any later install repair."""
    path = state / "receipts" / f"r{cur['n']}.json"
    try:
        archived = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        if not path.is_symlink():
            return None
        return _deploy_refuse("immutable completion archive is a dangling link")
    except (OSError, ValueError):
        return _deploy_refuse("immutable completion archive is unreadable")
    if not cur["release"].get("deployment", {}).get("runtime_confirmed_at"):
        return _deploy_refuse(
            "immutable completion archive conflicts with the retained journal"
        )
    expected = {
        "n": cur["n"],
        "base": cur.get("base", ""),
        "deployed": _completed_deployment(cur),
    }
    if archived != expected:
        return _deploy_refuse(
            "immutable completion archive conflicts with the retained journal"
        )
    if dry_run:
        print(
            "dry-run: would reconcile archived terminal completion; no deploy or new runtime claim"
        )
        return 0
    cur["deployed"] = archived["deployed"]
    del cur["release"]
    _write(state / "round.json", cur)
    print(
        f"round {cur['n']} completion already archived at {cur['deployed']['sha'][:9]}; "
        "reconciled terminal state; no new deploy performed"
    )
    return 0


def _peer() -> str:
    override = os.environ.get("PRECIS_ROUND_PEER", "").strip()
    if override:
        return override
    top = _git("rev-parse", "--show-toplevel")
    if not top:
        sys.exit("scripts/round: not inside a git checkout")
    return Path(top).name


def _worktrees() -> list[str]:
    """Names of every linked worktree — the primary checkout is not a peer."""
    out = _git("worktree", "list", "--porcelain") or ""
    paths = [
        ln.split(" ", 1)[1] for ln in out.splitlines() if ln.startswith("worktree ")
    ]
    return sorted(Path(p).name for p in paths[1:])


def _open_round(state: Path) -> dict[str, Any] | None:
    cur = _read_round(state)
    return cur if cur and cur.get("open") else None


def _marks(state: Path, n: int) -> dict[str, dict[str, Any]]:
    marks: dict[str, dict[str, Any]] = {}
    for path in sorted((state / "marks").glob("*.json")):
        mark = _read(path)
        if mark and mark.get("round") == n:
            marks[path.stem] = mark
    return marks


def _require_round(state: Path) -> dict[str, Any]:
    cur = _open_round(state)
    if cur is None:
        sys.exit(
            "scripts/round: no round is open — nothing to mark (the coordinator runs `scripts/round open`)"
        )
    return cur


def _origin_round_max(root: Path) -> int | None:
    """Highest N among origin's deployed/r<N> tags and release/r<N> branches.

    None = origin unreadable. Numbering is global on origin, not per checkout.
    """
    out = _rgit(
        root,
        "ls-remote",
        "origin",
        "refs/tags/deployed/r*",
        "refs/heads/release/r*",
    )
    if out is None:
        return None
    nums = [
        int(m.group(1))
        for ln in out.splitlines()
        if (m := re.search(r"\trefs/(?:tags/deployed|heads/release)/r(\d+)$", ln))
    ]
    return max(nums, default=0)


def cmd_open(state: Path, args: argparse.Namespace) -> int:
    prev = _read_round(state)
    if prev.get("open") or prev.get("release"):
        print(
            f"scripts/round: round {prev['n']} is still open or has a release — close it first",
            file=sys.stderr,
        )
        return 1
    taken = _origin_round_max(_root())
    if taken is None:
        print(
            "scripts/round: open refused — could not read origin's deployed/release"
            " refs (git ls-remote failed); round numbers are global on origin",
            file=sys.stderr,
        )
        return 1
    _archive_receipt(state, prev)
    base = args.base or _git("rev-parse", "origin/prod") or ""
    cur = {
        "n": max(int(prev.get("n", 0)), taken) + 1,
        "open": True,
        "opened": _now(),
        "by": _peer(),
        "base": base,
    }
    _write(state / "round.json", cur)
    print(f"round {cur['n']} open (base {base[:8] or 'unknown'})")
    return 0


def cmd_close(state: Path, _args: argparse.Namespace) -> int:
    cur = _require_round(state)
    if cur.get("release"):
        sys.exit(
            "scripts/round: close refused — deploy/retire or abandon the open release first"
        )
    cur.update({"open": False, "closed": _now()})
    _write(state / "round.json", cur)
    print(f"round {cur['n']} closed")
    return 0


def cmd_mark(state: Path, args: argparse.Namespace) -> int:
    cur = _require_round(state)
    peer = _peer()
    path = state / "marks" / f"{peer}.json"
    mark = _read(path) or {}
    if mark.get("round") != cur["n"]:
        mark = {"round": cur["n"], "shas": []}
    if args.cmd == "in":
        if not _SHA.match(args.sha):
            print(f"scripts/round: {args.sha!r} is not a commit sha", file=sys.stderr)
            return 1
        full = _git("rev-parse", "--verify", "--quiet", f"{args.sha}^{{commit}}")
        if not full:
            print(
                f"scripts/round: {args.sha} is not a commit in this checkout — fetch, or check the sha",
                file=sys.stderr,
            )
            return 1
        if full not in mark["shas"]:
            mark["shas"].append(full)
    note = " ".join(args.note).strip()
    if args.cmd == "eta" and not note:
        print("scripts/round: eta needs text — what remains, and when", file=sys.stderr)
        return 1
    mark.update({"status": args.cmd, "note": note, "at": _now()})
    _write(path, mark)
    shas = " ".join(s[:8] for s in mark["shas"])
    print(
        f"round {cur['n']}: {peer} → {args.cmd}{' ' + shas if shas else ''}{' — ' + note if note else ''}"
    )
    return 0


def _rows(state: Path, cur: dict[str, Any]) -> list[dict[str, Any]]:
    marks = _marks(state, cur["n"])
    rows = []
    for name in sorted(set(_worktrees()) | set(marks)):
        mark = marks.get(name, {})
        rows.append(
            {
                "peer": name,
                "status": mark.get("status", ""),
                "shas": mark.get("shas", []),
                "note": mark.get("note", ""),
                "at": mark.get("at", ""),
            }
        )
    return rows


def cmd_status(state: Path, args: argparse.Namespace) -> int:
    cur = _open_round(state)
    if cur is None:
        last = _read_round(state)
        if args.json:
            print(json.dumps({"round": last, "peers": []}))
        else:
            print(f"no round open (last: {last['n']})" if last else "no round open")
        return 0
    rows = _rows(state, cur)
    if args.json:
        print(json.dumps({"round": cur, "peers": rows}))
        return 0
    print(
        f"round {cur['n']} — opened {cur['opened']} by {cur['by']}, base {cur.get('base', '')[:8] or 'unknown'}"
    )
    rel = cur.get("release")
    if rel:
        hrs = _hours_since(rel.get("cut_at", ""))
        age = f"{hrs:.1f}h" if hrs is not None else "unknown age"
        head = _remote(_root(), f"refs/heads/{rel['branch']}")
        print(
            f"  release {rel['branch']} at {(head or rel['base'])[:9]}, cut {age} ago"
        )
        if head is None:
            print("  release head unreadable on origin")
        elif not head:
            print("  release absent on origin — retirement record remains")
        deployment = rel.get("deployment")
        if deployment:
            print(
                f"  deployment pinned {deployment['sha'][:9]} ({'runtime confirmed by coordinator; retirement pending' if deployment.get('runtime_confirmed_at') else 'rollout returned success; runtime confirmation pending' if deployment.get('rollout_at') else 'incomplete; resume same SHA'})"
            )
    width = max([len(r["peer"]) for r in rows] + [4])
    for r in rows:
        shas = " ".join(s[:8] for s in r["shas"])
        tail = " — ".join(x for x in (shas, r["note"]) if x)
        print(
            f"  {r['peer']:<{width}}  {r['status'] or '—':<4}  {r['at'][11:16] + 'Z' if r['at'] else '     '}  {tail}".rstrip()
        )
    return 0


def cmd_pending(state: Path, _args: argparse.Namespace) -> int:
    cur = _require_round(state)
    for r in _rows(state, cur):
        if r["status"] not in FINAL:
            print(r["peer"])
    return 0


def _root() -> Path:
    """The checkout whose scripts/ and refs the gate verbs use.

    PRECIS_ROUND_ROOT selects the payload checkout, including its CI helper,
    pinned deploy subprocess/cwd, refs, ship pin and common-dir state. A
    separately reviewed maintenance controller can target a frozen payload;
    its strict state reader stays adjacent to this controller. Otherwise the
    target is the checkout this script lives in.
    """
    override = os.environ.get("PRECIS_ROUND_ROOT", "").strip()
    return Path(override) if override else Path(__file__).resolve().parent.parent


def _rgit(root: Path, *args: str) -> str | None:
    return _git("-C", str(root), *args)


def _round_state(root: Path) -> Path:
    common = _rgit(root, "rev-parse", "--path-format=absolute", "--git-common-dir")
    if not common:
        sys.exit("scripts/round: not inside a git checkout")
    return Path(common) / "precis-round"


@contextmanager
def _lifecycle(root: Path) -> Iterator[int]:
    """Independent of the ship lock; kernel releases on crash, no age steals."""
    path = _round_state(root).parent / "precis-round-lifecycle.lock"
    with path.open("a") as lock:
        try:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            sys.exit(
                "scripts/round: release lifecycle busy — retry after the current operation completes"
            )
        yield lock.fileno()


def _remote(root: Path, ref: str) -> str | None:
    """None = unreadable, empty = absent. Never trust a stale tracking ref."""
    out = _rgit(root, "ls-remote", "origin", ref)
    if out is None:
        return None
    return next((ln.split()[0] for ln in out.splitlines() if ln.split()[1] == ref), "")


def _release_candidate(root: Path, rel: dict[str, Any]) -> dict[str, Any]:
    branch = rel["branch"]
    head = _remote(root, f"refs/heads/{branch}")
    facts: dict[str, Any] = {
        "candidate": "",
        "candidate_state": "release_missing",
        "age_hours": None,
        "release_branch": branch,
        "release_head": head or "",
    }
    if head is None:
        facts["candidate_state"] = "unreadable"
        return facts
    if not head:
        return facts
    if _rgit(root, "fetch", "-q", "origin", f"refs/heads/{branch}") is None or not _ref(
        root, f"{head}^{{commit}}"
    ):
        facts["candidate_state"] = "unreadable"
        return facts
    try:
        cp = subprocess.run(
            [str(root / "scripts/last-gated-main-sha"), "--exact", head],
            cwd=root,
            capture_output=True,
            text=True,
            encoding="utf-8",
            check=False,
            timeout=60,
        )
        verdict = json.loads(cp.stdout)
        if cp.returncode or not isinstance(verdict, dict):
            raise ValueError
        if verdict.get("candidate") not in ("", head):
            raise ValueError
        facts.update(verdict)
    except (OSError, ValueError, subprocess.SubprocessError):
        facts["candidate_state"] = "unreadable"
    return facts


def _ref(root: Path, name: str) -> str:
    return _rgit(root, "rev-parse", "--verify", "--quiet", name) or ""


def _ancestor(root: Path, older: str, newer: str) -> bool:
    cp = subprocess.run(
        ["git", "-C", str(root), "merge-base", "--is-ancestor", older, newer],
        capture_output=True,
        check=False,
    )
    return cp.returncode == 0


def _lgm(root: Path, *flags: str) -> str:
    """Output of scripts/last-gated-main-sha (a black box; empty = unknowable)."""
    return _lgm_run(root, *flags)[0]


def _lgm_run(root: Path, *flags: str) -> tuple[str, int]:
    """(stdout, exit code). Exit 2 + empty = the whole window had no green sha."""
    argv = [
        str(root / "scripts" / "last-gated-main-sha"),
        *flags,
        "--start",
        "origin/main",
    ]
    try:
        cp = subprocess.run(
            argv,
            cwd=root,
            capture_output=True,
            text=True,
            encoding="utf-8",
            check=False,
        )
    except OSError:
        return "", 1
    return cp.stdout.strip(), cp.returncode


def _fetch(root: Path) -> None:
    """Refresh origin's main/gated/prod; a failure is a warning, not fatal."""
    if _rgit(root, "fetch", "-q", "origin", "main", "gated", "prod") is not None:
        return
    # A ref absent on origin fails the combined fetch: retry one by one.
    got = [
        _rgit(root, "fetch", "-q", "origin", r) is not None
        for r in ("main", "gated", "prod")
    ]
    if not got[0]:
        print(
            "scripts/round: warning: fetch of origin/main failed — using local refs",
            file=sys.stderr,
        )


def _docs_only(root: Path, sha: str) -> bool:
    files = (_rgit(root, "diff", "--name-only", f"{sha}^", sha) or "").splitlines()
    return all(re.match(DOCS_ONLY_RE, f) for f in files)


def _red_above(root: Path, cand: str) -> dict[str, Any] | None:
    """Newest of the 10 main commits above `cand` whose verdict is complete and failed."""
    jq = r'.check_runs[] | [.name, .status, (.conclusion // ""), .html_url] | @tsv'
    for sha in (
        _rgit(root, "rev-list", "--first-parent", "-n10", f"{cand}..origin/main") or ""
    ).split():
        try:
            cp = subprocess.run(
                [
                    "gh",
                    "api",
                    f"repos/{{owner}}/{{repo}}/commits/{sha}/check-runs",
                    "--jq",
                    jq,
                ],
                cwd=root,
                capture_output=True,
                text=True,
                encoding="utf-8",
                check=False,
                timeout=60,
            )
        except (OSError, subprocess.SubprocessError):
            return None
        if cp.returncode != 0:
            return None
        rows = [ln.split("\t") for ln in cp.stdout.splitlines() if ln.count("\t") == 3]
        rows = [r for r in rows if r[0] == "lint" or r[0].startswith("test-linux")]
        if not rows or any(r[1] != "completed" for r in rows):
            continue
        failed = [r for r in rows if r[2] in ("failure", "timed_out")]
        if failed:
            m = re.search(r"/actions/runs/(\d+)", failed[0][3])
            return {
                "sha": sha,
                "jobs": [r[0] for r in failed],
                "url": failed[0][3],
                "run_id": m.group(1) if m else "",
            }
    return None


def _candidate(root: Path) -> tuple[str, str]:
    """(candidate sha, state) after a fetch; state is found|none_in_window|unreadable."""
    _fetch(root)
    sha, rc = _lgm_run(root)
    cand = _ref(root, f"{sha}^{{commit}}") if sha else ""
    # exit 2 + empty = walked the whole window, none green; any other empty
    # answer (gh/auth/API trouble) is "could not look".
    state = (
        "found" if cand else "none_in_window" if (rc == 2 and not sha) else "unreadable"
    )
    return cand, state


def _refresh_release_pointers(
    root: Path, facts: dict[str, Any], observed_gated: list[str]
) -> bool:
    """Read actual refs, never stale tracking refs, around release publication.

    Retention is release-only and uses full DAG ancestry. A descendant gate
    certifies later main without changing the independently certified payload.
    Normal publication still uses the existing non-forced fast-forward helper;
    a rejected racing push is a refusal, never permission to install anyway.
    Keep authoritative gated observations as a lower bound across refreshes
    and unfinished install journals: deletion/regression must not authorize
    recreating an older gate. Cached tracking refs never establish this bound.
    """
    cand = facts["candidate"]
    for name in ("main", "gated", "prod"):
        sha = _remote(root, f"refs/heads/{name}")
        if sha is None:
            return False
        if sha and not _ref(root, f"{sha}^{{commit}}"):
            if _rgit(root, "fetch", "-q", "origin", name) is None or not _ref(
                root, f"{sha}^{{commit}}"
            ):
                return False
        if name == "gated":
            for prior in observed_gated:
                if not isinstance(prior, str) or (
                    prior and (len(prior) != 40 or not _SHA.fullmatch(prior))
                ):
                    return False
                if prior and (not sha or not _ancestor(root, prior, sha)):
                    return False
            observed_gated[:] = [sha]
        facts[f"origin_{name}"] = sha
    if not facts["origin_main"]:
        return False
    gated, main, prod = (facts[f"origin_{name}"] for name in ("gated", "main", "prod"))
    facts["candidate_in_main"] = bool(cand) and _ancestor(root, cand, main)
    facts["gated_is_ancestor"] = not gated or _ancestor(root, gated, cand)
    facts["gated_is_descendant_in_main"] = (
        bool(facts.get("release_branch") and cand and gated)
        and not facts["gated_is_ancestor"]
        and _ancestor(root, cand, gated)
        and _ancestor(root, gated, main)
    )
    facts["prod_is_ancestor"] = not prod or _ancestor(root, prod, cand)
    return True


def _gate_facts(root: Path) -> dict[str, Any]:
    cur = _read_round(_round_state(root))
    rel = cur.get("release")
    if rel:
        _fetch(root)
        release_facts = _release_candidate(root, rel)
        cand, state, age = (
            release_facts[k] for k in ("candidate", "candidate_state", "age_hours")
        )
    else:
        release_facts = {}
        cand, state = _candidate(root)
        try:
            age = float(_lgm(root, "--age-hours")) if cand else None
        except ValueError:
            age = None
    facts: dict[str, Any] = {
        "candidate": cand,
        "candidate_state": state,
        "subject": (_rgit(root, "log", "-1", "--format=%s", cand) or "")
        if cand
        else "",
        "age_hours": age,
        "origin_main": _ref(root, "refs/remotes/origin/main"),
        "origin_gated": _ref(root, "refs/remotes/origin/gated"),
        "origin_prod": _ref(root, "refs/remotes/origin/prod"),
        "candidate_in_main": False,
        "gated_is_ancestor": True,
        "gated_is_descendant_in_main": False,
        "prod_is_ancestor": True,
        "ahead_docs": 0,
        "ahead_code": 0,
        "red": None,
    }
    facts.update(release_facts)
    if rel:
        if rel.get("cut_pending"):
            facts.update(candidate="", candidate_state="unreadable")
            return facts
        if not cur.get("open") or rel.get("branch") != f"release/r{cur.get('n')}":
            facts.update(candidate="", candidate_state="unreadable")
            return facts
        journal = rel.get("deployment", {})
        observed_gated = [
            journal[k] for k in ("gated_before", "gated_after") if k in journal
        ]
        if not _refresh_release_pointers(root, facts, observed_gated):
            facts.update(candidate="", candidate_state="unreadable")
            return facts
    if not cand:
        return facts
    facts["candidate_in_main"] = _ancestor(root, cand, facts["origin_main"])
    if facts["origin_gated"]:
        facts["gated_is_ancestor"] = _ancestor(root, facts["origin_gated"], cand)
    if facts["origin_prod"]:
        facts["prod_is_ancestor"] = _ancestor(root, facts["origin_prod"], cand)
    if facts["candidate_in_main"]:
        above = (
            _rgit(root, "rev-list", "--first-parent", f"{cand}..origin/main") or ""
        ).split()
        facts["ahead_docs"] = sum(_docs_only(root, s) for s in above)
        facts["ahead_code"] = len(above) - facts["ahead_docs"]
        if above:
            facts["red"] = _red_above(root, cand)
    return facts


_NO_CANDIDATE = {
    "none_in_window": "no main sha with a green CI verdict in the last 48 h",
    "unreadable": "could not read CI verdicts (gh/auth/API) — no candidate",
    "release_missing": "recorded release is missing on origin — no main fallback; resume retirement or investigate",
    "release_not_green": "exact release head has no full green CI verdict — no earlier SHA fallback",
}


def cmd_gate(args: argparse.Namespace) -> int:
    f = _gate_facts(_root())
    if args.json:
        print(json.dumps(f))
    elif not f["candidate"]:
        print(_NO_CANDIDATE[f["candidate_state"]])
    else:
        if f.get("release_branch"):
            print(f"release {f['release_branch']} head {f['release_head'][:8]}")
        age = f"{f['age_hours']:.1f}h" if f["age_hours"] is not None else "unknown"
        print(f"candidate {f['candidate'][:8]} {f['subject']}")
        print(f"verdict age {age}")
        print(
            f"origin/main {f['origin_main'][:8] or '—'}  origin/gated {f['origin_gated'][:8] or '—'}"
            f"  origin/prod {f['origin_prod'][:8] or '—'}"
        )
        if not f["candidate_in_main"]:
            print(
                "candidate is NOT an ancestor of origin/main"
                + (
                    " (final forward merge required)"
                    if f.get("release_branch")
                    else " (deploy will refuse)"
                )
            )
        print(
            f"main is ahead of the candidate by {f['ahead_docs']} docs-only"
            f" and {f['ahead_code']} code commit(s)"
        )
        if f.get("gated_is_descendant_in_main"):
            print(
                "origin/gated is a descendant contained in main (deploy will retain it)"
            )
        elif not f["gated_is_ancestor"]:
            print(
                "origin/gated is not an ancestor of the candidate (deploy will refuse)"
            )
        red = f["red"]
        if red:
            print(
                f"red above the candidate: {red['sha'][:8]} — failing: {', '.join(red['jobs'])}"
            )
            print(f"  {red['url']}")
            if red["run_id"]:
                print(f"  gh run view {red['run_id']} --log-failed")
    return 0 if f["candidate"] else 1


def _refusal(root: Path, f: dict[str, Any]) -> str:
    """Why `deploy` must not go ahead on these facts; empty when it may."""
    cand = f["candidate"]
    if not cand:
        return _NO_CANDIDATE.get(f["candidate_state"], _NO_CANDIDATE["unreadable"])
    if not f["candidate_in_main"] and not f.get("release_branch"):
        return f"candidate {cand[:8]} is not an ancestor of origin/main"
    age = f["age_hours"]
    if not isinstance(age, (int, float)) or not math.isfinite(age) or age < 0:
        return f"the age of candidate {cand[:8]}'s verdict is unknown"
    try:
        max_h = float(
            os.environ.get("PRECIS_ROUND_MAX_CANDIDATE_HOURS", "")
            or DEFAULT_MAX_CANDIDATE_HOURS
        )
    except ValueError:
        return "PRECIS_ROUND_MAX_CANDIDATE_HOURS must be a positive finite number"
    if not math.isfinite(max_h) or max_h <= 0:
        return "PRECIS_ROUND_MAX_CANDIDATE_HOURS must be a positive finite number"
    if age > max_h:
        return (
            f"candidate {cand[:8]}'s verdict is {age:.1f}h old"
            f" (limit {max_h:g}h, PRECIS_ROUND_MAX_CANDIDATE_HOURS)"
        )
    if not f["gated_is_ancestor"] and not (
        f.get("release_branch") and f.get("gated_is_descendant_in_main")
    ):
        return (
            f"origin/gated {f['origin_gated'][:8]} is not an ancestor of candidate"
            f" {cand[:8]} (gated moves fast-forward only)"
        )
    if not f["prod_is_ancestor"]:
        return f"origin/prod {f['origin_prod'][:8]} is not an ancestor of candidate {cand[:8]} (prod moves fast-forward only)"
    if f.get("release_branch"):
        certificate = f.get("certificate")
        if (
            not isinstance(certificate, dict)
            or certificate.get("sha") != cand
            or certificate.get("workflow") != ".github/workflows/check.yml"
            or not certificate.get("run_id")
            or not certificate.get("attempt")
        ):
            return "exact release head lacks its CI workflow/run/attempt certificate"
        rel = (_read_round(_round_state(root))).get("release", {})
        if not _ancestor(root, rel.get("base", ""), cand):
            return "release head is not a descendant of its recorded cut SHA"
    pin = root / ".ship-sha"
    pinned = pin.read_text(encoding="utf-8").strip() if pin.is_file() else ""
    if pinned and pinned != cand:
        return f".ship-sha here pins {pinned[:8]}, not {cand[:8]} — remove it or run from a clean tree"
    return ""


def _deploy_refuse(why: str) -> int:
    print(
        f"scripts/round: deploy refused — {why}; release record retained, re-run after resolving",
        file=sys.stderr,
    )
    return 1


def _release_unchanged(root: Path, branch: str, sha: str) -> bool:
    return _remote(root, f"refs/heads/{branch}") == sha


def _deploy_receipt(root: Path, sha: str, since: int = 0) -> dict[str, Any] | None:
    """Exact local install generation plus prod, never service-health evidence."""
    common = _round_state(root).parent
    try:
        with (common / "precis-deploy-state").open() as marker:
            before = os.fstat(marker.fileno())
            content = marker.read()
            after = os.fstat(marker.fileno())
        fields = content.split()
        if (
            len(fields) >= 3
            and fields[0] == sha
            and int(fields[1]) >= since
            and fields[2] == "success"
            and before.st_mtime_ns == after.st_mtime_ns
            and before.st_size == after.st_size
            and not (common / "precis-deploy-attempt").exists()
            and _remote(root, "refs/heads/prod") == sha
        ):
            return {
                "content": content,
                "mtime_ns": after.st_mtime_ns,
                "inode": after.st_ino,
            }
    except (OSError, ValueError):
        pass
    return None


def _deploy_receipt_matches(root: Path, sha: str, since: int = 0) -> bool:
    return _deploy_receipt(root, sha, since) is not None


def _confirmed_receipt_matches(root: Path, journal: dict[str, Any]) -> bool:
    receipt = _deploy_receipt(root, journal["sha"], journal["started_epoch"])
    return receipt is not None and receipt == journal.get("runtime_receipt")


def _confirmation_head_matches(root: Path, cur: dict[str, Any]) -> bool:
    """Missing branch is recoverable only with prior proof and completed refs."""
    rel = cur["release"]
    journal = rel["deployment"]
    sha = journal["sha"]
    head = _remote(root, f"refs/heads/{rel['branch']}")
    if head == sha:
        return True
    if head != "" or not any(
        isinstance(proof, dict)
        and proof.get("runtime_confirmed_at")
        and proof.get("runtime_evidence")
        and proof.get("runtime_receipt")
        for proof in journal.get("superseded_runtime", [])
    ):
        return False
    if (
        _remote(root, f"refs/tags/deployed/r{cur['n']}") != sha
        or not _deploy_receipt_matches(root, sha, journal["started_epoch"])
        or _rgit(root, "fetch", "-q", "origin", "main") is None
    ):
        return False
    main = _remote(root, "refs/heads/main")
    return bool(main and _ancestor(root, sha, main))


def _finish_release(root: Path, state: Path, cur: dict[str, Any]) -> int:
    """Idempotent tag → final forward merge → leased delete → record clear."""
    rel = cur["release"]
    sha = rel["deployment"]["sha"]
    branch, tag = rel["branch"], f"deployed/r{cur['n']}"
    if not _confirmed_receipt_matches(root, rel["deployment"]):
        return _deploy_refuse(
            "runtime-confirmed deployment receipt/prod no longer matches the pinned SHA"
        )
    head = _remote(root, f"refs/heads/{branch}")
    if head not in ("", sha):
        return _deploy_refuse("release head moved or could not be read after rollout")
    tagged = _remote(root, f"refs/tags/{tag}")
    if tagged is None or (tagged and tagged != sha):
        return _deploy_refuse(
            f"immutable {tag} differs from the pinned SHA or is unreadable"
        )
    if not head and (
        not rel["deployment"].get("runtime_confirmed_at") or tagged != sha
    ):
        return _deploy_refuse(
            "missing release lacks coordinator runtime-confirmed retirement evidence"
        )
    if (
        not tagged
        and _rgit(root, "push", "-q", "origin", f"{sha}:refs/tags/{tag}") is None
    ):
        return _deploy_refuse(f"creating immutable {tag} failed")
    for _ in range(5):
        if _rgit(root, "fetch", "-q", "origin", "main") is None:
            return _deploy_refuse("could not fetch main for final forward merge")
        main = _remote(root, "refs/heads/main")
        if not main or not _ref(root, f"{main}^{{commit}}"):
            return _deploy_refuse("main moved beyond fetched objects — retry closeout")
        if _ancestor(root, sha, main):
            break
        if not head:
            return _deploy_refuse(
                "release already deleted but pinned SHA is not on main"
            )
        tree = _rgit(root, "merge-tree", "--write-tree", main, sha)
        if tree is None:
            return _deploy_refuse(
                "final forward merge conflicts — coordinator must resolve; tag and release preserved"
            )
        merged = _rgit(
            root,
            "commit-tree",
            tree.splitlines()[0],
            "-p",
            main,
            "-p",
            sha,
            "-m",
            f"merge({branch}): deployed {sha[:9]}\n\nGate: release-forward",
        )
        if not merged:
            return _deploy_refuse("could not create final forward merge")
        if (
            _rgit(
                root,
                "push",
                "-q",
                f"--force-with-lease=refs/heads/main:{main}",
                "origin",
                f"{merged}:refs/heads/main",
            )
            is not None
        ):
            break
        if _remote(root, "refs/heads/main") == main:
            return _deploy_refuse("final main push refused (not a race)")
    else:
        return _deploy_refuse(
            "main kept moving during final forward merge (5 CAS attempts)"
        )
    if (
        _remote(root, f"refs/tags/{tag}") != sha
        or not _confirmed_receipt_matches(root, rel["deployment"])
        or _rgit(root, "fetch", "-q", "origin", "main") is None
        or not _ancestor(root, sha, _ref(root, "refs/remotes/origin/main"))
    ):
        return _deploy_refuse("prod/tag/main evidence changed before retirement")
    current = _remote(root, f"refs/heads/{branch}")
    if current not in ("", sha):
        return _deploy_refuse("release head changed before retirement")
    if (
        current
        and _rgit(
            root,
            "push",
            "-q",
            f"--force-with-lease=refs/heads/{branch}:{sha}",
            "origin",
            f":refs/heads/{branch}",
        )
        is None
    ):
        return _deploy_refuse("leased release retirement failed; tag/main preserved")
    if _remote(root, f"refs/heads/{branch}") != "":
        return _deploy_refuse("release deletion was not confirmed")
    cur["deployed"] = _completed_deployment(cur)
    _archive_receipt(state, cur)
    del cur["release"]
    _write(state / "round.json", cur)
    print(
        f"{branch} deployed at {sha[:9]}, tagged {tag}, forwarded to main and retired"
    )
    return 0


def cmd_deploy(args: argparse.Namespace, lock_fd: int) -> int:
    root = _root()
    state = _round_state(root)
    cur = _read_round(state)
    rel = cur.get("release")
    if not rel and cur.get("deployed"):
        print(
            f"round {cur['n']} already completed at {cur['deployed']['sha'][:9]}; "
            "no new deploy performed — close/open the next round"
        )
        return 0
    if rel and (not cur.get("open") or rel.get("branch") != f"release/r{cur.get('n')}"):
        return _deploy_refuse(
            "recorded release identity/open-round state is inconsistent"
        )
    if rel:
        recovered = _reconcile_completion_archive(state, cur, dry_run=args.dry_run)
        if recovered is not None:
            return recovered
    journal = rel.get("deployment", {}) if rel else {}
    if rel and journal.get("rollout_at"):
        sha = journal["sha"]
        receipt = _deploy_receipt(root, sha, journal["started_epoch"])
        if receipt is None:
            return _deploy_refuse(
                "rollout publication/marker no longer matches the pinned SHA"
            )
        if (
            journal.get("runtime_confirmed_at")
            and journal.get("runtime_receipt") != receipt
        ):
            if not args.dry_run:
                journal.setdefault("superseded_runtime", []).append(
                    {
                        k: journal.pop(k)
                        for k in (
                            "runtime_confirmed_at",
                            "runtime_evidence",
                            "confirmed_by",
                            "runtime_receipt",
                        )
                        if k in journal
                    }
                )
                _write(state / "round.json", cur)
        elif journal.get("runtime_confirmed_at"):
            if args.dry_run:
                print(
                    "dry-run: would resume runtime-confirmed release retirement without reinstalling"
                )
                return 0
            return _finish_release(root, state, cur)
        if args.confirm_runtime != sha or not args.runtime_evidence.strip():
            print(
                f"rollout {sha[:9]} returned success; runtime verification pending. Verify every required daemon's boot SHA/readiness and the session MCP, then run:\n"
                f"  scripts/round deploy --confirm-runtime {sha} --runtime-evidence '<observed service/boot-SHA/readiness evidence>'"
            )
            return 3
        if not _confirmation_head_matches(root, cur):
            return _deploy_refuse(
                "release head changed or missing retirement evidence before runtime confirmation"
            )
        if args.dry_run:
            print(
                "dry-run: would record coordinator runtime confirmation, tag/forward/retire"
            )
            return 0
        journal.update(
            runtime_confirmed_at=_now(),
            runtime_evidence=args.runtime_evidence.strip(),
            confirmed_by=_peer(),
            runtime_receipt=receipt,
        )
        _write(state / "round.json", cur)
        return _finish_release(root, state, cur)
    if args.confirm_runtime or args.runtime_evidence:
        return _deploy_refuse(
            "runtime confirmation requires an existing successful rollout journal"
        )
    if not rel:
        branches = _release_branches(root)
        if branches is None or branches:
            return _deploy_refuse(
                "origin has unrecorded release branches or could not be read; no main fallback"
            )
    f = _gate_facts(root)
    reason = _refusal(root, f)
    if reason:
        print(f"scripts/round: deploy refused — {reason}", file=sys.stderr)
        return 1
    cand = f["candidate"]
    if rel:
        pinned = rel.get("deployment", {}).get("sha")
        if pinned and pinned != cand:
            return _deploy_refuse(
                f"partial deployment pins {pinned[:9]}, release now names {cand[:9]}"
            )
        tag = _remote(root, f"refs/tags/deployed/r{cur['n']}")
        if tag is None or (tag and tag != cand):
            return _deploy_refuse("immutable deployed tag differs or is unreadable")
        if not _release_unchanged(root, rel["branch"], cand):
            return _deploy_refuse("release head moved during the CI lookup")
    # Release facts above came from the first authoritative remote snapshot.
    # Its bound persists even when a later snapshot still relates to candidate.
    observed_gated = [f["origin_gated"]] if rel else []
    if rel:
        if not _refresh_release_pointers(root, f, observed_gated):
            return _deploy_refuse(
                "release environment refs became unreadable or regressed before publication"
            )
        reason = _refusal(root, f)
        if reason:
            return _deploy_refuse(reason)
    retain_gated = bool(rel and f.get("gated_is_descendant_in_main"))
    if args.dry_run:
        if retain_gated:
            print(f"dry-run: would retain origin/gated at {f['origin_gated'][:8]}")
        else:
            print(
                f"dry-run: would move origin/gated to {cand[:8]} (env_pointer_move gated)"
            )
        print(f"dry-run: would run scripts/deploy {cand} --pinned")
        if rel:
            print(
                f"dry-run: would tag deployed/r{cur['n']}, forward merge and lease-retire {rel['branch']}"
            )
        return 0
    if rel:
        rel["deployment"] = {
            "sha": cand,
            "started_at": _now(),
            "ci": f.get("certificate"),
            "gated_before": f["origin_gated"],
            "gated_action": "retain" if retain_gated else "publish",
        }
        _write(state / "round.json", cur)
    if retain_gated:
        if os.environ.get("PRECIS_ENV_POINTERS", "1") == "0":
            return _deploy_refuse(
                "env_pointer_move gated disabled; cannot retain a release pointer"
            )
        print(
            f"retain origin/gated {f['origin_gated'][:8]} (certified release descendant in main)"
        )
    else:
        move = subprocess.run(
            [
                "bash",
                "-c",
                '. scripts/lib/env-pointers.sh && env_pointer_move "$1" gated "$2"',
                "_",
                str(root),
                cand,
            ],
            cwd=root,
            check=False,
        )
        if move.returncode != 0:
            print(
                f"scripts/round: deploy refused — env_pointer_move gated {cand[:8]} failed"
                f" (exit {move.returncode}: push refused or pointers disabled)",
                file=sys.stderr,
            )
            return 1
    if rel:
        if not _refresh_release_pointers(root, f, observed_gated):
            return _deploy_refuse(
                "release environment refs became unreadable or regressed after publication"
            )
        reason = _refusal(root, f)
        if reason:
            return _deploy_refuse(reason)
        if f["origin_gated"] != cand and not f["gated_is_descendant_in_main"]:
            return _deploy_refuse(f"origin/gated did not move to {cand[:8]}")
    else:
        _fetch(root)
        if _ref(root, "refs/remotes/origin/gated") != cand:
            print(
                f"scripts/round: deploy refused — origin/gated did not move to {cand[:8]}",
                file=sys.stderr,
            )
            return 1
    if rel and not _release_unchanged(root, rel["branch"], cand):
        return _deploy_refuse("release head changed before rollout")
    started = int(datetime.now(UTC).timestamp())
    if rel:
        rel["deployment"]["gated_after"] = f["origin_gated"]
        rel["deployment"]["started_epoch"] = started
        _write(state / "round.json", cur)
    rc = subprocess.run(
        [str(root / "scripts" / "deploy"), cand, "--pinned"],
        cwd=root,
        check=False,
        pass_fds=(lock_fd,),
    ).returncode
    if not rel or rc:
        return rc
    if not _release_unchanged(root, rel["branch"], cand):
        return _deploy_refuse(
            "release head changed during rollout; pinned SHA may be live"
        )
    if not _deploy_receipt_matches(root, cand, started):
        return _deploy_refuse(
            "deploy returned zero without matching fresh success marker and origin/prod"
        )
    rel["deployment"]["rollout_at"] = _now()
    _write(state / "round.json", cur)
    print(
        f"rollout {cand[:9]} returned success; runtime verification pending — verify required daemon boot SHA/readiness and session MCP, then `scripts/round deploy --confirm-runtime {cand} --runtime-evidence '<proof>'`"
    )
    return 3


MIG_DIR = "src/precis/migrations"
_MIG_NUM = re.compile(r"^(\d{4})_.*\.sql$")


def _migration_numbers(root: Path, rev: str) -> dict[str, list[str]]:
    """Migration number -> file names in `rev`'s tree (top level of MIG_DIR only)."""
    out = _rgit(root, "ls-tree", "-r", "--name-only", rev, "--", f"{MIG_DIR}/") or ""
    nums: dict[str, list[str]] = {}
    for rel in out.splitlines():
        parent, _, name = rel.rpartition("/")
        m = _MIG_NUM.match(name)
        if parent == MIG_DIR and m:
            nums.setdefault(m.group(1), []).append(name)
    return nums


def _migration_collisions(root: Path, sha: str) -> list[str]:
    """Numbers carried by more files in `sha` than in origin/prod (floor 1).

    Same rule as scripts/migration-check: a duplicate number that already
    shipped is accepted history, a new one is a collision. That script reads
    worktrees and branch heads, not an arbitrary tree, hence this.
    """
    prod = _ref(root, "refs/remotes/origin/prod")
    shipped = _migration_numbers(root, prod) if prod else {}
    bad = []
    for num, names in sorted(_migration_numbers(root, sha).items()):
        if len(names) > max(len(shipped.get(num, [])), 1):
            bad.append(f"{num}: {', '.join(sorted(names))}")
    return bad


def _release_branches(root: Path) -> list[str] | None:
    """Release branch names on origin; None when origin could not be read."""
    out = _rgit(root, "ls-remote", "--heads", "origin", "refs/heads/release/*")
    if out is None:
        return None
    return [ln.split("refs/heads/", 1)[1] for ln in out.splitlines() if "\trefs/" in ln]


def _on_first_parent(root: Path, sha: str) -> bool:
    line = _rgit(root, "rev-list", "--first-parent", "refs/remotes/origin/main") or ""
    return sha in line.split()


def _late(root: Path, state: Path, n: int, cut: str) -> list[str]:
    lines = []
    for peer, mark in _marks(state, n).items():
        for sha in mark.get("shas", []):
            if not _ref(root, f"{sha}^{{commit}}"):
                lines.append(f"late: {peer} {sha[:9]} (not fetched)")
            elif not _ancestor(root, sha, cut):
                lines.append(f"late: {peer} {sha[:9]}")
    return lines


def _hours_since(stamp: str) -> float | None:
    try:
        then = datetime.strptime(stamp, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=UTC)
    except ValueError:
        return None
    return (datetime.now(UTC) - then).total_seconds() / 3600


def _cut_refuse(why: str) -> int:
    print(f"scripts/round: cut refused — {why}", file=sys.stderr)
    return 1


def cmd_cut(state: Path, args: argparse.Namespace) -> int:
    root = _root()
    if args.abandon:
        return _cut_abandon(root, state, args)
    cur = _require_round(state)
    n = cur["n"]
    branch = f"release/r{n}"
    if cur.get("release"):
        rel = cur["release"]
        if rel.get("cut_pending"):
            tip = _remote(root, f"refs/heads/{rel['branch']}")
            if tip is None or (tip and tip != rel["base"]):
                return _cut_refuse(
                    "pending cut's remote SHA differs or is unreadable — coordinator must reconcile"
                )
            if args.sha and _ref(root, f"{args.sha}^{{commit}}") != rel["base"]:
                return _cut_refuse("pending cut pins a different SHA — resume it first")
            if args.dry_run:
                print(
                    f"dry-run: would reconcile pending cut {rel['branch']} at {rel['base'][:9]}"
                )
                return 0
            if (
                not tip
                and _rgit(
                    root,
                    "push",
                    "-q",
                    f"--force-with-lease=refs/heads/{rel['branch']}:",
                    "origin",
                    f"{rel['base']}:refs/heads/{rel['branch']}",
                )
                is None
            ):
                return _cut_refuse("pending cut push failed — intent retained")
            if _remote(root, f"refs/heads/{rel['branch']}") != rel["base"]:
                return _cut_refuse(
                    "pending cut remote SHA not confirmed — intent retained"
                )
            del rel["cut_pending"]
            _write(state / "round.json", cur)
            print(f"{rel['branch']} pending cut reconciled at {rel['base'][:9]}")
            return 0
        return _cut_refuse(
            f"round.json already records an open release ({cur['release']['branch']})"
            " — `scripts/round cut --abandon` first"
        )
    if cur.get("deployed"):
        return _cut_refuse(
            "this round already deployed — close and open the next round"
        )
    _fetch(root)
    taken = _remote(root, f"refs/tags/deployed/r{n}")
    if taken is None:
        return _cut_refuse("could not read origin's tags (git ls-remote failed)")
    if taken:
        return _cut_refuse(
            f"deployed/r{n} is already tagged on origin (another checkout used this"
            " number) — `scripts/round close` then `scripts/round open` to renumber"
        )
    existing = _release_branches(root)
    if existing is None:
        return _cut_refuse("could not read origin's branches (git ls-remote failed)")
    if existing:
        return _cut_refuse(
            f"release branch {', '.join(existing)} exists on origin — one release at a time"
        )
    if args.sha:
        sha = _ref(root, f"{args.sha}^{{commit}}")
        if not sha:
            return _cut_refuse(f"{args.sha} is not a commit in this checkout")
    else:
        sha, cstate = _candidate(root)
        if not sha:
            return _cut_refuse(_NO_CANDIDATE[cstate])
    if not _on_first_parent(root, sha):
        return _cut_refuse(f"{sha[:9]} is not on origin/main's first-parent line")
    for name in ("gated", "prod"):
        tip = _remote(root, f"refs/heads/{name}")
        if tip is None or (tip and not _ancestor(root, tip, sha)):
            return _cut_refuse(
                f"origin/{name} cannot fast-forward to the cut SHA — choose a compatible green main SHA"
            )
    collisions = _migration_collisions(root, sha)
    if collisions:
        return _cut_refuse(
            "migration number collision in the cut tree (renumber the unshipped one on main first):\n  "
            + "\n  ".join(collisions)
        )
    for line in _late(root, state, n, sha):
        print(line)
    if args.dry_run:
        print(f"dry-run: would push {sha[:9]} to origin as {branch}")
        return 0
    cur["release"] = {
        "branch": branch,
        "base": sha,
        "cut_at": _now(),
        "cut_pending": True,
    }
    _write(state / "round.json", cur)
    push = _rgit(
        root,
        "push",
        "-q",
        f"--force-with-lease=refs/heads/{branch}:",
        "origin",
        f"{sha}:refs/heads/{branch}",
    )
    if push is None:
        if _remote(root, f"refs/heads/{branch}") == sha:
            print(
                "scripts/round: cut push acknowledgement failed, remote SHA confirmed",
                file=sys.stderr,
            )
        else:
            return _cut_refuse(
                f"git push of {branch} failed; intent retained, re-run cut"
            )
    del cur["release"]["cut_pending"]
    _write(state / "round.json", cur)
    print(f"{branch} cut at {sha[:9]} — late work lands on main for round {n + 1}")
    print(
        f'fleet say -m "{branch} is cut at {sha[:9]} and frozen for round {n}: a release'
        f" fix goes in only when the coordinator asks, via scripts/ship --release from"
        f" a fresh tree cut from the release head (claude -w, or git worktree add"
        f" <path> origin/{branch}) holding only the fix; everything else qlands to"
        f' main for round {n + 1}"'
    )
    return 0


def _cut_abandon(root: Path, state: Path, args: argparse.Namespace) -> int:
    cur = _read_round(state)
    rel = cur.get("release")
    if not rel:
        return _cut_refuse(
            "no open release is recorded in round.json — nothing to abandon"
        )
    branch = rel["branch"]
    if rel.get("deployment"):
        return _cut_refuse(
            "deployment journal exists — resume `scripts/round deploy`, do not abandon partial rollout"
        )
    head = _remote(root, f"refs/heads/{branch}")
    if head is None:
        return _cut_refuse("could not read origin's branches (git ls-remote failed)")
    if not head:
        # Already gone from origin: only the record is left to clear.
        print(f"{branch} is not on origin — clearing the record")
        if not args.dry_run:
            del cur["release"]
            _write(state / "round.json", cur)
        return 0
    if (
        _rgit(root, "fetch", "-q", "origin", f"refs/heads/{branch}") is None
        or _rgit(root, "fetch", "-q", "origin", "main") is None
    ):
        return _cut_refuse(f"could not fetch {branch} and main from origin")
    if not _ancestor(root, head, _ref(root, "refs/remotes/origin/main")):
        unforwarded = _rgit(root, "log", "--oneline", f"origin/main..origin/{branch}")
        print(
            f"scripts/round: abandon refused — {branch} carries commits not on origin/main"
            " — forward them to main before abandoning:",
            file=sys.stderr,
        )
        print(unforwarded or "(could not list them)", file=sys.stderr)
        return 1
    if args.dry_run:
        print(f"dry-run: would delete {branch} on origin ({head[:9]} is on main)")
        return 0
    if (
        _rgit(
            root,
            "push",
            "-q",
            f"--force-with-lease=refs/heads/{branch}:{head}",
            "origin",
            f":refs/heads/{branch}",
        )
        is None
    ):
        return _cut_refuse(f"deleting {branch} on origin failed")
    del cur["release"]
    _write(state / "round.json", cur)
    print(
        f"{branch} abandoned ({head[:9]} is on main) — deleted on origin, record cleared"
    )
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="scripts/round", description="Shared tally for a coordinated peer round."
    )
    sub = parser.add_subparsers(dest="cmd", required=True)
    p_open = sub.add_parser("open")
    p_open.add_argument("--base", default="")
    sub.add_parser("close")
    p_in = sub.add_parser("in")
    p_in.add_argument("sha")
    p_in.add_argument("note", nargs="*")
    for name in ("none", "eta"):
        sub.add_parser(name).add_argument("note", nargs="*")
    sub.add_parser("status").add_argument("--json", action="store_true")
    sub.add_parser("pending")
    sub.add_parser("gate").add_argument("--json", action="store_true")
    p_deploy = sub.add_parser("deploy")
    p_deploy.add_argument("--dry-run", action="store_true")
    p_deploy.add_argument(
        "--confirm-runtime",
        default="",
        metavar="SHA",
        help="coordinator attests required live daemon boot SHA/readiness after rollout",
    )
    p_deploy.add_argument(
        "--runtime-evidence",
        default="",
        help="observed daemon and session MCP health evidence (no credentials)",
    )
    p_cut = sub.add_parser("cut")
    p_cut.add_argument("--sha", default="")
    p_cut.add_argument("--dry-run", action="store_true")
    p_cut.add_argument("--abandon", action="store_true")
    args = parser.parse_args(argv)
    if args.cmd == "gate":
        return cmd_gate(args)
    if args.cmd == "deploy":
        with _lifecycle(_root()) as lock_fd:
            return cmd_deploy(args, lock_fd)
    state = _state_dir()
    handlers = {
        "open": cmd_open,
        "close": cmd_close,
        "in": cmd_mark,
        "none": cmd_mark,
        "eta": cmd_mark,
        "status": cmd_status,
        "pending": cmd_pending,
        "cut": cmd_cut,
    }
    if args.cmd in ("open", "close", "cut", "in", "none", "eta"):
        # State is common-dir even when invoked by a linked peer worktree.
        with _lifecycle(Path.cwd()):
            return handlers[args.cmd](state, args)
    return handlers[args.cmd](state, args)


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