chemenu/tools/chemenu/commands/git_publish.py

"""Wrap the repo's publishing contract: stage everything, commit with a
templated message (auto-listing changed files), and push to a remote branch.
Exits non-zero on any failure so the caller must stop and ask the user - it
never force-pushes, and the one retry it does attempt (below) is bounded to a
single safe reconcile, never a loop.

Also implements the pull/rebase path (`reconcile`, exposed standalone as
`sync` and run automatically by `publish` before staging and again if the
push is rejected): fetch the remote, fast-forward or rebase when that cannot
collide with what this call is about to publish, and stop for review - the
same exit-42 idiom as the gate below - when it might. This is what keeps a
rejected push from stranding the commit that made it: a previous `publish`
whose push failed leaves a real, unpushed commit sitting on the branch, and
the next `publish` now pushes it instead of reporting "Nothing to commit"
forever.

Also implements the Mass-Update Gate (wiki/concepts/Mass-Update Gate.md):
a push to origin/main is the one action in this system with a real,
irreversible external effect (publicly visible commit history, possible CI
triggers, other clients pulling). Small/normal publishes (< threshold
counted files) go straight through, same as always - only publishes at or
above the threshold require human approval, so routine single-page
operations don't suffer "reviewer fatigue" from a gate that fires on every
commit. Paths under GATE_EXEMPT_PREFIXES are committed but not counted. The
gate is evaluated before anything is staged, so refusing a publish leaves
the working tree exactly as it was found.

**How the gate is cleared, and what that does and does not prove.** A tripped
gate exits `EXIT_NEEDS_CLEARANCE` (42) - its own code, distinct from a
validation error - and prints the counted file list plus the exact command to
re-run. That command carries `--confirm <token>`, where the token is a digest
of the file list and publish target (`changeset_token`). Two properties fall
out of that, and one deliberately does not:

* **Approval is bound to a changeset.** Touch one more file after the refusal
  and the token no longer matches; the gate asks again with the new list. The
  old `--yes` published whatever was in the tree at the time it ran, which is
  not necessarily what the human was shown.
* **Clearance is legible in a trace.** The refusal and the `--confirm` that
  follows it are separate, matchable events, so a scorer can ask whether the
  agent actually stopped in between (`evals/trajectory.py`).
* **It does not prove a human typed anything.** The token sits in the agent's
  own context; an agent that wants to bypass this can. That is a deliberate
  trade: enforcement here is cheap and never blocks legitimate work, and the
  "did a human really clear it?" question is answered in the eval layer
  instead of by making every publish an interactive ceremony.
"""
from __future__ import annotations

import hashlib
import json
import shlex
import subprocess
from dataclasses import dataclass, field
from typing import NamedTuple, Optional

import typer

from chemenu import config
from chemenu.commands._util import fail, needs_clearance, success
from chemenu.telemetry import emit

# "Farza's Rule" from the Mass-Update Gate concept: halt and request
# confirmation for operations affecting >= 10 pages.
DEFAULT_MASS_UPDATE_THRESHOLD = 10

# Path prefixes that are committed like everything else but do not count toward
# the gate. The gate's justification is that a push to origin/main publishes
# knowledge irreversibly; `work/` holds a run's working notes, which are tracked
# only so a multi-session task survives, and are deleted when it closes (see
# work/CONTRACT.md). Making a reviewer approve twelve scratch files is the
# review fatigue the threshold exists to avoid.
#
# Deliberately a constant and not an option: a `--gate-exempt` flag would be a
# gate an agent could open on its own initiative, which AGENTS.md forbids.
GATE_EXEMPT_PREFIXES = ("work/",)


def _run(args: list[str]) -> subprocess.CompletedProcess:
    return subprocess.run(args, cwd=config.ROOT, capture_output=True, text=True)


# --- Publish-Remote Gate -----------------------------------------------------
#
# The Mass-Update Gate asks "is this too much to publish?". This one asks the
# question underneath it: "is this the right place to publish to at all?".
#
# A checkout holding private content typically has two remotes - its own, and
# the public upstream it takes stack updates from. Nothing in git distinguishes
# them at push time, so a single wrong `--remote` puts a private corpus on a
# public repository, where a force-push does not take it back: the objects stay
# fetchable by SHA until someone expires the server's reflogs.
#
# Like the other two gates this refuses with exit 42 and has **no flag that
# opens it**. The way past it is to name the URL in the file, which is an edit
# the user makes deliberately rather than something an agent can decide mid-run.


def read_allowed_push_urls() -> Optional[list[str]]:
    """The URLs this checkout permits `publish` to push to, or None when the
    file is absent (unrestricted - see config.PUBLISH_REMOTES_FILENAME)."""
    path = config.ROOT / config.PUBLISH_REMOTES_FILENAME
    if not path.is_file():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        fail(
            f"{config.PUBLISH_REMOTES_FILENAME} is unreadable ({exc}). It decides where "
            "`publish` may push, so a broken one is not treated as 'no restriction' - "
            "fix the file or delete it deliberately."
        )
        return None
    urls = data.get("allowed_push_urls")
    if not isinstance(urls, list) or not all(isinstance(u, str) for u in urls):
        fail(
            f"{config.PUBLISH_REMOTES_FILENAME} has no usable `allowed_push_urls` list of "
            "strings. Expected: {\"schema\": 1, \"allowed_push_urls\": [\"<url>\"]}"
        )
        return None
    return urls


def push_url_for(remote: str) -> Optional[str]:
    """The URL `git push <remote>` would actually write to - `pushurl` when the
    remote sets one, otherwise its fetch URL. Reading the resolved value rather
    than the name is the whole point: a repointed `origin` must not pass."""
    result = _run(["git", "remote", "get-url", "--push", remote])
    if result.returncode != 0:
        return None
    return result.stdout.strip() or None


def publish_remote_refusal(remote: str, branch: str) -> Optional[str]:
    """The gate's message when `remote` is not an allowed push target, else None."""
    allowed = read_allowed_push_urls()
    if allowed is None:
        return None
    url = push_url_for(remote)
    if url is None:
        return (
            f"Publish-Remote Gate: '{remote}' resolves to no push URL, so this publish "
            f"cannot be checked against {config.PUBLISH_REMOTES_FILENAME}. Nothing was "
            "committed or pushed."
        )
    if url in allowed:
        return None
    listed = "\n".join(f"  - {u}" for u in allowed) or "  (the list is empty)"
    return (
        f"Publish-Remote Gate: this checkout does not allow publishing to '{remote}'.\n\n"
        f"  would push to: {url}\n"
        f"  allowed here:\n{listed}\n\n"
        "Nothing was committed or pushed. This checkout holds content that belongs to it "
        "alone, and a push to the wrong remote is not cheaply reversible - the objects stay "
        "fetchable by SHA even after a force-push, until the server's reflogs are expired.\n\n"
        "THE USER CANNOT SEE THIS OUTPUT. It went to your context, not to their screen.\n"
        "Show them the two lines above and stop. There is no flag that opens this gate: if "
        f"the target really is right, the user adds its URL to {config.PUBLISH_REMOTES_FILENAME} "
        "themselves. Do not edit that file to get past this."
    )


def parse_porcelain_entries(stdout: str) -> list[tuple[str, str]]:
    """Parse `git status --porcelain -z` output into (status_code, path) pairs.

    NUL-delimited output is used instead of line-splitting because it is the
    only form that survives paths containing spaces, quotes, or newlines
    (line mode quotes and escapes them instead). Rename/copy entries carry a
    second NUL-separated field with the original path; the new path is what
    gets committed, so the original is consumed and discarded.
    """
    fields = [f for f in stdout.split("\0") if f != ""]
    entries: list[tuple[str, str]] = []
    index = 0
    while index < len(fields):
        entry = fields[index]
        index += 1
        if len(entry) < 4:
            continue
        status_code, path = entry[:2], entry[3:]
        if status_code[0] in ("R", "C") or status_code[1] in ("R", "C"):
            index += 1  # skip the original path of a rename/copy
        entries.append((status_code, path))
    return entries


def parse_porcelain_z(stdout: str) -> list[str]:
    """Just the paths - what the gate counts and what `git add -A` would stage."""
    return [path for _, path in parse_porcelain_entries(stdout)]


def describe_status(status_code: str) -> str:
    """Porcelain XY code -> the word a reviewer needs. Deletions and additions
    are what a reader scans for first, so they must not be flattened into a
    generic "changed"."""
    if status_code == "??":
        return "added"
    if "D" in status_code:
        return "deleted"
    if "R" in status_code or "C" in status_code:
        return "renamed"
    if "A" in status_code:
        return "added"
    return "modified"


class FileChange(NamedTuple):
    """One path in the changeset, with enough detail to judge it without
    opening it. `added`/`removed` are -1 when git reports the file as binary
    or the churn could not be determined."""
    path: str
    status: str
    added: int
    removed: int
    digest: str

    @property
    def churn(self) -> int:
        return max(self.added, 0) + max(self.removed, 0)

    @property
    def churn_text(self) -> str:
        if self.status == "deleted":
            # The size of what is being removed, not just the fact of it: a
            # one-line stub and a 700-line document read identically as
            # "deleted", and they are not the same decision.
            return f"-{self.removed} deleted" if self.removed else "deleted"
        if self.added < 0:
            return "binary"
        return f"+{self.added}/-{self.removed}"


def _numstat(paths: list[str]) -> dict[str, tuple[int, int]]:
    """Lines added/removed per tracked file, against HEAD.

    `git diff HEAD` covers staged and unstaged changes together, which is what
    `publish` is about to commit. Untracked files are absent from it and are
    measured by reading them instead. A repository with no commits yet has no
    HEAD to diff against - that is the first-commit case from
    `instructions/setup-instance.md`, where everything is untracked anyway - so
    a failure here is normal and yields no entries rather than an error.

    `-z` is required, not a preference: without it git renders a path with any
    non-ASCII byte in quoted form ("kb/W\\303\\266rterbuch.md"), while
    `_changed_files` reads raw paths from `git status --porcelain -z`. The two
    then never match, the caller's lookup misses, and the file is measured as
    though git had never seen it - every line an addition, no removals. A
    rewritten page reported as a pure insertion hides exactly what a reviewer
    is being asked to approve.
    """
    result = _run(["git", "diff", "--numstat", "-z", "HEAD", "--", *paths])
    if result.returncode != 0:
        return {}
    stats: dict[str, tuple[int, int]] = {}
    # With -z each record is "added\tremoved\tpath" terminated by NUL. A rename
    # or copy leaves the path empty and follows with two more records, the old
    # and the new path; the new one is what `git status` reports.
    records = result.stdout.split("\0")
    index = 0
    while index < len(records):
        record = records[index]
        index += 1
        if not record:
            continue
        fields = record.split("\t")
        if len(fields) < 3:
            continue
        added, removed, path = fields[0], fields[1], fields[2]
        if not path:
            if index + 1 >= len(records):
                continue
            path = records[index + 1]
            index += 2
        # git writes "-" for both counts on a binary file.
        stats[path] = (-1, -1) if added == "-" else (int(added), int(removed))
    return stats


def _untracked_stat(path: str) -> tuple[int, int, str]:
    """(added, removed, digest) for a file git has never seen: every line is an
    addition. Anything unreadable as UTF-8 counts as binary rather than
    guessing at a line count."""
    full = config.ROOT / path
    try:
        data = full.read_bytes()
    except OSError:
        return (-1, -1, "")
    digest = hashlib.sha256(data).hexdigest()[:16]
    try:
        text = data.decode("utf-8")
    except UnicodeDecodeError:
        return (-1, -1, digest)
    return (len(text.splitlines()), 0, digest)


def _digest_of(path: str) -> str:
    """A short content digest of the working-tree file, or "" if it is gone.

    This is what binds a clearance to file *contents* and not merely to file
    *names*: without it, approving a list and then rewriting one of those files
    before confirming would still publish, which is the same "approved A,
    published B" hole the token exists to close.
    """
    try:
        return hashlib.sha256((config.ROOT / path).read_bytes()).hexdigest()[:16]
    except OSError:
        return ""


def collect_changes(paths: list[str]) -> list[FileChange]:
    """Every path `git add -A` would stage, with its status, churn and content
    digest. Ordered by path so the result is stable."""
    result = _run(["git", "status", "--porcelain", "-z", "-uall", "--", *paths])
    if result.returncode != 0:
        fail(f"git status failed:\n{result.stderr}")
    entries = parse_porcelain_entries(result.stdout)
    numstat = _numstat(paths)

    changes: list[FileChange] = []
    for status_code, path in entries:
        status = describe_status(status_code)
        if status == "deleted":
            # A deletion's churn is every line the file had, and git already
            # knows it. Short-circuiting to 0/0 here (the first version of this
            # code) silently understated the headline: one 718-line file went
            # out reported as `-174` against git's own `-891`, hiding four
            # fifths of the removals in the one direction a reviewer most needs
            # not understated. The digest stays empty - there is no content
            # left to fingerprint - which is itself what moves the token.
            _, removed = numstat.get(path, (0, 0))
            changes.append(FileChange(path, status, 0, max(removed, 0), ""))
        elif path in numstat:
            added, removed = numstat[path]
            changes.append(FileChange(path, status, added, removed, _digest_of(path)))
        else:
            added, removed, digest = _untracked_stat(path)
            changes.append(FileChange(path, status, added, removed, digest))
    return sorted(changes, key=lambda change: change.path)


def _changed_files(paths: list[str]) -> list[str]:
    """Every path `git add -A` would stage, including untracked files,
    optionally restricted to a pathspec."""
    result = _run(["git", "status", "--porcelain", "-z", "-uall", "--", *paths])
    if result.returncode != 0:
        fail(f"git status failed:\n{result.stderr}")
    return parse_porcelain_z(result.stdout)


def current_branch() -> Optional[str]:
    """The checked-out branch, or None in a detached HEAD / non-checkout."""
    result = _run(["git", "rev-parse", "--abbrev-ref", "HEAD"])
    if result.returncode != 0:
        return None
    name = result.stdout.strip()
    return None if not name or name == "HEAD" else name


def branch_mismatch_message(checked_out: Optional[str], branch: str) -> str:
    """The ERROR shown when the push target is not the checked-out branch.

    `git push <remote> <branch>` pushes the *ref named* `branch`, not `HEAD`.
    On a feature branch that silently pushes an unrelated (usually unchanged)
    ref and still exits 0, so the command reported "Published" while the commit
    it had just made stayed local - the same class of failure as an ignore rule
    that quietly un-publishes content.
    """
    where = f"branch `{checked_out}`" if checked_out else "a detached HEAD"
    return (
        f"Refusing to push `{branch}` while on {where}. `git push` would push the ref named "
        f"`{branch}`, not the commit just made, and would report success without publishing "
        "anything. Either check out the branch you mean to publish, or pass "
        f"`--branch {checked_out}` to push this one."
        if checked_out
        else (
            f"Refusing to push `{branch}` from {where}. Check out a branch first - pushing a "
            "named ref from a detached HEAD would publish something other than the commit "
            "just made."
        )
    )


# Files `wikitool` writes itself. They are committed and pushed like everything
# else, but they carry no decision: each is recomputable from the tree at any
# commit by `index rebuild` / `sources rebuild-index` / the log append that
# produced it. Nothing is lost by not reading them, and nothing is protected by
# making a reviewer approve them.
#
# They are therefore exempt from the gate's *count*, for the same reason `work/`
# is. A routine ingest touches five or six of these - kb/index.md, kb/log.md,
# kb/provenance.md and one INDEX.md per collection it wrote into - so counting
# them pushed ordinary single-source ingests over a threshold meant for mass
# updates. Three consecutive ingests on 2026-08-31 each stopped at the gate with
# 5-7 of their files generated; none of them was a mass update.
GENERATED_SUFFIXES = ("/INDEX.md",)
GENERATED_PATHS = ("kb/index.md", "kb/log.md", "kb/provenance.md")


def is_generated(path: str) -> bool:
    return path in GENERATED_PATHS or path.endswith(GENERATED_SUFFIXES)


def is_exempt(path: str, prefixes: tuple[str, ...] = GATE_EXEMPT_PREFIXES) -> bool:
    """Whether the Mass-Update Gate leaves this path out of its count."""
    return path.startswith(prefixes) or is_generated(path)


def counted_files(changed_files: list[str], prefixes: tuple[str, ...] = GATE_EXEMPT_PREFIXES) -> list[str]:
    """The subset of `changed_files` the Mass-Update Gate counts.

    Exempt paths are still staged, committed and pushed - only the threshold
    ignores them.
    """
    return [path for path in changed_files if not is_exempt(path, prefixes)]


def counted_files_of(
    changes: list[FileChange], prefixes: tuple[str, ...] = GATE_EXEMPT_PREFIXES
) -> list[FileChange]:
    """`counted_files` over the richer records. One exemption rule, applied in
    both shapes, so the count the gate enforces and the list it shows can never
    disagree."""
    return [change for change in changes if not is_exempt(change.path, prefixes)]


YES_REMOVED_MESSAGE = (
    "--yes no longer exists. The Mass-Update Gate is cleared with `--confirm <token>`, and the "
    "token comes from the gate's own refusal output - run this command without it first to see "
    "the file list and the exact line to re-run."
)


def _exempt_note(changed_files: list[str], counted: list[str]) -> str:
    """Account for every changed file the count leaves out, by reason.

    A reviewer who sees "11 counted" next to a 17-file commit needs the other
    six explained, or the number reads like a bug. Naming the two reasons
    separately also keeps them honestly distinct: `work/` is scratch state,
    generated files are recomputable output.
    """
    if len(changed_files) == len(counted):
        return ""
    scratch = sum(1 for p in changed_files if p.startswith(GATE_EXEMPT_PREFIXES))
    generated = sum(1 for p in changed_files if is_generated(p))
    reasons = []
    if scratch:
        prefixes = "/, ".join(p.rstrip("/") for p in GATE_EXEMPT_PREFIXES)
        reasons.append(f"{scratch} under {prefixes}/")
    if generated:
        reasons.append(f"{generated} generated by wikitool")
    return (
        f" ({len(changed_files)} changed in total; "
        f"{' and '.join(reasons)} committed but not counted)"
    )


# How the file list is grouped for review, most-consequential first. Mirrors
# the pipeline stages AGENTS.md defines rather than inventing a second taxonomy:
# a reader who knows the repo already knows these names. The `note` is what the
# group means for the *reviewer* - which is the part a bare path cannot say.
#
# Order matters and is not alphabetical: published knowledge and the control
# plane are what a wrong publish does lasting damage to, so they are read first
# and the mechanically-regenerated files are read last, if at all.
FILE_GROUPS: tuple[tuple[str, tuple[str, ...], str], ...] = (
    ("Published knowledge", ("kb/",), "goes public on push; what the wiki claims to know"),
    ("Agent control plane", ("AGENTS.md", "instructions/", "types/"),
     "changes how every future session behaves"),
    ("Source material", ("raw/",), "immutable inputs - changes here are unusual"),
    ("Tooling", ("tools/",), "the compiler itself"),
    ("Harness config", (".claude/", ".github/", ".vibe/", ".agents/"),
     "hooks and permissions - affects what the harness lets an agent do"),
    ("Human docs", ("README.md", "INSTALL.md", "EVALS.md", "CHANGES.md"), "prose, no runtime effect"),
    ("Workshop", ("work/",), "scratch state, deleted when the run closes"),
)
OTHER_GROUP = ("Other", (), "")


def group_of(path: str) -> tuple[str, str]:
    """(group title, reviewer note) for one path."""
    if is_generated(path):
        return ("Generated", "rebuilt by wikitool - no review needed")
    for title, prefixes, note in FILE_GROUPS:
        if path.startswith(prefixes):
            return (title, note)
    return (OTHER_GROUP[0], OTHER_GROUP[2])


def _plural(count: int, noun: str) -> str:
    return f"{count} {noun}" if count == 1 else f"{count} {noun}s"


def scale_line(changes: list[FileChange]) -> str:
    """One line answering "how big is this?" before any of the detail."""
    added = sum(max(c.added, 0) for c in changes)
    removed = sum(max(c.removed, 0) for c in changes)
    by_status: dict[str, int] = {}
    for change in changes:
        by_status[change.status] = by_status.get(change.status, 0) + 1
    breakdown = ", ".join(
        f"{by_status[status]} {status}"
        for status in ("added", "modified", "renamed", "deleted")
        if by_status.get(status)
    )
    return (
        f"Scale: {_plural(len(changes), 'file')}, +{added:,}/-{removed:,} lines ({breakdown})."
    )


def attention_notes(changes: list[FileChange]) -> list[str]:
    """The handful of facts that make one publish riskier than another.

    Deliberately not a score. Each line is a checkable statement about the
    changeset, so a reader can disagree with it by looking; a number would
    invite trusting it instead. Only lines that apply are emitted - a list of
    "0 deletions" reassurances is how a reviewer learns to skim.
    """
    notes: list[str] = []

    deleted = [c for c in changes if c.status == "deleted"]
    if deleted:
        shown = ", ".join(c.path for c in deleted[:3])
        more = f", +{len(deleted) - 3} more" if len(deleted) > 3 else ""
        notes.append(
            f"{_plural(len(deleted), 'file')} DELETED: {shown}{more} - a push makes that "
            "removal public"
        )

    control = [c for c in changes if group_of(c.path)[0] == "Agent control plane"]
    if control:
        notes.append(
            f"{_plural(len(control), 'file')} in the agent control plane "
            "(AGENTS.md/instructions/types) - changes behaviour in every future session"
        )

    harness = [c for c in changes if group_of(c.path)[0] == "Harness config"]
    if harness:
        notes.append(
            f"{_plural(len(harness), 'file')} of harness config - hooks and permissions decide "
            "what an agent is allowed to do without asking"
        )

    pages = [c for c in changes
             if c.path.startswith("kb/") and c.path.endswith(".md") and not is_generated(c.path)]
    if pages:
        notes.append(f"{_plural(len(pages), 'published wiki page')} changed")

    scored = [c for c in changes if c.churn > 0]
    if scored:
        largest = max(scored, key=lambda c: c.churn)
        if largest.churn >= 100:
            notes.append(f"largest single change: {largest.path} ({largest.churn_text})")

    binary = [c for c in changes if c.added < 0 and c.status != "deleted"]
    if binary:
        notes.append(
            f"{_plural(len(binary), 'binary file')} - contents cannot be reviewed as a diff"
        )
    return notes


def format_changes(changes: list[FileChange]) -> str:
    """The grouped file list. Every path still appears exactly once - grouping
    reorders and annotates, it never summarises anything away, because the
    complete list is the artifact the user is being asked to approve."""
    order = [title for title, _, _ in FILE_GROUPS] + ["Generated", OTHER_GROUP[0]]
    grouped: dict[str, list[FileChange]] = {}
    notes: dict[str, str] = {}
    for change in changes:
        title, note = group_of(change.path)
        grouped.setdefault(title, []).append(change)
        notes[title] = note

    width = max((len(c.path) for c in changes), default=0)
    blocks: list[str] = []
    for title in order:
        members = grouped.get(title)
        if not members:
            continue
        added = sum(max(c.added, 0) for c in members)
        removed = sum(max(c.removed, 0) for c in members)
        note = f" - {notes[title]}" if notes.get(title) else ""
        header = f"{title} ({_plural(len(members), 'file')}, +{added:,}/-{removed:,}){note}"
        lines = [
            f"  {c.status[0].upper()} {c.path.ljust(width)}  {c.churn_text}"
            for c in members
        ]
        blocks.append(header + "\n" + "\n".join(lines))
    return "\n\n".join(blocks)


def changeset_token(
    counted: list[FileChange], threshold: int, remote: str, branch: str, paths: list[str]
) -> str:
    """sha256 over the counted files - path *and* content digest - plus the
    publish target, cut to 12 hex chars. Same changeset, same token; anything
    else needs its own clearance.

    Content is part of the input, not just the file names. A token over names
    alone would let an approved list be published with different contents:
    clear a list, rewrite one of those files, confirm, and the user's approval
    would cover text they never saw. That is the same "approved A, published B"
    hole the token exists to close, one level down.

    `counted` is already filtered through `counted_files()`, so an exempt
    `work/` path changing does not move the token - the gate never asked about
    those files in the first place.

    Stateless on purpose: there is no ticket file, no TTL, and no session
    bookkeeping to get out of sync. The token *is* the record of what was
    shown, and it can be recomputed from the working tree at any time.
    """
    payload = json.dumps(
        {
            "counted": sorted((c.path, c.digest) for c in counted),
            "threshold": threshold,
            "remote": remote,
            "branch": branch,
            "paths": sorted(paths),
        },
        sort_keys=True,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]


def rerun_command(token: str, message: str, push: bool, threshold: int, remote: str,
                  branch: str, paths: list[str]) -> str:
    """The exact command line that clears this gate, ready to copy.

    Reconstructed rather than echoed from argv so it is correct regardless of
    how the original call was spelled, and quoted so a message with spaces
    survives. `--confirm` comes first so the whole line has a stable prefix a
    harness permission rule can match on (see .claude/settings.json)."""
    parts = ["tools/wikitool", "publish", "--confirm", token, "--message", message]
    if not push:
        parts.append("--no-push")
    if threshold != DEFAULT_MASS_UPDATE_THRESHOLD:
        parts += ["--threshold", str(threshold)]
    if remote != "origin":
        parts += ["--remote", remote]
    if branch != "main":
        parts += ["--branch", branch]
    for p in paths:
        parts += ["--path", p]
    return " ".join(shlex.quote(part) for part in parts)


def clearance_message(changes: list[FileChange], threshold: int, token: str, rerun: str,
                      remote: str, branch: str, stale_token: Optional[str] = None) -> str:
    """The whole user-facing artifact of this gate.

    Written to be shown to a human verbatim: what would happen, the evidence
    (every counted file), and the one line that proceeds. The agent-facing
    instruction layer deliberately holds none of this - a procedure written
    down in advance is a ritual an agent can perform without ever involving a
    human, which is what the three 2026-08 incidents all looked like.

    **On the wording of the instruction line.** It used to read "SHOW THIS
    OUTPUT TO THE USER", and the first agent to receive it answered with a file
    *count* and a pointer to "the output above" - because on a harness that
    runs `wikitool` through a shell tool, a command's stdout goes to the agent's
    context, not to the user's screen. Printing and showing are different acts
    there, and an instruction that conflates them reads as already satisfied the
    moment the text exists. So the line now names the act that is actually
    required: reproduce the list, in the reply, because the user cannot see
    this.
    """
    counted = counted_files_of(changes)
    exempt_note = _exempt_note([c.path for c in changes], [c.path for c in counted])
    notes = attention_notes(counted)
    attention = ("\n\nWorth a closer look:\n" + "\n".join(f"- {n}" for n in notes)) if notes else ""
    stale_line = (
        f"\nThe token you passed ({stale_token}) does not match this changeset. Files or their "
        "contents changed since it was issued, so that clearance does not carry - here is the "
        "current state.\n"
        if stale_token
        else ""
    )
    return (
        f"Mass-Update Gate: this publish would commit and push {len(counted)} counted files "
        f"(>= threshold {threshold}){exempt_note} to {remote}/{branch}. A push there is "
        f"immediately visible and not cheaply reversible.\n{stale_line}"
        f"\n{scale_line(counted)}{attention}\n"
        "\nTHE USER CANNOT SEE THIS OUTPUT. It went to your context, not to their screen.\n"
        f"Reproduce the {len(counted)}-file breakdown below in your reply - the groups, the "
        "paths, and the sizes - and stop there. A count, a summary, a description of the change, "
        'or a pointer to "the output above" is not the list, and leaves the user approving '
        "something they never saw. Run no further commands in this turn.\n\n"
        "Once they have replied approving this exact changeset, the line that publishes it is:\n\n"
        f"  {rerun}\n\n"
        f"The token {token} covers both the file list and the contents below; edit any of it and "
        "the token changes, and clearance is asked again.\n\n"
        f"CHANGES BY AREA ({len(counted)} files) - reproduce this in your reply:\n\n"
        f"{format_changes(counted)}"
    )


def remote_ref_exists(remote: str, branch: str) -> bool:
    """Whether `<remote>/<branch>` resolves at all - false for a remote that was never
    fetched, or a branch that has never been pushed (the very first `publish`)."""
    return _run(["git", "rev-parse", "--verify", "-q", f"{remote}/{branch}"]).returncode == 0


def fetch_remote(remote: str, branch: str) -> bool:
    """`git fetch <remote> <branch>`, true on success. A failure here (no remote configured,
    network/auth, or a branch that does not exist on the remote yet) is never fatal on its own -
    every caller falls back to today's behaviour and lets the eventual `git push` report the
    real error, so an offline or brand-new instance sees no new failure mode."""
    return _run(["git", "fetch", remote, branch]).returncode == 0


def divergence(local_ref: str, remote_ref: str) -> str:
    """Where `local_ref` stands relative to `remote_ref`: "up-to-date", "ff-possible" (remote
    only, local can fast-forward), "local-ahead" (local only, nothing to pull), or "diverged"
    (both sides have commits the other lacks - the race the rest of this module exists for)."""
    result = _run(["git", "rev-list", "--left-right", "--count", f"{local_ref}...{remote_ref}"])
    local_only, remote_only = (int(n) for n in result.stdout.split())
    if local_only == 0 and remote_only == 0:
        return "up-to-date"
    if local_only == 0:
        return "ff-possible"
    if remote_only == 0:
        return "local-ahead"
    return "diverged"


def merge_base(a: str, b: str) -> Optional[str]:
    result = _run(["git", "merge-base", a, b])
    return result.stdout.strip() if result.returncode == 0 else None


def touched_files(base: str, tip: str) -> set[str]:
    """The files that changed between `base` and `tip` - the mechanical proxy this module uses
    for "could these two commit ranges collide": disjoint file sets cannot produce a content
    conflict, so a rebase between them needs no human/LLM review, only overlapping ones do."""
    result = _run(["git", "diff", "--name-only", base, tip])
    return {line for line in result.stdout.splitlines() if line}


def oneline_log(base: str, tip: str) -> list[str]:
    result = _run(["git", "log", "--oneline", f"{base}..{tip}"])
    return [line for line in result.stdout.splitlines() if line]


def rebase_review_token(remote: str, branch: str, local_before: str, remote_tip: str,
                         overlap_files: list[str]) -> str:
    """sha256 over the exact upstream state and overlapping-file set a rebase-review gate was
    issued for, cut to 12 hex chars - same shape as `changeset_token`. Either side moving
    (remote gains another commit, or the overlap set changes) changes the token, so a stale
    `--confirm-rebase` is rejected the same way a stale Mass-Update `--confirm` already is."""
    payload = json.dumps(
        {
            "remote": remote, "branch": branch, "local_before": local_before,
            "remote_tip": remote_tip, "overlap_files": sorted(overlap_files),
        },
        sort_keys=True,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]


@dataclass
class ReconcileOutcome:
    """What happened when the local branch was brought up to date with the remote before a
    publish/sync. `status` is one of: "no-remote-or-fetch-failed", "up-to-date",
    "fast-forwarded", "local-ahead", "rebased", "needs-review", "conflict"."""
    status: str
    pulled_commits: list[str] = field(default_factory=list)
    overlap_files: list[str] = field(default_factory=list)
    overlap_diff: str = ""
    token: str = ""
    was_reviewed: bool = False
    detail: str = ""


def reconcile(remote: str, branch: str, confirm_rebase: Optional[str] = None) -> ReconcileOutcome:
    """Fetch `<remote>/<branch>` and reconcile the local `branch` with it: fast-forward when
    the remote is simply ahead, rebase local commits on top when both sides moved but touch
    disjoint files (a content conflict is then impossible by construction), and refuse - via
    the same exit-42 idiom as the Mass-Update Gate - when both sides touched the same file,
    until `confirm_rebase` matches the token that refusal issues. Read-only until the moment it
    actually rewrites history: an overlap that has not been cleared performs no rebase attempt
    at all, so a refused call leaves the branch exactly where it was found.

    Never commits, never pushes, never force-anything. Callers are `sync` (this is its whole
    job) and `publish` (proactively before staging, and once more if the eventual push is
    rejected - the real, narrow race this whole module exists to close)."""
    if not fetch_remote(remote, branch) or not remote_ref_exists(remote, branch):
        return ReconcileOutcome(status="no-remote-or-fetch-failed")

    remote_ref = f"{remote}/{branch}"
    state = divergence(branch, remote_ref)
    if state == "up-to-date":
        return ReconcileOutcome(status="up-to-date")

    if state == "ff-possible":
        pulled = oneline_log(branch, remote_ref)
        result = _run(["git", "merge", "--ff-only", remote_ref])
        if result.returncode != 0:
            # Only reachable if uncommitted local changes would be overwritten by the merge -
            # git itself refuses and touches nothing, so this is a safe abort, not a half-done
            # state.
            return ReconcileOutcome(status="conflict", detail=result.stderr)
        return ReconcileOutcome(status="fast-forwarded", pulled_commits=pulled)

    if state == "local-ahead":
        return ReconcileOutcome(status="local-ahead")

    # "diverged": both sides have unshared commits - the actual race.
    local_before = _run(["git", "rev-parse", branch]).stdout.strip()
    remote_tip = _run(["git", "rev-parse", remote_ref]).stdout.strip()
    base = merge_base(branch, remote_ref)
    if not base:
        return ReconcileOutcome(status="conflict", detail="no common history with remote")

    upstream_commits = oneline_log(base, remote_tip)
    overlap = sorted(touched_files(base, local_before) & touched_files(base, remote_tip))
    was_reviewed = False
    token = ""

    if overlap:
        token = rebase_review_token(remote, branch, local_before, remote_tip, overlap)
        if confirm_rebase != token:
            diff = _run(["git", "diff", base, remote_tip, "--", *overlap]).stdout
            return ReconcileOutcome(
                status="needs-review", pulled_commits=upstream_commits,
                overlap_files=overlap, overlap_diff=diff, token=token,
            )
        was_reviewed = True

    rebase_result = _run(["git", "rebase", remote_ref])
    if rebase_result.returncode != 0:
        _run(["git", "rebase", "--abort"])
        return ReconcileOutcome(status="conflict", detail=rebase_result.stderr)
    return ReconcileOutcome(
        status="rebased", pulled_commits=upstream_commits,
        overlap_files=overlap, was_reviewed=was_reviewed, token=token,
    )


def _local_ahead_of_remote(remote: str, branch: str) -> bool:
    """Whether `branch` currently has a commit `<remote>/<branch>` lacks - true right after a
    stranded commit from a previous failed push, and also right after `reconcile` rebases local
    work on top of a moved remote. Used to decide whether an otherwise-empty working tree still
    has something to push."""
    if not remote_ref_exists(remote, branch):
        return False
    result = _run(["git", "rev-list", "--count", f"{remote}/{branch}..{branch}"])
    return result.returncode == 0 and result.stdout.strip() not in ("", "0")


def rebase_review_message(outcome: ReconcileOutcome, remote: str, branch: str, command: str) -> str:
    """The whole user-facing artifact of the rebase-review gate, written in the same voice as
    `clearance_message`: what arrived, what it touches, the diff to actually read, and the exact
    line that proceeds once a human has seen it."""
    commits = "\n".join(f"  {c}" for c in outcome.pulled_commits) or "  (none)"
    files = "\n".join(f"  - {f}" for f in outcome.overlap_files)
    rerun = f"tools/wikitool {command} --confirm-rebase {outcome.token}"
    return (
        f"Rebase Review: {remote}/{branch} has moved, and the incoming commit(s) touch "
        f"{len(outcome.overlap_files)} file(s) this {command} is also changing. Rebasing would "
        "replay local work on top of them without anyone having read what changed - unlike a "
        "disjoint rebase, a content collision here is actually possible.\n\n"
        "THE USER CANNOT SEE THIS OUTPUT. It went to your context, not to their screen.\n\n"
        f"Commits arriving from {remote}/{branch}:\n{commits}\n\n"
        f"Files touched on both sides:\n{files}\n\n"
        f"Diff of those files as they stand on {remote}/{branch} since the common ancestor:\n\n"
        f"{outcome.overlap_diff}\n\n"
        "Read that diff, judge whether it conflicts logically with what you are about to "
        "publish, and summarize your judgment to the user - reproduce the file list above, not "
        'a count or "the output above" - before proceeding. Once they have agreed, the line '
        f"that continues is:\n\n  {rerun}\n\n"
        f"The token {outcome.token} covers this exact upstream state and file overlap; either "
        "moving before you re-run changes it, and this gate asks again with the current state."
    )


def _reconcile_summary(outcome: ReconcileOutcome, remote: str, branch: str) -> str:
    """One line for the outcomes that do not fail or gate - what to tell the caller, or "" for
    the ones not worth narrating every time (`up-to-date`, no remote)."""
    if outcome.status == "fast-forwarded":
        return f"Pulled {len(outcome.pulled_commits)} commit(s) from {remote}/{branch}."
    if outcome.status == "local-ahead":
        return f"Local branch is ahead of {remote}/{branch}; nothing to pull."
    if outcome.status == "rebased":
        reviewed = " after review" if outcome.was_reviewed else " (no file overlap, automatic)"
        return f"Rebased onto {len(outcome.pulled_commits)} new commit(s) from {remote}/{branch}{reviewed}."
    if outcome.status == "up-to-date":
        return f"Already up to date with {remote}/{branch}."
    if outcome.status == "no-remote-or-fetch-failed":
        return f"No remote configured, or {remote} could not be reached - continuing without a pull."
    return ""


def apply_reconcile(outcome: ReconcileOutcome, remote: str, branch: str, command: str) -> str:
    """Turn a `ReconcileOutcome` into this module's fail/needs_clearance contract: raises via
    `fail()` on `conflict`, raises via `needs_clearance()` on `needs-review` (emitting the same
    `gate.refused` shape the Mass-Update Gate uses, so the generic trajectory rules cover this
    gate for free), emits `gate.cleared` when a rebase just consumed a matching
    `--confirm-rebase`, and otherwise returns a one-line summary for the caller to print."""
    if outcome.status == "conflict":
        fail(
            f"Automatic rebase against {remote}/{branch} failed - resolve manually "
            f"(network/auth/merge conflict), do not force-push without asking the user:\n{outcome.detail}"
        )
    if outcome.status == "needs-review":
        emit(
            "wikitool", "gate.refused",
            {
                "gate": "rebase-review", "reason": "needs-clearance", "token": outcome.token,
                "presented_token": None, "remote": remote, "branch": branch,
                "overlap_files": outcome.overlap_files,
            },
        )
        needs_clearance(rebase_review_message(outcome, remote, branch, command))
    if outcome.status == "rebased" and outcome.was_reviewed:
        emit(
            "wikitool", "gate.cleared",
            {
                "gate": "rebase-review", "token": outcome.token, "remote": remote,
                "branch": branch, "overlap_files": outcome.overlap_files,
            },
        )
    return _reconcile_summary(outcome, remote, branch)


def sync_command(
    remote: str = typer.Option("origin", "--remote", help="Git remote to reconcile against"),
    branch: str = typer.Option("main", "--branch", help="Branch to reconcile"),
    confirm_rebase: Optional[str] = typer.Option(
        None,
        "--confirm-rebase",
        metavar="TOKEN",
        help="Clear the rebase-review gate for the exact upstream state and file overlap this "
        "token was issued for. The token comes from the gate's own refusal output.",
    ),
):
    """Fetch `<remote>/<branch>` and bring the local branch up to date with it: fast-forward
    when possible, rebase local commits on top when that is safe, and ask for review when it is
    not. Never commits, never pushes - the read-only-until-safe counterpart to `publish`'s own
    reconcile step, meant to run once at the start of a writing session so the rest of it works
    against a current tree instead of discovering the drift at the final push."""
    outcome = reconcile(remote, branch, confirm_rebase)
    summary = apply_reconcile(outcome, remote, branch, "sync")
    success(summary or f"Nothing to reconcile against {remote}/{branch}.")


def publish_command(
    message: str = typer.Option(..., "--message", help="Commit message summary, e.g. 'ingest: docker-cheatsheet'"),
    push: bool = typer.Option(True, "--push/--no-push"),
    confirm: Optional[str] = typer.Option(
        None,
        "--confirm",
        metavar="TOKEN",
        help="Clear the Mass-Update Gate for the exact changeset this token was issued for. "
        "The token comes from the gate's own refusal output - run without it first to see the "
        "file list, show that output to the user, and only pass this once they have approved it.",
    ),
    confirm_rebase: Optional[str] = typer.Option(
        None,
        "--confirm-rebase",
        metavar="TOKEN",
        help="Clear the rebase-review gate raised by the pre-push reconcile step. The token "
        "comes from that gate's own refusal output - see `tools/wikitool sync`.",
    ),
    yes: bool = typer.Option(
        False,
        "--yes",
        "-y",
        hidden=True,
        help="Removed - kept only so a stale invocation gets a real ERROR instead of a Typer "
        "usage error. The gate is cleared with --confirm <token>.",
    ),
    threshold: int = typer.Option(
        DEFAULT_MASS_UPDATE_THRESHOLD,
        "--threshold",
        help="Mass-Update Gate threshold: number of counted files at/above which clearance is required.",
    ),
    remote: str = typer.Option("origin", "--remote", help="Git remote to push to"),
    branch: str = typer.Option("main", "--branch", help="Branch to push"),
    path: Optional[list[str]] = typer.Option(
        None,
        "--path",
        help="Limit this publish to a path (repeatable). Lets a large change be committed in "
        "reviewable batches instead of one opaque commit - the Mass-Update Gate then counts "
        "only the files in scope.",
    ),
):
    """Stage all changes, commit, and push to <remote>/<branch>."""
    if yes:
        fail(YES_REMOVED_MESSAGE)

    paths = list(path or [])

    # Checked before anything is staged, for the same reason as the gate below:
    # a refused publish must leave the working tree exactly as it was found.
    # `--branch` names the ref git will push, which is not necessarily the ref
    # this commit lands on - see branch_mismatch_message().
    checked_out = current_branch()
    if push and checked_out != branch:
        fail(branch_mismatch_message(checked_out, branch))

    # Before the reconcile below, which is the first thing that talks to the
    # remote at all: a publish aimed at the wrong repository should not even
    # fetch from it, and the refusal message promises that nothing was
    # committed or pushed.
    if push:
        refusal = publish_remote_refusal(remote, branch)
        if refusal:
            emit(
                "wikitool",
                "gate.refused",
                {
                    "gate": "publish-remote",
                    "reason": "remote-not-allowed",
                    "remote": remote,
                    "branch": branch,
                    "url": push_url_for(remote),
                },
            )
            needs_clearance(refusal)

    # Pull against the remote before anything else, to minimise the window in which this
    # publish could diverge from it - and, as a side effect, to finally publish a commit left
    # stranded by a previous push that failed (see `_local_ahead_of_remote` below). Skipped
    # entirely for `--no-push`: nothing is being published this call, so there is no race to
    # protect against, and rewriting local history when nobody asked to publish would surprise
    # a deliberate local-only commit.
    if push:
        outcome = reconcile(remote, branch, confirm_rebase)
        summary = apply_reconcile(outcome, remote, branch, "publish")
        if summary:
            typer.echo(summary)

    # The gate is evaluated *before* anything is staged, so a refused publish
    # leaves the working tree exactly as it was found.
    changes = collect_changes(paths)
    local_ahead = push and _local_ahead_of_remote(remote, branch)
    if not changes:
        if not local_ahead:
            success("Nothing to commit.")
            return
        # A stranded commit from an earlier failed publish (or one the reconcile step above
        # just rebased back on top of the remote): nothing new to stage, but there is still
        # something to push below.
        typer.echo("No new changes to commit; publishing the existing local commit(s).")
    changed_files = [c.path for c in changes]

    counted = counted_files_of(changes)
    if len(counted) >= threshold:
        token = changeset_token(counted, threshold, remote, branch, paths)
        if confirm != token:
            # Both "no clearance yet" and "clearance for a changeset that has
            # since moved" end here: the user has not seen *this* state, so the
            # answer is the same, only the explanation differs.
            emit(
                "wikitool",
                "gate.refused",
                {
                    "gate": "mass-update",
                    "reason": "stale-token" if confirm else "needs-clearance",
                    "token": token,
                    "presented_token": confirm,
                    "changed": len(changes),
                    "counted": len(counted),
                    "threshold": threshold,
                    "files": [c.path for c in counted],
                },
            )
            needs_clearance(clearance_message(
                changes, threshold, token,
                rerun_command(token, message, push, threshold, remote, branch, paths),
                remote, branch, stale_token=confirm,
            ))
        emit(
            "wikitool",
            "gate.cleared",
            {
                "gate": "mass-update",
                "token": token,
                "counted": len(counted),
                "threshold": threshold,
                "files": [c.path for c in counted],
            },
        )

    if changes:
        add_result = _run(["git", "add", "-A", "--", *paths])
        if add_result.returncode != 0:
            fail(f"git add failed:\n{add_result.stderr}")

        file_list = "\n".join(f"- {f}" for f in changed_files)
        full_message = f"{message}\n\nFiles changed:\n{file_list}"

        # With a pathspec, `git commit -- <paths>` commits exactly those paths and
        # ignores anything else that happens to be staged, so batches stay disjoint.
        commit_args = ["git", "commit", "-m", full_message]
        if paths:
            commit_args += ["--", *paths]
        commit_result = _run(commit_args)
        if commit_result.returncode != 0:
            fail(f"git commit failed:\n{commit_result.stderr}")
        typer.echo(commit_result.stdout)

    if push:
        push_result = _run(["git", "push", remote, branch])
        if push_result.returncode != 0:
            # The real, narrow race this module exists to close: something landed on the
            # remote between the reconcile above and this push. One more reconcile-and-retry,
            # never a loop - if it finds nothing new, the rejection had some other cause
            # (branch protection, a hook), and the *original* error is what gets reported.
            retry_outcome = reconcile(remote, branch, confirm_rebase)
            if retry_outcome.status in ("fast-forwarded", "rebased"):
                retry_summary = apply_reconcile(retry_outcome, remote, branch, "publish")
                if retry_summary:
                    typer.echo(retry_summary)
                push_result = _run(["git", "push", remote, branch])
            elif retry_outcome.status in ("conflict", "needs-review"):
                apply_reconcile(retry_outcome, remote, branch, "publish")  # raises
            if push_result.returncode != 0:
                fail(
                    "git push failed - resolve manually (network/auth/merge conflict), "
                    f"do not force-push without asking the user:\n{push_result.stderr}"
                )
        typer.echo(push_result.stdout)

    emit(
        "wikitool",
        "publish.commit",
        {
            "summary": message,
            "files": changed_files,
            "changed": len(changed_files),
            "counted": len(counted_files(changed_files)),
            "paths": paths,
            "pushed": push,
            "remote": remote,
            "branch": branch,
        },
    )
    success(f"Published changes to {remote}/{branch}." if push else "Committed changes (not pushed).")