open-swe/scripts/triage.py

#!/usr/bin/env python3
"""Upstream cherry-pick triage ledger tool.

Source of truth is ``docs/upstream-sync/triage.jsonl`` (one JSON object per line);
``docs/upstream-sync/triage.md`` is a GENERATED, human-readable view of it.

Subcommands:
    migrate <triage.md>   one-shot: parse the hand-written markdown ledger -> triage.jsonl
    generate [--check]    render triage.jsonl -> triage.md (--check: fail if md is stale)
    reconcile             drain .git journal -> mark SHAs landed -> render -> git add
    sync                  add new base..ref commits (default dev..upstream/main) as untriaged
    lookup <sha>          print a SHA's disposition (+ reason)
    check-reject <sha>    exit 3 iff the SHA is disposition "wont-merge" (used by hooks)
    set <sha> ...         edit a row (disposition / pr / subject / reason / branch)

Stdlib only. Hooks call this without uv, so keep it dependency-free.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
from datetime import UTC, datetime
from pathlib import Path

DISPOSITIONS = ("landed", "wont-merge", "deferred", "untriaged")

DISPOSITION_LABELS = {
    "landed": "Landed",
    "wont-merge": "Won't merge",
    "deferred": "Deferred",
    "untriaged": "Untriaged",
}

# Normalized label text -> disposition. Keys are lowercased with apostrophes stripped.
_LABEL_TO_DISPOSITION = {
    "landed": "landed",
    "wont merge": "wont-merge",
    "wontmerge": "wont-merge",
    "wont-merge": "wont-merge",
    "deferred": "deferred",
    "untriaged": "untriaged",
}

GENERATED_BANNER = (
    "<!-- GENERATED — do not hand-edit. Edit docs/upstream-sync/triage.jsonl, then run "
    "`make triage-render`. Staleness is enforced in CI by `make triage-check`. -->"
)

PREAMBLE = (
    "# Upstream triage ledger\n"
    "\n"
    "Commits on `upstream/main` (langchain-ai/open-swe) not yet in `dev`, and the decision on each.\n"
    "Rows key on the **upstream SHA** (stable across local cherry-picks). Deferred rows are provisional\n"
    "— re-inspect before picking. See the fork-maintenance runbook in `CLAUDE.md`.\n"
)

MAINTENANCE_NOTE = (
    "_Maintenance: after a `git sync`, add new `dev..upstream/main` SHAs as **Untriaged** "
    '(edit `triage.jsonl`) and bump "Last synced". A successful `git cherry-pick -x` auto-moves '
    "the row to **Landed** via the `post-commit` journal + `make triage-reconcile`._\n"
)

JOURNAL_NAME = "sh-cherrypick-journal"


# --------------------------------------------------------------------------- paths


def repo_root() -> Path:
    try:
        out = subprocess.run(
            ["git", "rev-parse", "--show-toplevel"],
            capture_output=True,
            text=True,
            check=True,
        )
        return Path(out.stdout.strip())
    except Exception:
        return Path(__file__).resolve().parents[1]


def git_dir() -> Path | None:
    try:
        out = subprocess.run(
            ["git", "rev-parse", "--absolute-git-dir"],
            capture_output=True,
            text=True,
            check=True,
        )
        return Path(out.stdout.strip())
    except Exception:
        return None


def jsonl_path(root: Path) -> Path:
    return root / "docs" / "upstream-sync" / "triage.jsonl"


def md_path(root: Path) -> Path:
    return root / "docs" / "upstream-sync" / "triage.md"


# ------------------------------------------------------------------------- records


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


def load_ledger(path: Path) -> tuple[dict, list[dict]]:
    """Return (meta, records). First line may be a ``{"_meta": {...}}`` object."""
    meta: dict = {}
    records: list[dict] = []
    if not path.exists():
        return meta, records
    for lineno, raw in enumerate(path.read_text().splitlines(), start=1):
        line = raw.strip()
        if not line:
            continue
        try:
            obj = json.loads(line)
        except json.JSONDecodeError as exc:
            raise ValueError(f"{path}:{lineno}: invalid JSON: {exc}") from exc
        if "_meta" in obj:
            meta = obj["_meta"]
            continue
        records.append(obj)
    return meta, records


def write_ledger(path: Path, meta: dict, records: list[dict]) -> None:
    lines = [json.dumps({"_meta": meta}, ensure_ascii=False)]
    for rec in records:
        lines.append(json.dumps(rec, ensure_ascii=False))
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text("\n".join(lines) + "\n")
    os.replace(tmp, path)


def sha_match(a: str, b: str) -> bool:
    """Prefix match either direction — ledger holds short SHAs, journal holds full SHAs."""
    a = a.lower()
    b = b.lower()
    return a.startswith(b) or b.startswith(a)


def find_record(records: list[dict], sha: str) -> dict | None:
    for rec in records:
        if sha_match(rec.get("sha", ""), sha):
            return rec
    return None


# ------------------------------------------------------------------------- render


def _disposition_rank(disp: str) -> int:
    try:
        return DISPOSITIONS.index(disp)
    except ValueError:
        return len(DISPOSITIONS)


def render_md(meta: dict, records: list[dict]) -> str:
    last_synced = meta.get("last_synced", "")
    last_synced_date = meta.get("last_synced_date", "")
    synced_line = ""
    if last_synced:
        synced_line = f"**Last synced `upstream/main`:** `{last_synced}`"
        if last_synced_date:
            synced_line += f" ({last_synced_date})"
        synced_line += "\n"

    ordered = sorted(
        enumerate(records),
        key=lambda item: (_disposition_rank(item[1].get("disposition", "")), item[0]),
    )

    rows = [
        "| sha | pr | subject | decision | why | branch |",
        "|---|---|---|---|---|---|",
    ]
    for _, rec in ordered:
        sha = rec.get("sha", "")
        pr = rec.get("pr")
        pr_cell = f"#{pr}" if pr not in (None, "") else ""
        subject = rec.get("subject", "") or ""
        decision = DISPOSITION_LABELS.get(rec.get("disposition", ""), rec.get("disposition", ""))
        why = rec.get("reason", "") or ""
        branch = rec.get("branch", "") or ""
        cells = [f"`{sha}`", pr_cell, subject, decision, why, branch]
        rows.append("| " + " | ".join(cells) + " |")

    parts = [GENERATED_BANNER, "", PREAMBLE.rstrip("\n"), ""]
    if synced_line:
        parts.append(synced_line.rstrip("\n"))
        parts.append("")
    parts.append("\n".join(rows))
    parts.append("")
    parts.append(MAINTENANCE_NOTE.rstrip("\n"))
    return "\n".join(parts) + "\n"


# ------------------------------------------------------------------------ migrate


_ROW_RE = re.compile(r"^\|(.+)\|\s*$")


def _norm_label(text: str) -> str:
    return re.sub(r"[^a-z\- ]", "", text.strip().lower()).strip()


def parse_md_ledger(text: str) -> tuple[dict, list[dict]]:
    meta: dict = {}
    records: list[dict] = []

    m = re.search(r"Last synced.*?`([0-9a-f]{6,40})`\s*(?:\(([^)]*)\))?", text)
    if m:
        meta["last_synced"] = m.group(1)
        if m.group(2):
            meta["last_synced_date"] = m.group(2)

    for raw in text.splitlines():
        m = _ROW_RE.match(raw)
        if not m:
            continue
        cells = [c.strip() for c in m.group(1).split("|")]
        if len(cells) < 6:
            continue
        first = cells[0].strip("`").strip().lower()
        # skip header + separator rows
        if first in ("sha", "") or set(cells[0]) <= {"-", " "}:
            continue
        if not re.fullmatch(r"[0-9a-f]{7,40}", first):
            continue
        sha = first
        pr_raw = cells[1].lstrip("#").strip()
        pr = int(pr_raw) if pr_raw.isdigit() else None
        subject = cells[2].strip()
        disposition = _LABEL_TO_DISPOSITION.get(_norm_label(cells[3]))
        if disposition is None:
            raise ValueError(f"unrecognized decision {cells[3]!r} for {sha}")
        reason = cells[4].strip()
        branch = cells[5].strip()
        updated = (
            f"{meta.get('last_synced_date')}T00:00:00Z"
            if meta.get("last_synced_date")
            else _now_iso()
        )
        records.append(
            {
                "sha": sha,
                "pr": pr,
                "subject": subject,
                "disposition": disposition,
                "reason": reason,
                "branch": branch,
                "local_sha": None,
                "updated": updated,
            }
        )
    return meta, records


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


def cmd_migrate(args: argparse.Namespace) -> int:
    root = repo_root()
    src = Path(args.source)
    if not src.exists():
        print(f"error: {src} not found", file=sys.stderr)
        return 2
    meta, records = parse_md_ledger(src.read_text())
    out = jsonl_path(root)
    out.parent.mkdir(parents=True, exist_ok=True)
    write_ledger(out, meta, records)
    print(f"migrated {len(records)} rows -> {out}")
    md_path(root).write_text(render_md(meta, records))
    print(f"rendered {md_path(root)}")
    return 0


def cmd_generate(args: argparse.Namespace) -> int:
    root = repo_root()
    jl = jsonl_path(root)
    try:
        meta, records = load_ledger(jl)
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    rendered = render_md(meta, records)
    md = md_path(root)
    if args.check:
        current = md.read_text() if md.exists() else ""
        if current != rendered:
            print(
                f"error: {md} is stale vs {jl}. Run `make triage-render` and commit.",
                file=sys.stderr,
            )
            return 1
        print(f"ok: {md.name} is up to date with {jl.name}")
        return 0
    md.write_text(rendered)
    print(f"rendered {md}")
    return 0


def _read_journal(journal: Path) -> list[tuple[str, str, str]]:
    entries: list[tuple[str, str, str]] = []
    seen: set[str] = set()
    if not journal.exists():
        return entries
    for raw in journal.read_text().splitlines():
        line = raw.strip()
        if not line:
            continue
        parts = line.split("\t")
        up = parts[0]
        local = parts[1] if len(parts) > 1 else ""
        ts = parts[2] if len(parts) > 2 else _now_iso()
        if up in seen:
            # keep the latest entry for a given upstream sha
            entries = [e for e in entries if e[0] != up]
        seen.add(up)
        entries.append((up, local, ts))
    return entries


def cmd_reconcile(args: argparse.Namespace) -> int:
    root = repo_root()
    gd = git_dir()
    if gd is None:
        print("error: not a git repository", file=sys.stderr)
        return 2
    journal = gd / JOURNAL_NAME
    entries = _read_journal(journal)
    if not entries:
        print("reconcile: journal empty — nothing to do")
        return 0

    jl = jsonl_path(root)
    try:
        meta, records = load_ledger(jl)
    except ValueError as exc:
        print(f"error: {exc} — aborting without writing; journal preserved", file=sys.stderr)
        return 2

    landed = 0
    inserted = 0
    for up_sha, local_sha, ts in entries:
        rec = find_record(records, up_sha)
        if rec is None:
            subject = ""
            if local_sha:
                try:
                    subject = subprocess.run(
                        ["git", "log", "-1", "--format=%s", local_sha],
                        capture_output=True,
                        text=True,
                        check=True,
                    ).stdout.strip()
                except Exception:
                    subject = ""
            records.append(
                {
                    "sha": up_sha[:8],
                    "pr": None,
                    "subject": subject,
                    "disposition": "landed",
                    "reason": "",
                    "branch": "",
                    "local_sha": local_sha or None,
                    "updated": ts,
                }
            )
            inserted += 1
            landed += 1
            continue
        already = rec.get("disposition") == "landed" and rec.get("local_sha") == (local_sha or None)
        if already:
            continue
        rec["disposition"] = "landed"
        rec["local_sha"] = local_sha or None
        rec["updated"] = ts
        landed += 1

    write_ledger(jl, meta, records)
    md = md_path(root)
    md.write_text(render_md(meta, records))

    # stage the tracked ledger; drain the journal
    try:
        subprocess.run(["git", "add", str(jl), str(md)], check=True)
    except Exception as exc:  # noqa: BLE001
        print(f"warning: `git add` failed ({exc}); files updated but not staged", file=sys.stderr)
    journal.write_text("")

    print(
        f"reconcile: {landed} landed ({inserted} new), staged {jl.name} + {md.name}, "
        "journal drained. Commit the ledger separately."
    )
    return 0


def cmd_lookup(args: argparse.Namespace) -> int:
    root = repo_root()
    try:
        _meta, records = load_ledger(jsonl_path(root))
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    rec = find_record(records, args.sha)
    if rec is None:
        print("unknown")
        return 0
    disp = rec.get("disposition", "")
    reason = rec.get("reason", "")
    print(f"{disp}\t{reason}")
    return 0


def cmd_check_reject(args: argparse.Namespace) -> int:
    """Exit 3 iff SHA is disposition 'wont-merge'. Exit 0 otherwise. Exit 2 on ledger error."""
    root = repo_root()
    try:
        _meta, records = load_ledger(jsonl_path(root))
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    rec = find_record(records, args.sha)
    if rec is not None and rec.get("disposition") == "wont-merge":
        print(rec.get("reason", "") or "(no reason recorded)")
        return 3
    return 0


def cmd_set(args: argparse.Namespace) -> int:
    root = repo_root()
    jl = jsonl_path(root)
    try:
        meta, records = load_ledger(jl)
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    rec = find_record(records, args.sha)
    if rec is None:
        rec = {
            "sha": args.sha[:8],
            "pr": None,
            "subject": "",
            "disposition": "untriaged",
            "reason": "",
            "branch": "",
            "local_sha": None,
            "updated": _now_iso(),
        }
        records.append(rec)
    if args.disposition:
        if args.disposition not in DISPOSITIONS:
            print(f"error: disposition must be one of {DISPOSITIONS}", file=sys.stderr)
            return 2
        rec["disposition"] = args.disposition
    if args.pr is not None:
        rec["pr"] = args.pr
    if args.subject is not None:
        rec["subject"] = args.subject
    if args.reason is not None:
        rec["reason"] = args.reason
    if args.branch is not None:
        rec["branch"] = args.branch
    rec["updated"] = _now_iso()
    write_ledger(jl, meta, records)
    md_path(root).write_text(render_md(meta, records))
    print(f"updated {rec['sha']} -> {rec['disposition']}")
    return 0


def _diverged_commits(base: str, ref: str) -> list[tuple[str, int | None, str]]:
    """Return ``[(short_sha, pr, subject)]`` for non-merge commits in ``base..ref``.

    Newest first, matching ``git log`` default order. ``pr`` is parsed from a
    trailing ``(#1234)`` in the subject, else ``None``.
    """
    out = subprocess.run(
        ["git", "log", "--no-merges", "--format=%h%x09%s", f"{base}..{ref}"],
        capture_output=True,
        text=True,
        check=True,
    ).stdout
    commits: list[tuple[str, int | None, str]] = []
    for line in out.splitlines():
        if "\t" not in line:
            continue
        sha, subject = line.split("\t", 1)
        prs = re.findall(r"\(#(\d+)\)", subject)
        pr = int(prs[-1]) if prs else None
        commits.append((sha.strip(), pr, subject.strip()))
    return commits


def cmd_sync(args: argparse.Namespace) -> int:
    root = repo_root()
    ref = args.ref
    if not args.no_fetch:
        remote = ref.split("/", 1)[0] if "/" in ref else "upstream"
        try:
            subprocess.run(["git", "fetch", remote], check=True)
        except Exception as exc:  # noqa: BLE001
            print(f"error: `git fetch {remote}` failed: {exc}", file=sys.stderr)
            return 2
    try:
        commits = _diverged_commits(args.base, ref)
    except subprocess.CalledProcessError as exc:
        print(f"error: `git log {args.base}..{ref}` failed: {exc}", file=sys.stderr)
        return 2

    jl = jsonl_path(root)
    try:
        meta, records = load_ledger(jl)
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    added = 0
    # Append oldest-first so freshly-discovered rows read chronologically.
    for sha, pr, subject in reversed(commits):
        if find_record(records, sha) is not None:
            continue
        records.append(
            {
                "sha": sha[:8],
                "pr": pr,
                "subject": subject,
                "disposition": "untriaged",
                "reason": "",
                "branch": "",
                "local_sha": None,
                "updated": _now_iso(),
            }
        )
        added += 1

    try:
        tip = subprocess.run(
            ["git", "rev-parse", "--short", ref],
            capture_output=True,
            text=True,
            check=True,
        ).stdout.strip()
        meta["last_synced"] = tip
        meta["last_synced_date"] = datetime.now(UTC).strftime("%Y-%m-%d")
    except Exception:  # noqa: BLE001
        pass

    write_ledger(jl, meta, records)
    md_path(root).write_text(render_md(meta, records))
    print(
        f"sync: {added} new untriaged from {args.base}..{ref}; "
        f"last synced -> {meta.get('last_synced', '?')} ({len(records)} rows total). "
        "Triage the new rows, then commit triage.jsonl + triage.md."
    )
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)

    p = sub.add_parser("migrate", help="parse hand-written triage.md -> triage.jsonl (one-time)")
    p.add_argument("source", help="path to the current triage.md")
    p.set_defaults(func=cmd_migrate)

    p = sub.add_parser("generate", help="render triage.jsonl -> triage.md")
    p.add_argument("--check", action="store_true", help="fail if triage.md is stale (for CI)")
    p.set_defaults(func=cmd_generate)

    p = sub.add_parser("reconcile", help="drain journal, mark landed, render, stage")
    p.set_defaults(func=cmd_reconcile)

    p = sub.add_parser("sync", help="add new base..ref commits as untriaged")
    p.add_argument("--base", default="dev", help="base branch already in the fork (default: dev)")
    p.add_argument("--ref", default="upstream/main", help="upstream ref (default: upstream/main)")
    p.add_argument("--no-fetch", action="store_true", help="skip `git fetch` of the remote")
    p.set_defaults(func=cmd_sync)

    p = sub.add_parser("lookup", help="print a SHA's disposition")
    p.add_argument("sha")
    p.set_defaults(func=cmd_lookup)

    p = sub.add_parser("check-reject", help="exit 3 iff SHA is 'wont-merge'")
    p.add_argument("sha")
    p.set_defaults(func=cmd_check_reject)

    p = sub.add_parser("set", help="edit or add a ledger row")
    p.add_argument("sha")
    p.add_argument("--disposition", choices=DISPOSITIONS)
    p.add_argument("--pr", type=int)
    p.add_argument("--subject")
    p.add_argument("--reason")
    p.add_argument("--branch")
    p.set_defaults(func=cmd_set)

    args = parser.parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main())