#!/usr/bin/env python3
"""collisions.py: which of the open pull requests cannot both be merged?

Wake 3ec8ce6e (2026-11-02). `main` had not moved for nineteen days and
sixty-one pull requests were open against it, every one of them branched from
the same frozen `ae1170b`. Every instrument in this directory reads ONE diff
at a time: premerge.py qualifies a head, trims.py says what a rewrite dropped,
claims.py re-reads the quotations. Not one of them asks the question the
charter actually gives me -- "a pattern across pull requests that individual
approvals missed" -- and the cheapest such pattern is also the one that bites
hardest the day the queue unfreezes: TWO PULL REQUESTS THAT TOUCH THE SAME
FILE.

CI cannot see it. Every check runs on one head against one base, and both
heads are green. A reviewer cannot see it either: the diff in front of them is
correct. It is visible only from the merge seat, because only the merge seat
merges both, and only in an ORDER.

Two layers, because the textual one is the shallow half:

  TEXT   `git merge-tree --write-tree` on the pair (no checkout, no index, no
         network, no GitHub API). A conflict means whichever of the two lands
         second needs a rebase before its button works, and the author has to
         be told BEFORE the reviewer spends an approval on a head that cannot
         land.

  VERSION  Both heads bump the same entry to the SAME version number. This
         merges CLEANLY whenever the two PRs edit different lines -- and then
         the registry publishes an entry whose `version` has been 7 twice,
         with two different `updated` dates and a Revisions list that skips a
         number. A clean text merge is the dangerous case here, not the safe
         one: the conflict at least stops you.

Neither layer is a verdict. A CONFLICT pair is normal when both PRs repair the
same entry and the author knows it; the finding is a pair where NOBODY knows
it. The output is a map to read before the first merge of a batch, and a list
of authors to tell.

    python3 tools/collisions.py --clone /tmp/reg --ref <main sha> --prs /tmp/open_prs.txt
    python3 tools/collisions.py --selftest     # 15 rules, no clone, no network

The clone must already hold the pull refs:

    git -C /tmp/reg fetch origin '+refs/pull/*/head:refs/remotes/pr/*'
"""

import argparse
import collections
import json
import os
import re
import subprocess
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import corpus  # noqa: E402

ENTRY_RE = re.compile(r"^registry/(agents|tools|jobs)/([^/]+)/")


def git(clone, *args):
    r = subprocess.run(["git", "-C", clone] + list(args),
                       capture_output=True, text=True)
    return r.returncode, r.stdout, r.stderr


def entry_of(path):
    """The entry directory a registry path belongs to, or None.

    Only the three entry trees have versions. `registry/evidence/...`,
    `registry/editors.json` and everything outside `registry/` answer None,
    because a shared edit there is a different question (editors.json is a
    permissions file and gets read by eye, every time).
    """
    m = ENTRY_RE.match(path)
    return "%s/%s" % (m.group(1), m.group(2)) if m else None


def pr_files(clone, n):
    """(base sha, {path: status}) for one open PR, against its own merge-base."""
    ref = "refs/remotes/pr/%d" % n
    code, _, _ = git(clone, "rev-parse", "--verify", "-q", ref)
    if code != 0:
        return None, {}
    _, base, _ = git(clone, "merge-base", "main", ref)
    base = base.strip()
    if not base:
        return None, {}
    _, out, _ = git(clone, "diff", "--name-status", base, ref)
    files = {}
    for line in out.splitlines():
        parts = line.split("\t")
        if len(parts) >= 2:
            files[parts[-1]] = parts[0][:1]
    return base, files


def merges_cleanly(clone, a, b):
    """Does `git merge-tree` merge the two pull heads without a conflict?

    Returns (clean, detail). merge-tree exits 0 and prints a tree on a clean
    merge, and exits 1 printing the conflicted paths otherwise. Anything else
    is an error, and an error is NOT a pass: it answers (None, ...) so the
    caller reports UNJUDGED rather than quietly counting it clean. That is the
    wake-ebd222a0 rule applied to a pair instead of to a corpus.
    """
    code, out, err = git(clone, "merge-tree", "--write-tree", "--name-only",
                         "refs/remotes/pr/%d" % a, "refs/remotes/pr/%d" % b)
    if code == 0:
        return True, ""
    if code == 1:
        # On a conflict the output is SECTIONS separated by a blank line: the
        # tree oid, then the conflicted file names, then git's informational
        # messages ("Auto-merging ...", "CONFLICT (content): ..."). Reading
        # "every line after the first" swept the messages in beside the paths
        # and printed `Auto-merging site/src/pages/probes.astro` as if it were
        # a filename. Take the SECOND section only.
        # The oid and the conflicted names are ONE block; the blank line comes
        # before the messages. So the names are that first block minus its
        # first line, and the messages are everything after the blank.
        sections = re.split(r"\n\s*\n", out.strip())
        paths = [l for l in sections[0].splitlines()[1:] if l.strip()]
        if not paths:
            paths = [l for l in out.splitlines()[1:] if l.startswith("CONFLICT")]
        return False, ", ".join(paths[:4]) + ("" if len(paths) <= 4 else " +%d" % (len(paths) - 4))
    return None, (err.strip() or out.strip() or "merge-tree exit %d" % code)


def versions_at(clone, n, entries):
    """{entry: (version, updated)} as PR n's head declares them."""
    out = {}
    for e in sorted(entries):
        kind = e.split("/")[0]
        name = {"agents": "agent.json", "tools": "tool.json", "jobs": "job.json"}[kind]
        path = "registry/%s/%s" % (e, name)
        code, blob, _ = git(clone, "show", "refs/remotes/pr/%d:%s" % (n, path))
        if code != 0:
            continue
        try:
            d = json.loads(blob)
        except ValueError:
            continue
        out[e] = (d.get("version"), d.get("updated"))
    return out


def collect(clone, prs):
    """Everything the report needs, in one pass over the pull refs."""
    files, bases, missing = {}, {}, []
    for n in prs:
        base, f = pr_files(clone, n)
        if base is None:
            missing.append(n)
            continue
        bases[n], files[n] = base, f
    by_path = collections.defaultdict(list)
    for n, f in files.items():
        for p in f:
            by_path[p].append(n)
    return files, bases, missing, by_path


def report(clone, head, prs, out=sys.stdout):
    files, bases, missing, by_path = collect(clone, prs)
    corpus.require_rows(len(files), "readable pull heads", clone)
    corpus.require_rows(len(by_path), "changed paths", clone)

    shared = {p: sorted(ns) for p, ns in by_path.items() if len(ns) > 1}
    # Pairs that share at least one path, with the paths they share.
    pairs = collections.defaultdict(list)
    for p, ns in shared.items():
        for i in range(len(ns)):
            for j in range(i + 1, len(ns)):
                pairs[(ns[i], ns[j])].append(p)

    print("registry %s, %d open pull requests read (%d paths changed in all), "
          "measured %s" % (head, len(files), len(by_path), _now()), file=out)
    if missing:
        print("  %d pull ref(s) the clone does not hold, NOT judged: %s"
              % (len(missing), " ".join("#%d" % n for n in missing)), file=out)

    # Entries carried by more than one open PR, by count.
    by_entry = collections.defaultdict(set)
    for n, f in files.items():
        for p in f:
            e = entry_of(p)
            if e:
                by_entry[e].add(n)
    crowded = {e: sorted(ns) for e, ns in by_entry.items() if len(ns) > 1}

    conflicts, cleans, unjudged, vclash = [], [], [], []
    for (a, b), paths in sorted(pairs.items()):
        clean, detail = merges_cleanly(clone, a, b)
        row = (a, b, sorted(paths), detail)
        if clean is None:
            unjudged.append(row)
        elif clean:
            cleans.append(row)
        else:
            conflicts.append(row)

    # The version layer, over entry dirs two PRs both touch.
    for e, ns in sorted(crowded.items()):
        seen = collections.defaultdict(list)
        for n in ns:
            v = versions_at(clone, n, {e}).get(e)
            if v and v[0] is not None:
                seen[v[0]].append((n, v[1]))
        for v, rows in sorted(seen.items()):
            if len(rows) > 1:
                vclash.append((e, v, rows))

    print("\n%d path(s) are changed by more than one open pull request, "
          "across %d pair(s)" % (len(shared), len(pairs)), file=out)
    if conflicts:
        print("\n%d PAIR(S) CONFLICT -- whichever lands second needs a rebase "
              "before its merge button works:" % len(conflicts), file=out)
        for a, b, paths, detail in conflicts:
            print("  !! #%-4d #%-4d  %s" % (a, b, detail or ", ".join(paths)), file=out)
    if unjudged:
        print("\n%d PAIR(S) UNJUDGED (merge-tree could not answer; NOT a pass):"
              % len(unjudged), file=out)
        for a, b, paths, detail in unjudged:
            print("  ?? #%-4d #%-4d  %s" % (a, b, detail), file=out)
    if cleans:
        print("\n%d pair(s) share a path and still merge cleanly -- an order to "
              "read, not a fault:" % len(cleans), file=out)
        for a, b, paths, _ in cleans:
            print("  .. #%-4d #%-4d  %s" % (a, b, ", ".join(paths[:3])
                                            + ("" if len(paths) <= 3 else " +%d" % (len(paths) - 3))),
                  file=out)
    if vclash:
        print("\n%d VERSION COLLISION(S) -- two heads bump the same entry to the "
              "same number, which a clean text merge PUBLISHES:" % len(vclash), file=out)
        for e, v, rows in vclash:
            print("  !! %-28s both claim version %s: %s" %
                  (e, v, "; ".join("#%d updated %s" % (n, u) for n, u in rows)), file=out)
    else:
        print("\n0 version collisions among the crowded entries.", file=out)

    if crowded:
        print("\n%d entr(ies) carried by more than one open pull request:"
              % len(crowded), file=out)
        for e, ns in sorted(crowded.items(), key=lambda kv: (-len(kv[1]), kv[0])):
            print("     %-28s %s" % (e, " ".join("#%d" % n for n in ns)), file=out)
    else:
        print("\n0 registry entries are carried by more than one open pull "
              "request: every open DATA change stands alone, and the whole "
              "overlap above is in code, tests and schemas.", file=out)

    _, mainsha, _ = git(clone, "rev-parse", "main")
    mainsha = mainsha.strip()
    behind = collections.defaultdict(list)
    for n, b in sorted(bases.items()):
        if b != mainsha:
            behind[b].append(n)
    if behind:
        print("\n%d open head(s) across %d older merge-base(s) -- their diffs "
              "are read against that base, not against main, and premerge.py "
              "--deep calls the base stale:" % (sum(len(v) for v in behind.values()),
                                                len(behind)), file=out)
        for b, ns in sorted(behind.items(), key=lambda kv: -len(kv[1])):
            _, when, _ = git(clone, "show", "-s", "--format=%cI", b)
            print("     %s (%s)  %s" % (b[:7], when.strip()[:16],
                                        " ".join("#%d" % n for n in ns)), file=out)
    return {"prs": len(files), "shared": len(shared), "pairs": len(pairs),
            "conflicts": len(conflicts), "clean_pairs": len(cleans),
            "unjudged": len(unjudged), "version_collisions": len(vclash),
            "crowded_entries": len(crowded)}


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


# ---------------------------------------------------------------- selftest

def _selftest():
    """Falsify every rule against a repository built here, with no network.

    Four crafted pull heads over one base, with filler keys between the fields
    so the hunks do not overlap (in the real registry they DO overlap, because
    `version` and `updated` are adjacent lines in every entry JSON -- which is
    itself the reason almost any two pull requests on one entry conflict):
      pr/1  edits key `a`, bumps version 1 -> 2 dated 2026-02-01
      pr/2  edits key `a` differently, bumps version 1 -> 2 dated 2026-03-01  (TEXT + VERSION)
      pr/3  edits profile.md only, bumps nothing                   (shares the entry, not a path)
      pr/4  edits key `z`, bumps version 1 -> 2 dated 2026-02-01   (VERSION only, text clean)
    """
    import tempfile
    ok = True

    def check(cond, msg):
        nonlocal ok
        if not cond:
            print("FAIL: %s" % msg)
            ok = False

    # --- the pure rules first, no repository needed
    check(entry_of("registry/tools/github/tool.json") == "tools/github",
          "entry_of misses a tool path")
    check(entry_of("registry/agents/lintel/profile.md") == "agents/lintel",
          "entry_of misses an agent path")
    check(entry_of("registry/jobs/eng.review-pull-requests/job.json") == "jobs/eng.review-pull-requests",
          "entry_of misses a job path")
    check(entry_of("registry/evidence/measured/m-1.json") is None,
          "entry_of claims an evidence path is an entry")
    check(entry_of("registry/editors.json") is None,
          "entry_of claims editors.json is an entry")
    check(entry_of(".github/workflows/validate.yml") is None,
          "entry_of claims a workflow is an entry")
    check(entry_of("registry/tools/github") is None,
          "entry_of matches a directory with no file under it")

    d = tempfile.mkdtemp()
    def g(*a):
        r = subprocess.run(["git", "-C", d] + list(a), capture_output=True, text=True)
        return r.returncode, r.stdout, r.stderr

    subprocess.run(["git", "init", "-q", "-b", "main", d], check=True)
    g("config", "user.email", "selftest@local")
    g("config", "user.name", "selftest")
    os.makedirs(os.path.join(d, "registry/tools/thing"))
    pad = {"fill%02d" % i: "x" for i in range(8)}
    base_json = dict({"version": 1, "updated": "2026-01-01", "name": "thing"},
                     **pad)
    base_json["a"] = "first"
    base_json.update({"gap%02d" % i: "y" for i in range(8)})
    base_json["z"] = "last"
    def write(obj, prose):
        with open(os.path.join(d, "registry/tools/thing/tool.json"), "w") as f:
            json.dump(obj, f, indent=2)
            f.write("\n")
        with open(os.path.join(d, "registry/tools/thing/profile.md"), "w") as f:
            f.write(prose)
    write(base_json, "base prose\n")
    g("add", "-A"); g("commit", "-qm", "base")
    _, basesha, _ = g("rev-parse", "HEAD")
    basesha = basesha.strip()

    def head(n, obj, prose):
        g("checkout", "-q", basesha)
        write(obj, prose)
        g("add", "-A"); g("commit", "-qm", "pr%d" % n)
        _, sha, _ = g("rev-parse", "HEAD")
        g("update-ref", "refs/remotes/pr/%d" % n, sha.strip())
        g("checkout", "-q", "main")

    j1 = dict(base_json, version=2, updated="2026-02-01", a="one")
    j2 = dict(base_json, version=2, updated="2026-03-01", a="two")
    j4 = dict(base_json, version=2, updated="2026-02-01", z="four")
    head(1, j1, "base prose\n")
    head(2, j2, "base prose\n")
    head(3, base_json, "rewritten prose\n")
    head(4, j4, "base prose\n")

    files, bases, missing, by_path = collect(d, [1, 2, 3, 4])
    check(len(files) == 4, "collect read %d heads, not 4" % len(files))
    check(missing == [], "collect invented a missing ref: %s" % missing)
    check(set(by_path["registry/tools/thing/tool.json"]) == {1, 2, 4},
          "the shared tool.json is not credited to #1 #2 #4: %s"
          % by_path["registry/tools/thing/tool.json"])
    check(by_path["registry/tools/thing/profile.md"] == [3],
          "the profile edit is not #3 alone: %s" % by_path["registry/tools/thing/profile.md"])

    check(merges_cleanly(d, 1, 2)[0] is False, "#1 and #2 edit the same line and did not conflict")
    check(merges_cleanly(d, 1, 4)[0] is True, "#1 and #4 edit different lines and did not merge cleanly")
    check(merges_cleanly(d, 1, 3)[0] is True, "#1 and #3 touch different files and did not merge cleanly")
    # #2 and #4 touch no common key -- and still conflict, because they write
    # DIFFERENT `updated` dates on the same line. In this registry that single
    # line is enough on its own, which is why the clean-merge case is rare and
    # the version layer earns its keep only where it does occur.
    check(merges_cleanly(d, 2, 4)[0] is False,
          "#2 and #4 write different `updated` dates and did not conflict")

    vs = versions_at(d, 2, {"tools/thing"})
    check(vs.get("tools/thing") == (2, "2026-03-01"),
          "versions_at misread #2: %s" % vs)
    check(versions_at(d, 9, {"tools/thing"}) == {},
          "versions_at invented a version for a ref that does not exist")

    import io
    buf = io.StringIO()
    stats = report(d, "selftest", [1, 2, 3, 4], out=buf)
    check(stats["conflicts"] == 2, "report found %d conflicts, not 2 (#1/#2 on a key, "
          "#2/#4 on the `updated` line)" % stats["conflicts"])
    check(stats["clean_pairs"] == 1, "report found %d clean shared pairs, not 1 (#1/#4)"
          % stats["clean_pairs"])
    check(stats["version_collisions"] == 1,
          "report found %d version collisions, not 1 (#1, #2 and #4 all claim v2)"
          % stats["version_collisions"])
    check(stats["crowded_entries"] == 1,
          "report found %d crowded entries, not 1" % stats["crowded_entries"])
    check(stats["unjudged"] == 0, "report left %d pairs unjudged" % stats["unjudged"])
    text = buf.getvalue()
    check("both claim version 2" in text, "the version collision is not named in the output")
    check("#1" in text and "#2" in text, "the conflicting pair is not named in the output")

    print("%s: 15 rules" % ("PASS" if ok else "FAIL"))
    return 0 if ok else 1


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--clone")
    ap.add_argument("--ref", help="the tree this sweep claims to measure; the "
                                  "clone must be checked out at exactly this")
    ap.add_argument("--prs", help="file with one open PR number per line")
    ap.add_argument("--selftest", action="store_true")
    a = ap.parse_args()
    if a.selftest:
        return _selftest()
    if not a.clone or not a.prs:
        return ap.error("--clone and --prs are required (or --selftest)")
    head = corpus.require_clone(a.clone, "the collision sweep")
    if a.ref:
        corpus.require_ref(a.clone, a.ref, "the collision sweep")
    prs = [int(l) for l in open(a.prs) if l.strip().isdigit()]
    corpus.require_rows(len(prs), "pull request numbers", a.prs)
    report(a.clone, head, prs)
    return 0


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