#!/usr/bin/env python3
"""strays.py: does an open head carry another open head's work?

WHY THIS EXISTS. Wake 15a5217f, PR #343 ("rippling v3"). Its second commit
added `registry/evidence/measured/m-20261029-deepwiki-version-correctness.json`
-- 73 lines, a measured result about a DIFFERENT tool on a DIFFERENT job --
byte-identical to the single file added by open PR #328, and named nowhere in
#343's title, its 6,700-byte body, or its profile. `validate` was green on it
and correctly so: the record is schema-valid. **CI cannot see that a file
belongs to a different pull request, and a reviewer reading a diff against a
description cannot see it if the description does not mention it.** An approval
on #343 would have carried #328's content onto `main` under a rippling
description, and #328's own review would never have happened.

I found it by eye, because `premerge.py` prints the changed-file list as a
neutral row and I happened to read the row. That is not a gate. The question
this script asks is the one no single-PR instrument can ask, and my charter
names it as the value only this seat adds: **a pattern ACROSS pull requests
that individual approvals missed.**

WHAT IT ASKS, and both questions are about BLOB IDENTITY, never about text:

  DUPLICATE   one path, added by two or more open heads with the SAME blob
              and absent from main. Whoever merges first silently merges the
              other, and the second head's diff becomes empty.
  CONTAINS    head A's (path, blob) pairs are a strict superset of head B's
              ENTIRE change set. A is not "similar to" B; A *is* B plus more.
              This is the #343/#328 shape and the one to read first.
  COLLIDES    one path, two open heads, DIFFERENT blobs, absent from main.
              Not a stray -- two authors inventing the same filename -- but
              the merge order decides the content, so it is worth a line.
  STACKED     the same containment where the contained head's paths are all
              ALREADY ON MAIN: two edits of the same files, A including B's.
              Branch topology, NOT a stray. The first full run over every
              pull ref printed 391 rows and 389 were this; had they all read
              CONTAINS the instrument would have been worthless on its first
              day, which is the whole reason the distinction exists.

Deliberately NOT asked: "is this file unrelated to the title?" I drafted that
(flag a head whose files span two `registry/<family>` directories while its
title names one entry) and threw it away. 'Related' is a judgement about
prose, the registry has legitimate multi-family heads, and a gate that guesses
earns distrust the first time it cries wolf. Blob identity is a FACT. The
operator can read a CONTAINS row in one line and check it with `git show`.

No network, no GitHub API: everything comes from pull refs already in the
clone. Fetch them first:

  git -C <clone> fetch origin 'refs/pull/*/head:refs/remotes/pull/*'

A head listed in `refs/pull/` may be CLOSED -- git keeps those refs forever --
so pass `--open <n,n,...>` (or a file of numbers) when you have the open set
from `queue.py`, and the run says how many refs it scored either way.

    python3 tools/strays.py --clone /tmp/pa
    python3 tools/strays.py --selftest
"""

import argparse
import os
import subprocess
import sys

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


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


def pull_refs(clone):
    out = git(clone, "for-each-ref", "--format=%(refname:short)", "refs/remotes/pull/")
    if out is None:
        return []
    nums = []
    for line in out.split():
        n = line.rsplit("/", 1)[-1]
        if n.isdigit():
            nums.append(int(n))
    return sorted(nums)


def changeset(clone, ref, base):
    """The (path, blob) pairs this ref ADDS OR CHANGES against base.

    A DELETION is deliberately not a pair: it has no blob, and a head that
    deletes what another head deletes is the same change twice, which the
    merge resolves without anyone's content being smuggled anywhere.
    """
    out = git(clone, "diff", "--no-renames", "--raw", "-z", base, ref)
    if out is None:
        return None
    fields = out.split("\0")
    pairs, i = {}, 0
    while i < len(fields):
        meta = fields[i]
        if not meta.startswith(":"):
            i += 1
            continue
        parts = meta.split()
        status = parts[-1]
        dst_blob = parts[3]
        path = fields[i + 1] if i + 1 < len(fields) else ""
        if status != "D" and path:
            pairs[path] = dst_blob
        i += 2
    return pairs


def main_blobs(clone, base):
    out = git(clone, "ls-tree", "-r", base)
    blobs = {}
    for line in (out or "").splitlines():
        meta, path = line.split("\t", 1)
        blobs[path] = meta.split()[2]
    return blobs


def analyse(sets, on_main):
    """sets: {head number: {path: blob}}. Returns rows, pure and testable."""
    rows = []
    by_path = {}
    for n, pairs in sets.items():
        for path, blob in pairs.items():
            by_path.setdefault(path, {})[n] = blob

    for path, heads in sorted(by_path.items()):
        if len(heads) < 2 or path in on_main:
            continue
        groups = {}
        for n, blob in heads.items():
            groups.setdefault(blob, []).append(n)
        for blob, ns in sorted(groups.items()):
            if len(ns) > 1:
                rows.append(("DUPLICATE", sorted(ns), path, blob[:12]))
        if len(groups) > 1:
            rows.append(("COLLIDES", sorted(heads), path,
                         "%d distinct blobs" % len(groups)))

    for a, pa in sorted(sets.items()):
        for b, pb in sorted(sets.items()):
            if a == b or not pb:
                continue
            if all(pa.get(p) == blob for p, blob in pb.items()) and len(pa) > len(pb):
                # The distinction that decides whether a row is a FINDING or a
                # fact about branch topology, and the first full run taught it:
                # 391 rows, 389 of them stacked branches. If every path B
                # changes already exists on main, A and B are two edits of the
                # same files and A simply includes B's -- a merge-order
                # dependency, which thaw.py is the instrument for. If B's paths
                # are ALL NEW, A is carrying a whole body of work that exists
                # nowhere else: the #343/#328 shape.
                kind = ("CONTAINS" if all(p not in on_main for p in pb)
                        else "STACKED")
                rows.append((kind, [a, b], "%d of %d paths" % (len(pb), len(pa)),
                             ", ".join(sorted(pb))))
    order = {"CONTAINS": 0, "DUPLICATE": 1, "COLLIDES": 2, "STACKED": 3}
    rows.sort(key=lambda r: (order[r[0]], r[1]))
    return rows


def selftest():
    """Falsify the rules against fixed sets. No clone, no network."""
    ok = True

    def check(name, got, want):
        nonlocal ok
        if got != want:
            ok = False
            print("FAIL %s\n  got  %r\n  want %r" % (name, got, want))
        else:
            print("ok   %s" % name)

    # 1. The #343/#328 case: A is B plus its own work.
    rows = analyse({343: {"t/a": "b1", "e/x": "b9"}, 328: {"e/x": "b9"}}, set())
    check("CONTAINS fires on the real shape",
          [r[0] for r in rows], ["CONTAINS", "DUPLICATE"])
    check("CONTAINS names container first", rows[0][1], [343, 328])

    # 1b. The same containment where the contained head edits files ALREADY ON
    #     MAIN is a stacked branch, not a stray. 389 of the first full run's 391
    #     rows were this, and calling them strays would have made the
    #     instrument useless on its first day.
    rows = analyse({265: {"a": "b1", "b": "b2"}, 259: {"a": "b1"}}, {"a": "b0"})
    check("STACKED, not CONTAINS, when the paths are on main",
          [r[0] for r in rows], ["STACKED"])

    # 2. The same file at the same blob on two heads, neither containing the other.
    rows = analyse({1: {"e/x": "b9", "p": "b2"}, 2: {"e/x": "b9", "q": "b3"}}, set())
    check("DUPLICATE without containment", [r[0] for r in rows], ["DUPLICATE"])

    # 3. A file ALREADY ON MAIN that two heads both touch is ordinary work on a
    #    shared file, not a stray. This is the rule that keeps the noise out.
    rows = analyse({1: {"registry/tools.json": "b1"}, 2: {"registry/tools.json": "b1"}},
                   {"registry/tools.json": "b0"})
    check("a path on main is never a stray", rows, [])

    # 4. Same path, different content: merge order decides, but nobody's work is
    #    carried anywhere.
    rows = analyse({1: {"e/x": "b1"}, 2: {"e/x": "b2"}}, set())
    check("COLLIDES, not DUPLICATE", [r[0] for r in rows], ["COLLIDES"])

    # 5. Equal change sets are each other's duplicate and NEITHER contains the
    #    other -- len(pa) > len(pb) is why. Two identical heads are one
    #    finding, not two.
    rows = analyse({1: {"e/x": "b9"}, 2: {"e/x": "b9"}}, set())
    check("identical heads: DUPLICATE only", [r[0] for r in rows], ["DUPLICATE"])

    # 6. A head that merely touches more files does not contain a head whose
    #    own file differs.
    rows = analyse({1: {"a": "b1", "b": "b2"}, 2: {"a": "bX"}}, set())
    check("superset of PATHS is not containment", [r[0] for r in rows], ["COLLIDES"])

    # 7. An empty change set contains nothing and is contained by nothing.
    rows = analyse({1: {"a": "b1"}, 2: {}}, set())
    check("an empty head is not contained", rows, [])

    # 8. A deletion carries no blob, so changeset() must not pair it. Checked
    #    on the parser, not on analyse().
    fields = ":100644 000000 abc123 0000000 D\0gone.json\0:000000 100644 0000000 def456 A\0new.json\0"
    pairs, i = {}, 0
    out = fields.split("\0")
    while i < len(out):
        if not out[i].startswith(":"):
            i += 1
            continue
        parts = out[i].split()
        if parts[-1] != "D":
            pairs[out[i + 1]] = parts[3]
        i += 2
    check("a deletion is not a (path, blob) pair", pairs, {"new.json": "def456"})

    print("\n%s" % ("10 rules hold." if ok else "SELFTEST FAILED"))
    return ok


def parse_open(value):
    if not value:
        return None
    if os.path.isfile(value):
        value = open(value).read()
    return set(int(t) for t in value.replace(",", " ").split() if t.strip().isdigit())


def run(clone, base, open_set):
    head = corpus.require_clone(clone, "strays.py")
    nums = pull_refs(clone)
    if not nums:
        sys.exit("REFUSED: no refs/remotes/pull/* in %s, so strays.py would score an\n"
                 "         empty corpus. Run: git -C %s fetch origin "
                 "'refs/pull/*/head:refs/remotes/pull/*'" % (clone, clone))
    base_sha = (git(clone, "rev-parse", base) or "").strip()
    if not base_sha:
        sys.exit("REFUSED: %s cannot resolve %s." % (clone, base))

    scored, skipped = {}, []
    for n in nums:
        if open_set is not None and n not in open_set:
            continue
        pairs = changeset(clone, "pull/%d" % n, base_sha)
        if pairs is None:
            skipped.append(n)
            continue
        if pairs:
            scored[n] = pairs

    if not scored:
        sys.exit("REFUSED: 0 of %d pull refs produced a change set against %s."
                 % (len(nums), base))

    on_main = main_blobs(clone, base_sha)
    rows = analyse(scored, set(on_main))

    print("strays.py over %s (clone head %s), base %s %s"
          % (clone, head, base, base_sha[:8]))
    print("%d pull refs present, %d scored%s, %d empty against base%s"
          % (len(nums), len(scored),
             " (--open given, %d refs outside it)" % (len(nums) - len(
                 [n for n in nums if open_set is None or n in open_set]))
             if open_set is not None else "",
             len([n for n in nums if (open_set is None or n in open_set)
                  and n not in scored and n not in skipped]),
             ", %d unreadable" % len(skipped) if skipped else ""))
    print()
    if not rows:
        print("No head carries another head's work. 0 CONTAINS, 0 DUPLICATE, "
              "0 COLLIDES, 0 STACKED.")
        return 0
    tally = {}
    for kind, heads, a, b in rows:
        tally[kind] = tally.get(kind, 0) + 1
        if kind == "CONTAINS":
            print("CONTAINS  #%d carries ALL of #%d's change set (%s), every path NEW\n"
                  "          %s" % (heads[0], heads[1], a, b))
        elif kind == "DUPLICATE":
            print("DUPLICATE %s add %s at the same blob %s"
                  % (", ".join("#%d" % n for n in heads), a, b))
        elif kind == "COLLIDES":
            print("COLLIDES  %s add %s, %s"
                  % (", ".join("#%d" % n for n in heads), a, b))
        else:
            print("STACKED   #%d includes #%d's edits to files already on main (%s)"
                  % (heads[0], heads[1], a))
    print("\n%d row(s): %s. A CONTAINS row is the one to read first: an approval on\n"
          "the container is an approval on the contained head's content too, and that\n"
          "head's own review has not happened. A STACKED row is branch topology, not a\n"
          "stray -- it is a merge-order dependency and thaw.py is the instrument for it."
          % (len(rows), ", ".join("%d %s" % (v, k) for k, v in sorted(tally.items()))))
    return 1


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--clone")
    ap.add_argument("--base", default="origin/main")
    ap.add_argument("--open", help="comma/space list of open PR numbers, or a file of them")
    ap.add_argument("--selftest", action="store_true")
    a = ap.parse_args()
    if a.selftest:
        sys.exit(0 if selftest() else 1)
    if not a.clone:
        sys.exit("usage: strays.py --clone <path> [--base origin/main] [--open n,n] | --selftest")
    sys.exit(run(a.clone, a.base, parse_open(a.open)))