marketplace/plugins/tea/skills/sync/scripts/close.py

#!/usr/bin/env python3
"""
close.py — change an issue's state in Gitea, and in the local copy with it.

The one regular tracker operation that used to have no script: closing. Without
it the only way to move `state:` was a raw `tea api -X PATCH -d '{"state":
"closed"}' repos/OWNER/REPO/issues/N`, which spells out the owner, the repo and
the request body — the three things `_gitea.py` exists to hide — and which needs
`Bash(tea api *)`, a permission that also covers `-X DELETE` on the repository.

    close.py wire-sqlc-appclick            one issue, by slug
    close.py wire-sqlc-appclick 42 #43     several, by slug or number
    close.py --reopen 42                   the same thing backwards
    close.py --dry-run 42 43               what would happen, no request at all

STATE ONLY. This script sends `{"state": …}` and nothing else: no title, no
body, no labels, no milestone. Editing an issue is `pull.py` -> edit ->
`push.py --update`; closing it is not an edit.

**What may be named.** A local slug, or a Gitea key (`42`, `#42`,
`owner/repo#42`, an issue URL) — the same forms `pull.py` takes. Both are
needed, and for the same reason: a push deletes the local file, so most issues
in the tracker have no slug on disk to name them by. A slug is resolved through
the file's `gitea:` field when the file is there, and through the ledger
(`.remote.json`) when push has already dropped it.

**An `origin: local` issue cannot be closed.** It is not in the tracker, so
there is nothing to close there, and the run stops naming the id rather than
quietly editing one field of a local file. Delete it, or push it first.

**Explicit ids only.** No `--milestone`, no `--label`, no "close everything
that looks done". Which issues are finished is a judgement about content; this
script only carries it out, one named id at a time. Nothing here deletes an
issue either — Gitea can, and it is not an operation of this workflow.

The local file is written only after the tracker has confirmed the write:

    1. `tea` ran and exited 0 (a non-2xx exits the run inside `_gitea.api`), and
    2. the answer is an object carrying the very number that was PATCHed, and
    3. its `state` is the state we asked for.

Anything else and the file is left exactly as it was — see `confirmed`. An
issue whose local copy is gone (pushed and dropped) is closed in Gitea and
nothing is written; the state comes down with the next `pull.py`.

Gitea refuses to close an issue that its own dependency graph still blocks. That
refusal arrives as a non-2xx and stops the run with the tracker's own words:
close the blockers first, or unlink them in the web UI.

Login: the operator's pin from .claude/settings.local.json (see /tea:auth).
"""
import argparse
import os
import re
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path[:0] = [_HERE, os.path.normpath(os.path.join(_HERE, "..", "..", "issue", "scripts"))]

import _gitea  # noqa: E402
import issue  # noqa: E402
import issue_index  # noqa: E402
import map as gmap  # noqa: E402

# What `_gitea.parse_key` accepts, asked as a question instead of an assertion:
# parse_key exits on anything it cannot read, and here "not a key" is the normal
# case — it means the argument is a slug. A slug never contains `#`, `/` or `:`,
# so the two vocabularies cannot collide.
KEY_RE = re.compile(r'^(#?\d+|[\w.-]+/[\w.-]+#\d+|https?://\S+)$')


def looks_like_key(arg):
    return bool(KEY_RE.match((arg or "").strip()))


def ledger_pairs(remote_map, repo=None):
    """[(repo, number, slug)] from `.remote.json`, filtered to `repo`.

    A `--repo` that was not given means "whatever the ledger holds": resolving
    the repo's real name costs a request, and a dry run is required to make
    none. The ambiguity that opens — one number under two repos — is caught at
    lookup time rather than papered over."""
    out = []
    for key, slug in sorted(remote_map.items()):
        r, n = gmap.parse_remote_key(key)
        if n:
            if repo is None or r == repo:
                out.append((r, n, slug))
    return out


def one(candidates, what, arg):
    """The single `(repo, value)` in `candidates`, None when empty, or exit.

    Two answers mean the ledger knows this number (or this slug) under more than
    one repository, and only `--repo` can settle that."""
    got = sorted(set(candidates))
    if len(got) > 1:
        _gitea.die("%r matches %s under more than one repo (%s) — pass "
                   "--repo owner/repo" % (arg, what, ", ".join(r for r, _v in got)))
    return got[0] if got else None


def resolve(arg, issues, pairs):
    """(id, number, repo) for one argument. Either of `id` and `repo` is None
    when nothing this machine holds names it.

    Order, and it is the order of what is most authoritative about this machine:
    a file on disk, then the ledger, then nothing. A key skips straight to the
    ledger — its number is already the tracker's answer, and the slug is only
    wanted so the local copy, if there is one, can be kept honest.

    `repo` travels out with the number because a key may name one
    (`owner/repo#42`) and a `gitea:` field always does. Sending a foreign key to
    whatever repo the CWD happens to be in would close somebody else's issue of
    the same number, so the caller reconciles them before anything goes out."""
    if looks_like_key(arg):
        number, repo = _gitea.parse_key(arg)
        hit = one([(r, s) for r, n, s in pairs
                   if n == number and (repo is None or r == repo)], "a slug", arg)
        return (hit[1] if hit else None), number, repo or (hit[0] if hit else None)

    iss = issues.get(arg)
    if iss is not None:
        repo, number = gmap.parse_remote_key(iss.extra.get("gitea", ""))
        if not number:
            _gitea.die("%s is not in the tracker (origin: %s, no gitea: field) — "
                       "there is no state there to change; push.py %s first"
                       % (arg, iss.origin, arg))
        return arg, number, repo

    hit = one([(r, n) for r, n, s in pairs if s == arg], "a number", arg)
    if hit:
        return arg, hit[1], hit[0]  # pushed, and its file went with the push
    _gitea.die("no issue %r in the store or the ledger — pass a Gitea number "
               "(42, #42, owner/repo#42, a URL) to close one this machine has "
               "never seen" % arg)


def confirmed(got, number, state):
    """True when the tracker's answer confirms THIS write, and nothing else.

    The gate in front of the local write, and deliberately boring: an answer
    counts only when it is an object carrying the very number that was PATCHed
    (`bool` rejected explicitly — `True` is an `int`) and the state that was
    asked for. A non-2xx and a `tea` that would not run never reach here at all;
    `_gitea.api` exits on both, so the file survives those by never being
    written."""
    if not isinstance(got, dict):
        return False
    n = got.get("number")
    if isinstance(n, bool) or not isinstance(n, int) or n != number:
        return False
    return got.get("state") == state


def apply_state(root, iss, state, got):
    """Write the confirmed state onto the local file; return its path.

    `state:` is the domain's own field, so it is set on the issue and written
    out by the domain's own writer. The sync-owned freshness fields travel with
    it: the answer that authorized this write is also the newest thing the
    tracker has said about the issue, so `synced:` and `remote-updated:` are
    stamped from it rather than left describing an older read."""
    iss.state = state
    iss.extra["synced"] = _gitea.now_iso()
    if got.get("updated_at"):
        iss.extra["remote-updated"] = got["updated_at"]
    return issue.save(root, iss)


def main():
    ap = argparse.ArgumentParser(description="Close (or reopen) issues in Gitea")
    ap.add_argument("ids", nargs="+",
                    help="local ids, or Gitea keys: 42, #42, owner/repo#42, URL")
    ap.add_argument("--reopen", action="store_true",
                    help="set the state back to open instead of closed")
    ap.add_argument("--dry-run", action="store_true",
                    help="print what would change; makes no request at all")
    ap.add_argument("--repo", help="owner/repo (default: auto-detect from CWD git remote)")
    ap.add_argument("--out", default=issue.ISSUE_ROOT,
                    help="store root (default: <project>/.tea/issues)")
    args = ap.parse_args()

    root = args.out
    state = "open" if args.reopen else "closed"
    verb = "reopen" if args.reopen else "close"
    past = "reopened" if args.reopen else "closed"

    # A store that is not there is not an error here: a number needs no local
    # file, and closing an issue whose copy was dropped by push is the normal
    # case. `load_all` reads an absent directory as an empty one.
    issues = issue.load_all(root)
    pairs = ledger_pairs(_gitea.load_map(root), args.repo)

    # Every argument is resolved before anything is sent, so a typo in the third
    # id does not leave the first two closed.
    targets = []
    for arg in args.ids:
        got = resolve(arg, issues, pairs)
        if got not in targets:
            targets.append(got)

    # One run, one repo. An explicit --repo is the operator's word and wins;
    # without one, the repo comes from what the ids themselves said, and two
    # answers are a question rather than a guess — `repo_base` would otherwise
    # let `tea` fill the blank from the CWD and close the wrong #42.
    named = {r for _i, _n, r in targets if r}
    if not args.repo and len(named) > 1:
        _gitea.die("all ids must belong to one repo, got: %s" % ", ".join(sorted(named)))
    repo_arg = args.repo or (sorted(named)[0] if named else None)

    if args.dry_run:
        for id, number, _repo in targets:
            iss = issues.get(id)
            where = ("%s (state: %s)" % (issue.path_of(root, id), iss.state)
                     if iss is not None else "no local copy")
            print("would %s  %s  #%d  — %s" % (verb, id or "?", number, where))
        print("%d issue(s) would be %s; no request was made"
              % (len(targets), past))
        return

    login = _gitea.require_login()
    base = _gitea.repo_base(repo_arg)

    touched = 0
    for id, number, _repo in targets:
        got = _gitea.api(login, "%s/issues/%d" % (base, number), "PATCH",
                         {"state": state}, payload_name="state-%d" % number)
        # The gate. Above it nothing local has been written; below it the file
        # is about to say something the tracker had better agree with.
        if not confirmed(got, number, state):
            _gitea.die("#%d: %s failed — the tracker's answer does not confirm the "
                       "write (%.200r). Nothing local was changed."
                       % (number, verb, got))

        print("%s  %s  #%d  %s" % (past, id or "?", number,
                                   got.get("html_url", "")))

        iss = issues.get(id)
        if iss is None:
            print("    no local copy — pull.py %d to get one" % number)
            continue
        print("    state: %s  %s" % (state, apply_state(root, iss, state, got)))
        touched += 1

    if touched:
        path, n = issue_index.build(root)
        print("index: %s — %d issue(s)" % (path, n))


if __name__ == "__main__":
    main()