fix: close an area's stranded reports once a live one is open

progress/apply.py

@@ -22,6 +22,11 @@
 "our": anyone may open a pull request on a `progress/*` branch, and branch names are a pure function
 of the window, so honouring a stranger's would let them decide what this operator is allowed to
 publish -- permanently, by opening and closing one pull request.
+
+Reconciling the window is not enough on its own, because an area can hold more than one open report
+and the others are not reconciled by anything. Once ours is open we therefore close the area's
+reports that can no longer merge -- see `superseded_prs` for the two ways that happens. Without it
+the repository grows one unmergeable report per area per round and never sheds them.
 """
 
 import json
@@ -143,6 +148,142 @@ def existing_pr(branch, repo=gh.ROADMAP_REPO, owner=None, states=("merged",)):
     return rows[0] if rows else None
 
 
+def report_meta(body):
+    """The `tauceti-progress-pr:v1` metadata a report carries in its body, or None.
+
+    `pr_body` writes `from_sha`, `to_sha` and the window's PR numbers there, so a sibling states its
+    own window exactly rather than in the seven-character abbreviations of a branch name. That is
+    what makes a containment test possible without a TauCeti clone.
+
+    The body is mutable and anyone may edit it, so nothing here is trusted for a decision that could
+    destroy work: the caller only ever uses it to prove that a report covers LESS than ours, and a
+    forged wider claim merely spares a report from being closed.
+    """
+    marker = BODY_MARKER.split("{}")[0]
+    for line in (body or "").splitlines():
+        line = line.strip()
+        if not line.startswith(marker):
+            continue
+        blob = line[len(marker):]
+        blob = blob[:blob.rfind("-->")] if "-->" in blob else blob
+        try:
+            meta = json.loads(blob)
+        except json.JSONDecodeError:
+            return None
+        return meta if isinstance(meta, dict) else None
+    return None
+
+
+def superseded_prs(open_prs, area, live_cursor, plan_cursor, keep_branch, owners, our_prs=None):
+    """Open reports for `area` that this run's report replaces: `[(row, reason)]`.
+
+    **Nothing is swept unless we are the live report.** `live_cursor` is read from the roadmap
+    repository now; `plan_cursor` is where the cursor was when this run planned. If they differ, a
+    sibling landed in between and it is OUR report that is dead, not anyone else's. Sweeping then
+    inverts the rule below and closes the one report that can still merge, so the answer is to sweep
+    nothing and let the next round reconcile.
+
+    Given that, two things put a report beyond saving, and neither closes itself.
+
+    **Its window no longer starts at the cursor.** The merge gate requires a byte-exact append at the
+    cursor recorded in `PROGRESS.md`, so a report whose `from_sha` is not that cursor can never
+    satisfy it -- dead by construction, not merely behind. This is what happens the moment a sibling
+    for the same area merges: the cursor advances and every other open report is orphaned instantly.
+    Nothing noticed, so they accumulated.
+
+    **It starts at the cursor and we cover everything it covers.** Proved, not inferred: same
+    `from_sha`, and its window's PR numbers are a proper subset of ours. Creation order will not do
+    this job. A worker that planned early and stalled creates a NARROW report late, so "newest wins"
+    would close a wider report and lose coverage that had already been written; and neither a
+    timestamp nor a pull request number says anything about which window contains which. Containment
+    is antisymmetric, so two workers racing cannot each conclude the other is dominated, and the
+    stalled-worker case resolves correctly whichever of them opens first.
+
+    A sibling whose window is incomparable to ours is left alone. So is every report, if `our_prs` is
+    unknown: closing then would assert a containment we have not established, and a duplicate left
+    for the next round is far cheaper than a report discarded.
+
+    Scoped to reports we could have opened, like every other judgement in this module. Branch names
+    are a pure function of the window, so a stranger can create one; closing theirs would let this
+    operator silently veto someone else's contribution. The cost is real and worth naming: reports
+    published from another operator's fork are never swept, so this does not by itself eliminate a
+    multi-operator pile-up.
+    """
+    if not live_cursor or not plan_cursor or live_cursor != plan_cursor:
+        return []
+    ours = set(our_prs) if our_prs is not None else None
+    out = []
+    for row in open_prs:
+        parts = (row.get("headRefName") or "").split("/")
+        if len(parts) != 3 or parts[-1] != area:
+            continue
+        if row.get("headRefName") == keep_branch:
+            continue
+        if ((row.get("headRepositoryOwner") or {}).get("login") or "") not in owners:
+            continue
+        from7 = parts[1].split("-")[0] if "-" in parts[1] else ""
+        if not from7:
+            continue
+        if not live_cursor.startswith(from7):
+            out.append((row, f"its window starts at {from7}, but the {area} cursor is now "
+                             f"{live_cursor[:7]}, so it can never append"))
+            continue
+        if ours is None:
+            continue
+        meta = report_meta(row.get("body"))
+        if not meta or meta.get("from_sha") != live_cursor:
+            continue
+        try:
+            theirs = {int(n) for n in (meta.get("prs") or [])}
+        except (TypeError, ValueError):
+            continue
+        if theirs < ours:
+            out.append((row, f"its window covers {len(theirs)} of the {len(ours)} pull requests "
+                             f"this one reports, from the same cursor {from7}"))
+    return out
+
+
+def close_superseded(rows, replacement_url):
+    """Close each superseded report, saying what replaced it. Returns the number that failed.
+
+    A failure here is never fatal -- this only runs once the replacement is open, so the window is
+    published either way -- but it is returned rather than swallowed, because the caller has to keep
+    retrying. `sweep` is what makes that retry actually happen.
+    """
+    failed = 0
+    for row, reason in rows:
+        note = f"Superseded by {replacement_url}: {reason}."
+        try:
+            gh.gh(["pr", "close", str(row["number"]), "--repo", gh.ROADMAP_REPO,
+                   "--comment", note])
+            print(f"closed superseded #{row['number']}: {reason}")
+        except gh.GhError as exc:
+            failed += 1
+            print(f"could not close superseded #{row['number']} ({exc}); leaving it open")
+    return failed
+
+
+def sweep(plan, keep_branch, replacement_url):
+    """Close the area's stranded reports. Safe to call whenever our report is open.
+
+    Idempotent and deliberately reachable from BOTH paths in `run` -- the one that has just opened
+    the report and the one that finds it already open. That is not tidiness: a sweep that only ran
+    on the create path would never be retried at all, because the next run for the same window stops
+    at the in-flight check long before reaching it. A cleanup outage would then leave the pile-up
+    forever while every run reported success.
+
+    Only `gh` failures are caught. A parsing or programming error here is a bug and must surface.
+    """
+    live = files.cursor(gh.file_on_default_branch(plan["progress_path"]) or "") or ""
+    rows = superseded_prs(
+        gh.open_progress_prs(), plan["roadmap"], live, plan["from_sha"], keep_branch,
+        {gh.ROADMAP_REPO.split("/")[0], _own_login()}, our_prs=plan.get("prs"),
+    )
+    if not rows:
+        return 0
+    return close_superseded(rows, replacement_url)
+
+
 def remote_branch_exists(roadmap_dir, branch, remote="origin"):
     proc = _run(["git", "ls-remote", "--exit-code", "--heads", remote, branch],
                roadmap_dir, check=False)
@@ -248,6 +389,12 @@ def run(plan, status_body_file, section_body_file, roadmap_dir, dry_run=False, v
     open_pr = own_pr(branch, states=("open",))
     if open_pr is not None:
         print(f"already open, in flight: {open_pr['url']}")
+        # Sweep here too. This is the path every run after the first takes, so it is the only one
+        # that can retry a sweep an earlier run failed or never reached.
+        try:
+            sweep(plan, branch, open_pr.get("url") or "the open report")
+        except gh.GhError as exc:
+            print(f"could not sweep superseded reports ({exc}); the report is open regardless")
         return EX_NOPROGRESS
 
     # A CLOSED pull request means this window was refused, and reopening it every day is the loop
@@ -337,4 +484,18 @@ def run(plan, status_body_file, section_body_file, roadmap_dir, dry_run=False, v
         "--title", title, "--body", body,
     ])
     print(out.strip())
+
+    # Only now that ours is open: sweep the area's dead and superseded reports. Ordered this way so
+    # a failure here costs a tidy-up, never a publication -- and so we never close the only open
+    # report for an area.
+    #
+    # Without this the repository accumulates one unmergeable report per area per round, from two
+    # directions: a sibling merging orphans every other report for that area instantly, and any
+    # repository-wide breakage (a red `main` fails every report's `build`) makes each round open a
+    # fresh one that also cannot merge, since the planner's staleness expiry stops the previous one
+    # marking the area in flight. Neither ever closes itself. See `superseded_prs`.
+    try:
+        sweep(plan, branch, out.strip().splitlines()[-1] if out.strip() else "the new report")
+    except gh.GhError as exc:
+        print(f"could not sweep superseded reports ({exc}); the new report is open regardless")
     return 0

progress/gh.py

@@ -13,6 +13,7 @@
   right now".
 """
 
+import base64
 import json
 import subprocess
 import time
@@ -140,7 +141,24 @@ def open_progress_prs(repo=ROADMAP_REPO, branch_prefix="progress/"):
     out = gh([
         "pr", "list", "--repo", repo, "--state", "open",
         "--limit", "200", "--json",
-        "number,headRefName,title,url,createdAt,headRepositoryOwner",
+        "number,headRefName,title,url,createdAt,headRepositoryOwner,body",
     ])
     rows = json.loads(out)
     return [r for r in rows if (r.get("headRefName") or "").startswith(branch_prefix)]
+
+
+def file_on_default_branch(path, repo=ROADMAP_REPO, ref="main"):
+    """The text of `path` on `ref`, or None if it is not there.
+
+    Read through the API rather than from a clone on purpose. A worker's checkout is a snapshot from
+    whenever it started, and the questions this answers -- where is the cursor NOW, is this report
+    still the live one -- are exactly the ones a stale snapshot gets wrong.
+    """
+    try:
+        raw = gh(["api", f"repos/{repo}/contents/{path}?ref={ref}", "--jq", ".content"])
+    except GhError:
+        return None
+    try:
+        return base64.b64decode("".join(raw.split())).decode("utf-8", "surrogateescape")
+    except (ValueError, UnicodeDecodeError):
+        return None