Migrate from CLAUDE.md to AGENTS.md

claude-to-agents.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = ["httpx2>=2.8"]
# ///
"""Migrate CLAUDE.md -> AGENTS.md across a user's or org's repos via the GitHub API.

Finds candidates with the REST Git Trees API (one recursive call per repo),
reads contents and commits via GraphQL. In every directory of the default branch:

  * CLAUDE.md is a file, AGENTS.md absent:
      CLAUDE.md is deleted and AGENTS.md added with identical content.
  * CLAUDE.md is a file, AGENTS.md a symlink to it:
      handled in two commits so the symlink is fully removed before the file
      is recreated: (1/2) delete CLAUDE.md and the AGENTS.md symlink,
      (2/2) add AGENTS.md as a regular file.
  * CLAUDE.md is a symlink to a regular AGENTS.md in the same directory:
      CLAUDE.md is deleted.
  * Anything ambiguous is skipped and reported.

A repo gets one commit, or two if any directory has an AGENTS.md symlink.
If the second commit can't be made, the new AGENTS.md contents are written
under ./recovery/<owner>/<repo>/ so nothing is lost.

With --pr, commits go on a new branch (default: agent-md) cut from the default
branch, and a pull request is opened against it. The default branch is never
touched. Repos where that branch already exists are skipped.

Dry run by default; pass --apply to commit.

Usage:
  ./claude-to-agents.py skwashd
  ./claude-to-agents.py proactiveops --prefix tflint- --prefix picofun
  ./claude-to-agents.py davehallconsulting --only some-repo --apply
  ./claude-to-agents.py skwashd --skip-dir fixtures -v
  ./claude-to-agents.py proactiveops --pr --apply

Auth: GITHUB_TOKEN / GH_TOKEN, falling back to `gh auth token`.
Needs Contents: write (fine-grained) or `repo` (classic).
"""

from __future__ import annotations

import argparse
import base64
import os
import posixpath
import re
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path

import httpx2

GRAPHQL = "https://api.github.com/graphql"
GRAPHQL_HEADERS = {"Accept": "application/vnd.github.v4.idl"}  # avoids 502s
REST = "https://api.github.com"
REST_HEADERS = {
    "Accept": "application/vnd.github+json",
    "X-GitHub-Api-Version": "2022-11-28",
}

SRC, DST = "CLAUDE.md", "AGENTS.md"
COMMIT_MSG = "CLAUDE.md -> AGENTS.md"
SPLIT_MSG_1 = "CLAUDE.md -> AGENTS.md (1/2): remove old files"
SPLIT_MSG_2 = "CLAUDE.md -> AGENTS.md (2/2): add AGENTS.md"
UNLINK_MSG = "Remove CLAUDE.md symlink"
PR_TITLE = "CLAUDE.md -> AGENTS.md"
DEFAULT_PR_BRANCH = "agent-md"
RECOVERY_DIR = Path("recovery")

# REST tree modes are strings.
FILE_MODES = {"100644", "100755"}
SYMLINK = "120000"

DEFAULT_SKIP_DIRS = {
    "node_modules", "vendor", ".venv", "venv", ".tox", "__pycache__",
    ".terraform", "dist", "build", "target",
}

BLOB_BATCH = 50
OID_RE = re.compile(r"[0-9a-f]{40}|[0-9a-f]{64}")

REPO_FIELDS = """
fragment RepoFields on Repository {
  id
  name
  owner { login }
  isArchived
  isFork
  isEmpty
  defaultBranchRef {
    name
    target { ... on Commit { oid tree { oid } } }
  }
  prRef: ref(qualifiedName: $prRef) {
    associatedPullRequests(states: OPEN, first: 1) { nodes { url } }
  }
}
"""

LIST_REPOS = """
query($login: String!, $cursor: String, $prRef: String!) {
  repositoryOwner(login: $login) {
    repositories(first: 100, after: $cursor, orderBy: {field: NAME, direction: ASC}) {
      pageInfo { hasNextPage endCursor }
      nodes { ...RepoFields }
    }
  }
}
""" + REPO_FIELDS

GET_REPO = """
query($owner: String!, $name: String!, $prRef: String!) {
  repository(owner: $owner, name: $name) { ...RepoFields }
}
""" + REPO_FIELDS

HEAD_OID = """
query($owner: String!, $name: String!, $ref: String!) {
  repository(owner: $owner, name: $name) { ref(qualifiedName: $ref) { target { oid } } }
}
"""

CREATE_REF = """
mutation($input: CreateRefInput!) { createRef(input: $input) { ref { id } } }
"""

DELETE_REF = """
mutation($input: DeleteRefInput!) { deleteRef(input: $input) { clientMutationId } }
"""

CREATE_PR = """
mutation($input: CreatePullRequestInput!) {
  createPullRequest(input: $input) { pullRequest { url } }
}
"""

CREATE_COMMIT = """
mutation($input: CreateCommitOnBranchInput!) {
  createCommitOnBranch(input: $input) { commit { oid url tree { oid } } }
}
"""


class GraphQLError(RuntimeError):
    pass


class TreeTruncated(RuntimeError):
    pass


@dataclass
class Change:
    dir: str                  # "" for repo root
    kind: str                 # "rename" | "unlink"
    contents: bytes | None = None
    dst_symlink: bool = False # rename where AGENTS.md is currently a symlink

    def describe(self) -> str:
        if self.kind == "rename":
            via = " (replacing symlink)" if self.dst_symlink else ""
            return f"{at(self.dir, SRC)} -> {at(self.dir, DST)}{via}"
        return f"remove {at(self.dir, SRC)} symlink"


@dataclass
class RepoResult:
    repo: str
    status: str    # changed | pr-opened | would-change | nothing | skip | error
    detail: str = ""
    changes: list[Change] = field(default_factory=list)
    skips: list[str] = field(default_factory=list)


def at(d: str, f: str) -> str:
    return f"{d}/{f}" if d else f


def link_points_to(link_dir: str, target: str, want: str) -> bool:
    """Does a symlink in link_dir with this target resolve to repo path `want`?"""
    target = target.strip()
    if not target or target.startswith("/"):
        return False
    return posixpath.normpath(posixpath.join(link_dir, target)) == want


def get_token() -> str:
    for var in ("GITHUB_TOKEN", "GH_TOKEN"):
        if tok := os.environ.get(var):
            return tok
    try:
        return subprocess.run(
            ["gh", "auth", "token"], check=True, capture_output=True, text=True
        ).stdout.strip()
    except (FileNotFoundError, subprocess.CalledProcessError):
        sys.exit("No token: set GITHUB_TOKEN/GH_TOKEN or log in with `gh auth login`.")


class GitHub:
    def __init__(self, token: str) -> None:
        self.http = httpx2.Client(
            headers={"Authorization": f"Bearer {token}"}, timeout=60.0
        )

    def gql(self, query: str, **variables) -> dict:
        resp = self.http.post(
            GRAPHQL,
            json={"query": query, "variables": variables},
            headers=GRAPHQL_HEADERS,
        )
        resp.raise_for_status()
        body = resp.json()
        if body.get("errors"):
            raise GraphQLError("; ".join(e.get("message", str(e)) for e in body["errors"]))
        return body["data"]

    def list_repos(self, login: str, pr_branch: str) -> list[dict]:
        repos, cursor = [], None
        while True:
            owner = self.gql(
                LIST_REPOS, login=login, cursor=cursor, prRef=f"refs/heads/{pr_branch}"
            )["repositoryOwner"]
            if owner is None:
                sys.exit(f"No user or org named {login!r}.")
            conn = owner["repositories"]
            # A user's repo connection can include collaborations; keep owned only.
            repos += [
                r for r in conn["nodes"]
                if r["owner"]["login"].lower() == login.lower()
            ]
            if not conn["pageInfo"]["hasNextPage"]:
                return repos
            cursor = conn["pageInfo"]["endCursor"]

    def get_repo(self, owner: str, name: str, pr_branch: str) -> dict:
        """One repo by exact name, without listing everything the owner has."""
        return self.gql(
            GET_REPO, owner=owner, name=name, prRef=f"refs/heads/{pr_branch}"
        )["repository"]

    def head_oid(self, owner: str, name: str, branch: str) -> str:
        repo = self.gql(HEAD_OID, owner=owner, name=name, ref=f"refs/heads/{branch}")["repository"]
        return repo["ref"]["target"]["oid"]

    def create_branch(self, repo_id: str, branch: str, oid: str) -> str:
        return self.gql(
            CREATE_REF,
            input={"repositoryId": repo_id, "name": f"refs/heads/{branch}", "oid": oid},
        )["createRef"]["ref"]["id"]

    def delete_branch(self, ref_id: str) -> None:
        self.gql(DELETE_REF, input={"refId": ref_id})

    def create_pr(self, repo_id: str, base: str, head: str, title: str, body: str) -> str:
        return self.gql(
            CREATE_PR,
            input={"repositoryId": repo_id, "baseRefName": base, "headRefName": head,
                   "title": title, "body": body},
        )["createPullRequest"]["pullRequest"]["url"]

    def tree(self, owner: str, name: str, tree_sha: str) -> list[dict]:
        """Whole tree in one call. Pinned to a tree SHA, so it can't drift from HEAD."""
        resp = self.http.get(
            f"{REST}/repos/{owner}/{name}/git/trees/{tree_sha}",
            params={"recursive": "1"},
            headers=REST_HEADERS,
        )
        resp.raise_for_status()
        body = resp.json()
        if body.get("truncated"):
            raise TreeTruncated("recursive tree truncated by GitHub; repo too large to scan safely")
        return body["tree"]

    def read_blobs(self, owner: str, name: str, oids: list[str]) -> dict[str, dict]:
        """Fetch many blobs in one GraphQL query per batch, using aliases."""
        out: dict[str, dict] = {}
        uniq = sorted(set(oids))
        for oid in uniq:
            if not OID_RE.fullmatch(oid):
                raise ValueError(f"unexpected object id {oid!r}")
        for i in range(0, len(uniq), BLOB_BATCH):
            chunk = uniq[i : i + BLOB_BATCH]
            fields = "\n".join(
                f'b{j}: object(oid: "{oid}") '
                "{ ... on Blob { text byteSize isBinary isTruncated } }"
                for j, oid in enumerate(chunk)
            )
            query = (
                "query($owner: String!, $name: String!) "
                f"{{ repository(owner: $owner, name: $name) {{ {fields} }} }}"
            )
            data = self.gql(query, owner=owner, name=name)["repository"]
            for j, oid in enumerate(chunk):
                out[oid] = data[f"b{j}"] or {}
        return out

    def create_commit(self, owner: str, name: str, branch: str, head_oid: str,
                      headline: str, body: str,
                      additions: list[dict], deletions: list[dict]) -> dict:
        message = {"headline": headline}
        if body:
            message["body"] = body
        return self.gql(
            CREATE_COMMIT,
            input={
                "branch": {"repositoryNameWithOwner": f"{owner}/{name}", "branchName": branch},
                "expectedHeadOid": head_oid,
                "message": message,
                "fileChanges": {"additions": additions, "deletions": deletions},
            },
        )["createCommitOnBranch"]["commit"]


def decide(d: str, src: dict | None, dst: dict | None, blobs: dict[str, dict]) -> Change | str | None:
    """Return a Change, a skip reason, or None if there's nothing to do in this dir."""
    src_path, dst_path = at(d, SRC), at(d, DST)
    if src is None:
        return None

    if src["mode"] == SYMLINK:
        target = blobs[src["sha"]].get("text") or ""
        if not link_points_to(d, target, dst_path):
            return f"{src_path}: symlink to {target.strip()!r}, not {DST}"
        if dst is None or dst["mode"] not in FILE_MODES:
            return f"{src_path}: symlink to {DST}, but {DST} isn't a regular file"
        return Change(d, "unlink")

    if src["mode"] not in FILE_MODES:
        return f"{src_path}: unexpected mode {src['mode']}"

    blob = blobs[src["sha"]]
    if blob.get("isBinary") or blob.get("isTruncated") or blob.get("text") is None:
        return f"{src_path}: can't read as text (binary/truncated)"
    raw = blob["text"].encode("utf-8")
    if len(raw) != blob["byteSize"]:
        return f"{src_path}: byte size mismatch; refusing to rewrite"

    if dst is None:
        return Change(d, "rename", raw)
    if dst["mode"] != SYMLINK:
        return f"{dst_path}: already a regular file; resolve by hand"
    target = blobs[dst["sha"]].get("text") or ""
    if not link_points_to(d, target, src_path):
        return f"{dst_path}: symlink to {target.strip()!r}, not {SRC}"
    return Change(d, "rename", raw, dst_symlink=True)


def plan_repo(gh: GitHub, owner: str, name: str, tree_sha: str,
              skip_dirs: set[str]) -> tuple[list[Change], list[str]]:
    by_dir: dict[str, dict[str, dict]] = {}
    for e in gh.tree(owner, name, tree_sha):
        if e["type"] != "blob":
            continue
        d, base = posixpath.split(e["path"])
        if base not in (SRC, DST):
            continue
        if d and skip_dirs.intersection(d.split("/")):
            continue
        by_dir.setdefault(d, {})[base] = e

    # Only read what decisions depend on: CLAUDE.md always, AGENTS.md if it's a symlink.
    need = []
    for files in by_dir.values():
        if (src := files.get(SRC)) is None:
            continue
        need.append(src["sha"])
        if (dst := files.get(DST)) is not None and dst["mode"] == SYMLINK:
            need.append(dst["sha"])
    blobs = gh.read_blobs(owner, name, need) if need else {}

    changes, skips = [], []
    for d in sorted(by_dir):
        outcome = decide(d, by_dir[d].get(SRC), by_dir[d].get(DST), blobs)
        if isinstance(outcome, Change):
            changes.append(outcome)
        elif outcome:
            skips.append(outcome)
    return changes, skips


def additions_for(changes: list[Change]) -> list[dict]:
    return [
        {"path": at(c.dir, DST), "contents": base64.b64encode(c.contents).decode()}
        for c in changes if c.kind == "rename"
    ]


def body_for(changes: list[Change]) -> str:
    return "\n".join(f"- {c.describe()}" for c in changes) if len(changes) > 1 else ""


def save_recovery(owner: str, name: str, changes: list[Change]) -> Path:
    root = RECOVERY_DIR / owner / name
    for c in changes:
        if c.kind == "rename":
            path = root / at(c.dir, DST)
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_bytes(c.contents)
    return root


def commit_changes(gh: GitHub, owner: str, name: str, branch: str, head_oid: str,
                   changes: list[Change], recover: bool = True) -> tuple[str, list[str]]:
    """Make one or two commits on branch. Returns (commit urls, problems found)."""
    split = any(c.dst_symlink for c in changes)

    if not split:
        adds = additions_for(changes)
        commit = gh.create_commit(
            owner, name, branch, head_oid,
            COMMIT_MSG if adds else UNLINK_MSG, body_for(changes),
            additions=adds,
            deletions=[{"path": at(c.dir, SRC)} for c in changes],
        )
        urls = [commit["url"]]
    else:
        # 1/2: remove every CLAUDE.md we're handling, plus the AGENTS.md symlinks.
        deletions = [{"path": at(c.dir, SRC)} for c in changes]
        deletions += [{"path": at(c.dir, DST)} for c in changes if c.dst_symlink]
        first = gh.create_commit(
            owner, name, branch, head_oid, SPLIT_MSG_1, body_for(changes),
            additions=[], deletions=deletions,
        )
        # 2/2: add AGENTS.md back as regular files. Retry once against a fresh
        # head in case something landed between the two commits.
        adds = additions_for(changes)
        renames = [c for c in changes if c.kind == "rename"]
        try:
            commit = gh.create_commit(
                owner, name, branch, first["oid"], SPLIT_MSG_2, body_for(renames),
                additions=adds, deletions=[],
            )
        except (GraphQLError, httpx2.HTTPError):
            try:
                commit = gh.create_commit(
                    owner, name, branch, gh.head_oid(owner, name, branch),
                    SPLIT_MSG_2, body_for(renames), additions=adds, deletions=[],
                )
            except (GraphQLError, httpx2.HTTPError) as exc:
                missing = ", ".join(a["path"] for a in adds)
                if not recover:
                    return first["url"], [
                        f"commit 1/2 landed on {branch} but 2/2 failed ({exc}); "
                        f"{branch} is missing {missing}"
                    ]
                saved = save_recovery(owner, name, changes)
                return first["url"], [
                    f"commit 1/2 landed but 2/2 failed ({exc}); "
                    f"repo is missing {missing}; contents saved under {saved}/"
                ]
        urls = [first["url"], commit["url"]]

    # Verify the final tree: CLAUDE.md gone, AGENTS.md a regular file (not a symlink).
    after = {e["path"]: e["mode"] for e in gh.tree(owner, name, commit["tree"]["oid"])}
    problems = []
    for c in changes:
        if at(c.dir, SRC) in after:
            problems.append(f"{at(c.dir, SRC)} still present")
        if c.kind == "rename" and after.get(at(c.dir, DST)) not in FILE_MODES:
            problems.append(f"{at(c.dir, DST)} mode {after.get(at(c.dir, DST), 'missing')}")
    return " ".join(urls), problems


def pr_body(changes: list[Change], skips: list[str]) -> str:
    lines = [
        "Claude Code now reads `AGENTS.md`, so the context file moves to the "
        "cross-tool name.",
        "",
        "### Changes",
        *(f"- {c.describe()}" for c in changes),
    ]
    if skips:
        lines += ["", "### Needs a human", *(f"- {s}" for s in skips)]
    lines += ["", "_Generated by claude-to-agents.py._"]
    return "\n".join(lines)


def open_pr(gh: GitHub, owner: str, repo: dict, base: str, head_oid: str,
            pr_branch: str, changes: list[Change], skips: list[str]) -> tuple[str, list[str]]:
    """Cut pr_branch from base, commit there, open a PR. Returns (detail, problems)."""
    name = repo["name"]
    ref_id = gh.create_branch(repo["id"], pr_branch, head_oid)
    try:
        urls, problems = commit_changes(
            gh, owner, name, pr_branch, head_oid, changes, recover=False
        )
    except (GraphQLError, httpx2.HTTPError):
        # Nothing landed on the branch; don't leave an empty one behind.
        gh.delete_branch(ref_id)
        raise
    if problems:
        return f"branch {pr_branch}: {urls}", problems
    try:
        return gh.create_pr(repo["id"], base, pr_branch, PR_TITLE, pr_body(changes, skips)), []
    except (GraphQLError, httpx2.HTTPError) as exc:
        return f"branch {pr_branch}: {urls}", [f"commits pushed but PR creation failed ({exc})"]


def process_repo(gh: GitHub, owner: str, repo: dict, apply: bool,
                 skip_dirs: set[str], pr_branch: str | None) -> RepoResult:
    name = repo["name"]
    full = f"{owner}/{name}"
    if repo["isArchived"]:
        return RepoResult(full, "skip", "archived")
    ref = repo["defaultBranchRef"]
    if repo["isEmpty"] or not ref or not ref.get("target"):
        return RepoResult(full, "skip", "empty")

    branch = ref["name"]
    head_oid = ref["target"]["oid"]
    tree_sha = ref["target"]["tree"]["oid"]

    if pr_branch and repo.get("prRef") is not None:
        open_prs = repo["prRef"]["associatedPullRequests"]["nodes"]
        where = f" (open PR: {open_prs[0]['url']})" if open_prs else ""
        return RepoResult(full, "skip", f"branch {pr_branch} already exists{where}")

    changes, skips = plan_repo(gh, owner, name, tree_sha, skip_dirs)
    res = RepoResult(full, "", changes=changes, skips=skips)
    if not changes:
        res.status = "skip" if skips else "nothing"
        res.detail = "needs attention" if skips else ""
        return res
    if not apply:
        n = 2 if any(c.dst_symlink for c in changes) else 1
        where = f"PR {pr_branch} -> {branch}" if pr_branch else branch
        res.status = "would-change"
        res.detail = f"{where} @ {head_oid[:7]}, {n} commit{'s' if n > 1 else ''}"
        return res

    if pr_branch:
        detail, problems = open_pr(gh, owner, repo, branch, head_oid, pr_branch, changes, skips)
        if problems:
            res.status, res.detail = "error", f"{detail} but: " + "; ".join(problems)
        else:
            res.status, res.detail = "pr-opened", detail
        return res

    urls, problems = commit_changes(gh, owner, name, branch, head_oid, changes)
    if problems:
        res.status, res.detail = "error", f"committed {urls} but: " + "; ".join(problems)
    else:
        res.status, res.detail = "changed", urls
    return res


def main() -> int:
    p = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    p.add_argument("owner", help="GitHub user or org login")
    p.add_argument("--prefix", action="append", default=[],
                   help="only repos whose name starts with this (repeatable)")
    p.add_argument("--only", action="append", default=[],
                   help="only this exact repo name (repeatable)")
    p.add_argument("--skip-dir", action="append", default=[],
                   help=f"extra directory name to ignore (defaults: {', '.join(sorted(DEFAULT_SKIP_DIRS))})")
    p.add_argument("--include-forks", action="store_true")
    p.add_argument("--apply", action="store_true", help="actually commit (default: dry run)")
    p.add_argument("--pr", action="store_true",
                   help="commit on a new branch and open a PR instead of committing to the default branch")
    p.add_argument("--pr-branch", default=DEFAULT_PR_BRANCH,
                   help=f"branch name for --pr (default: {DEFAULT_PR_BRANCH})")
    p.add_argument("-v", "--verbose", action="store_true",
                   help="also list repos with nothing to do")
    args = p.parse_args()

    skip_dirs = DEFAULT_SKIP_DIRS | set(args.skip_dir)
    gh = GitHub(get_token())
    owner = args.owner
    pr_branch = args.pr_branch if args.pr else None

    def wanted(r: dict) -> bool:
        if r["isFork"] and not args.include_forks:
            return False
        if args.only and r["name"] not in args.only:
            return False
        if args.prefix and not any(r["name"].startswith(x) for x in args.prefix):
            return False
        return True

    mode = ("APPLY" if args.apply else "DRY RUN") + (f", PR via {pr_branch}" if pr_branch else "")
    counts: dict[str, int] = {}

    if args.only:
        # Exact names: fetch each repo directly instead of paging through them all.
        repos, lookup_errors = [], []
        for name in dict.fromkeys(args.only):
            try:
                r = gh.get_repo(owner, name, args.pr_branch)
            except (GraphQLError, httpx2.HTTPError) as exc:
                lookup_errors.append(RepoResult(f"{owner}/{name}", "error", str(exc)))
                continue
            moved_to = f"{r['owner']['login']}/{r['name']}"
            if moved_to.lower() != f"{owner}/{name}".lower():
                # GitHub follows rename/transfer redirects; make the user ask for it by its real name.
                lookup_errors.append(RepoResult(f"{owner}/{name}", "skip", f"moved to {moved_to}"))
                continue
            repos.append(r)
        selected = [r for r in repos if wanted(r)]
        print(f"[{mode}] {len(selected)} of {len(args.only)} named repos in {owner}\n")
        for res in lookup_errors:
            counts[res.status] = counts.get(res.status, 0) + 1
            print(f"{res.status:>13}  {res.repo}  {res.detail}")
    else:
        repos = gh.list_repos(owner, args.pr_branch)
        selected = [r for r in repos if wanted(r)]
        print(f"[{mode}] {len(selected)} of {len(repos)} repos in {owner}\n")

    for repo in selected:
        try:
            res = process_repo(gh, owner, repo, args.apply, skip_dirs, pr_branch)
        except (GraphQLError, TreeTruncated, ValueError, httpx2.HTTPError) as exc:
            res = RepoResult(f"{owner}/{repo['name']}", "error", str(exc))
        counts[res.status] = counts.get(res.status, 0) + 1

        if res.status == "nothing" and not args.verbose:
            continue
        print(f"{res.status:>13}  {res.repo}  {res.detail}")
        for c in res.changes:
            print(f"{'':>15}{c.kind:<7} {c.describe()}")
        for s in res.skips:
            print(f"{'':>15}{'skip':<7} {s}")

    print("\n" + ", ".join(f"{k}: {v}" for k, v in sorted(counts.items())))
    return 1 if counts.get("error") else 0


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