♻️ Extract changelog checks into scripts/changelog.py

Move the inline programs out of the update-changelog skill into
scripts/changelog.py (check-merged, cross-ref, report) and shrink
the skill to the workflow alone.

Teach scripts/gh.py to retry transient HTTP 504s, resolve lookups
in batches of 50 (new issue subcommand, milestone on prs output),
and report 💥 entries without the breaking change label.

Add the nine missing 2.17.0 changelog entries found while
trying the new flow.

AI-assisted-by: muse-spark-1.3
This commit is contained in:
Andrey Antukh 2026-10-06 16:26:37 +00:00
parent ff7a66f519
commit 4bdded86be
7 changed files with 1036 additions and 826 deletions

File diff suppressed because it is too large Load Diff

View File

@ -55,6 +55,7 @@ The memory is structured in a way that you can get the critical information abou
- `scripts/psql` — PostgreSQL client wrapper with devenv defaults. Companion: `scripts/db-schema` for DDL dumps. See `mem:scripts/psql`.
- `scripts/taiga.py` — Fetch public issues, user stories, and tasks from the Penpot Taiga project without authentication. See `mem:scripts/taiga`.
- `scripts/gh.py` — GitHub operations helper: list milestone issues, fetch PR details, compare against CHANGES.md. Requires `gh` CLI. See `mem:scripts/gh`.
- `scripts/changelog.py` — Changelog checks for the `update-changelog` skill: `check-merged` (PR merge status), `cross-ref` (milestone PRs vs changelog section), `report` (anomaly report to CHANGES-ISSUES.md). Calls `scripts/gh.py` via subprocess. See `mem:scripts/changelog`.
- `scripts/error-reports.mjs` — Query error reports via RPC API with token authentication. Supports list/get operations with filtering and pagination. See `mem:scripts/error-reports`.
- `scripts/clean-node-modules` — Remove stale `node_modules` from all pnpm workspaces (root, modules, member packages). Keeps the shared pnpm store at `<repo>/.pnpm-store` unless `--store`; ignores `external/` and `.opencode/`. Usage and reinstall steps: `mem:workflow/updating-pnpm`.
- `scripts/ci` — CI orchestration script: runs lint, tests, and format checks per module (`frontend backend common render-wasm exporter mcp plugins library`). Logs go to `.ci-logs/`; read the log file on failure. See `mem:scripts/ci`.

View File

@ -0,0 +1,18 @@
# Changelog helper
`scripts/changelog.py` holds the checks used by the `update-changelog` skill (extracted from the skill file so the skill stays workflow-only). Calls `scripts/gh.py` via subprocess; never call the GitHub API directly from this script.
## Subcommands
- `check-merged <PR...>` (`--file`, `--stdin` accepted) — warn on any non-merged PR, exit 1 when found; run before writing changelog entries.
- `cross-ref <MILESTONE> [--changes CHANGES.md]` — fetch milestone PRs with `--state all` once, report merged PRs missing from the version section plus CLOSED (unmerged) warnings; GHSA entries never appear here by design.
- `report <MILESTONE> [--changes CHANGES.md --output CHANGES-ISSUES.md]` — overwrite the anomaly report (milestone mismatches type A/B plus `:rocket:` gaps C/D); rule violations never belong in the report, they are fixed in CHANGES.md first.
## Conventions
- Milestone resolution is batched, never per-item: unknown PRs go through `gh.py prs <numbers>` and unknown issues through `gh.py issue <numbers>` (one GraphQL call per 50 items); a lookup miss resolves to null, never to an extra call.
- Version-section lookup accepts an optional `(suffix)` after the version header (e.g. `(Unreleased)`); a missing section is an error for `cross-ref` and an empty section for `report`.
- PR/issue reference parsing covers standard `[#N](.../pull/N)` / `[#N](.../issues/N)` links plus legacy `PR:[N]` / `[Github #N]` forms.
- Shared parsing helpers (`extract_version_section`, `extract_subsection`, `collect_changelog_prs/issues`) live at module top so future subcommands reuse them instead of duplicating regexes.
- Report anomaly types: A/B milestone mismatches, E `:boom:` entry whose issue lacks the `breaking change` label (counted, entries preserved — human labels the issue or moves the entry); C/D are `:rocket:` gaps (warnings, never counted).

View File

@ -53,7 +53,7 @@ python3 scripts/gh.py issues "2.16.0" --compare CHANGES.md
### `prs`
Fetch PR details by number or by milestone.
Fetch PR details by number or by milestone. Every entry includes a `milestone` field (title or null).
```bash
# Fetch specific PRs
@ -74,6 +74,12 @@ python3 scripts/gh.py prs --milestone "2.16.0" --state all
**Output**: JSON array to stdout; progress to stderr.
### `issue`
Fetch issue details by number in batches of 50 (same input styles as `prs`: numbers, `--file`, `--stdin`). No filters. A dead number fails its whole batch (unlike PRs, issues error instead of resolving to null), so the batch falls back to one-by-one lookups and reports misses as `{"number", "error": "not_found"}`. Any other failure aborts.
**Output**: JSON array to stdout; progress to stderr.
### `link-issue`
Explicitly assign a GitHub issue to a pull request and verify the relationship from both sides:
@ -111,7 +117,10 @@ python3 scripts/gh.py advisories GHSA-xvj6-fh9w-gjw7
## Key principles
- Changelog-specific checks (merge-status verification, milestone-vs-changelog cross-reference, anomaly report) live in `scripts/changelog.py`, which calls this script via subprocess — see `mem:scripts/changelog`.
- All output is JSON — pipe into `jq` or other tools for further processing.
- Transient HTTP 504 responses from `gh` are retried automatically (3 attempts, 5s/15s backoff); any other error fails fast with no retry.
- Milestone lookup is by exact title match.
- `issues` subcommand auto-paginates (100 items per page).
- `prs` subcommand batches PR number lookups (50 per GraphQL query).

View File

@ -257,6 +257,7 @@
- Use hard reload for render engine switching in the workspace menu [#10441](https://github.com/penpot/penpot/issues/10441) (PR: [#10444](https://github.com/penpot/penpot/pull/10444))
- Rotate size badge when shape is rotated [#10386](https://github.com/penpot/penpot/issues/10386) (PR: [#10393](https://github.com/penpot/penpot/pull/10393))
- Add separate internal URI for exporter to handle Docker deployments where internal and public URIs differ [#10627](https://github.com/penpot/penpot/issues/10627) (PR: [#10630](https://github.com/penpot/penpot/pull/10630))
- Make throwValidationErrors default to true for v2 manifest plugins [#10401](https://github.com/penpot/penpot/issues/10401) (PR: [#10433](https://github.com/penpot/penpot/pull/10433))
### :bug: Bugs fixed
@ -364,6 +365,14 @@
- Fix text shape position-data to include required fills in WASM and DOM calculation paths [#10646](https://github.com/penpot/penpot/issues/10646) (PR: [#10650](https://github.com/penpot/penpot/pull/10650))
- Log expired OIDC tokens as auth failures instead of server errors [#10635](https://github.com/penpot/penpot/issues/10635) (PR: [#10636](https://github.com/penpot/penpot/pull/10636))
- Return 400 instead of 500 when ImageMagick rejects invalid uploaded images [#10642](https://github.com/penpot/penpot/issues/10642) (PR: [#10643](https://github.com/penpot/penpot/pull/10643))
- Fix Plugin API board.guides setting and clearing throwing malli invalid-schema error (by @filipsajdak) [#9773](https://github.com/penpot/penpot/issues/9773) (PR: [#10503](https://github.com/penpot/penpot/pull/10503))
- Fix Plugin API rejecting open-overlay and toggle-overlay interactions when position is not set (by @mvanhorn) [#10201](https://github.com/penpot/penpot/issues/10201) (PR: [#10503](https://github.com/penpot/penpot/pull/10503))
- Fix Plugin API close-overlay interaction requiring animation field (by @mvanhorn) [#10202](https://github.com/penpot/penpot/issues/10202) (PR: [#10503](https://github.com/penpot/penpot/pull/10503))
- Fix boards hidden from View Mode reappearing when adding prototype interactions (by @jeffrey701) [#10289](https://github.com/penpot/penpot/issues/10289) (PR: [#9695](https://github.com/penpot/penpot/pull/9695))
- Fix Plugin API PenpotUtils.createVariantContainer assigning variant properties to wrong components [#10506](https://github.com/penpot/penpot/issues/10506) (PR: [#10562](https://github.com/penpot/penpot/pull/10562))
- Fix mask inside flex layout shifting position or size when child image visibility changes [#10537](https://github.com/penpot/penpot/issues/10537) (PR: [#10697](https://github.com/penpot/penpot/pull/10697))
- Fix path ends selector missing arrow icons and incorrect divider colors [#10593](https://github.com/penpot/penpot/issues/10593) (PR: [#10631](https://github.com/penpot/penpot/pull/10631))
- Fix webhook form prefilling hardcoded metadata URI default [#10722](https://github.com/penpot/penpot/issues/10722) (PR: [#10723](https://github.com/penpot/penpot/pull/10723))
## 2.16.2

616
scripts/changelog.py Normal file
View File

@ -0,0 +1,616 @@
#!/usr/bin/env python3
"""
changelog.py — Helper commands for the ``update-changelog`` skill.
Uses ``scripts/gh.py`` (via subprocess) and the local ``CHANGES.md`` so the
skill file itself stays free of inline programs.
Subcommands:
check-merged Verify that every given PR is merged (warns otherwise)
cross-ref Compare merged milestone PRs against the changelog section
report Generate the anomaly report (CHANGES-ISSUES.md)
Usage:
python3 scripts/changelog.py check-merged 9179 9204 9311
cat prs.txt | python3 scripts/changelog.py check-merged --stdin
python3 scripts/changelog.py cross-ref "2.16.0"
python3 scripts/changelog.py report "2.16.0"
python3 scripts/changelog.py report "2.16.0" --changes CHANGES.md --output CHANGES-ISSUES.md
Prerequisites:
- gh CLI authenticated (gh auth status)
- Python 3.8+
"""
import argparse
import json
import re
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
REPO = "penpot/penpot"
GH_PY = Path(__file__).resolve().parent / "gh.py"
# ─────────────────────────────────────────────
# Shared helpers
# ─────────────────────────────────────────────
def run_gh_py(*args: str, input_text: str | None = None) -> str:
"""Run ``scripts/gh.py`` with the given arguments, return stdout."""
cmd = [sys.executable, str(GH_PY), *args]
result = subprocess.run(cmd, input=input_text, capture_output=True, text=True)
if result.returncode != 0:
print(f"gh.py error: {result.stderr}", file=sys.stderr)
sys.exit(1)
return result.stdout
def read_pr_numbers(args: argparse.Namespace) -> list[int]:
"""Collect PR numbers from positional args, --file, and/or --stdin."""
numbers: list[int] = []
if args.numbers:
numbers.extend(args.numbers)
if args.file:
with open(args.file) as f:
for line in f:
line = line.strip()
if line:
numbers.append(int(line))
if args.stdin:
for line in sys.stdin:
line = line.strip()
if line:
numbers.append(int(line))
seen: set[int] = set()
return [n for n in numbers if not (n in seen or seen.add(n))]
def fetch_prs_by_numbers(pr_numbers: list[int]) -> list[dict]:
"""Fetch PR details via ``gh.py prs`` for explicit PR numbers."""
stdout = run_gh_py("prs", *[str(n) for n in pr_numbers])
return json.loads(stdout)
def fetch_milestone_prs(milestone: str, state: str) -> list[dict]:
"""Fetch all PRs in a milestone via ``gh.py prs --milestone``."""
stdout = run_gh_py("prs", "--milestone", milestone, "--state", state)
return json.loads(stdout)
def extract_version_section(content: str, milestone: str) -> str:
"""Return the changelog body of a ``## <VERSION>`` section (or "")."""
match = re.search(
rf"^## {re.escape(milestone)}(?:\s*\([^)]*\))?\n(.*?)(?=^## |\Z)",
content,
re.DOTALL | re.MULTILINE,
)
return match.group(1) if match else ""
def extract_subsection(section: str, heading_re: str) -> str:
"""Return the body of a ``### <heading>`` subsection (or "")."""
match = re.search(rf"^{heading_re}", section, re.MULTILINE)
if not match:
return ""
body = section[match.end():]
return re.split(r"(?m)^#{2,3}\s", body)[0]
def collect_changelog_prs(section: str) -> set[int]:
"""Collect all PR numbers referenced in a changelog section."""
numbers = set()
for num in re.findall(
r"\[#(\d+)\]\(https://github\.com/penpot/penpot/pull/\d+\)", section
):
numbers.add(int(num))
for num in re.findall(r"PR:\[(\d+)\]", section):
numbers.add(int(num))
return numbers
def collect_changelog_issues(section: str) -> set[int]:
"""Collect all issue numbers referenced in a changelog section."""
numbers = set()
for num in re.findall(
r"\[#(\d+)\]\(https://github\.com/penpot/penpot/issues/\d+\)", section
):
numbers.add(int(num))
for num in re.findall(r"\[Github #(\d+)\]", section):
numbers.add(int(num))
return numbers
# ─────────────────────────────────────────────
# Subcommand: check-merged
# ─────────────────────────────────────────────
def cmd_check_merged(args: argparse.Namespace) -> None:
"""Warn about any given PR that is not merged. Exits 1 when found."""
pr_numbers = read_pr_numbers(args)
if not pr_numbers:
print(
"ERROR: no PR numbers provided (pass numbers, --file, or --stdin)",
file=sys.stderr,
)
sys.exit(1)
prs = fetch_prs_by_numbers(pr_numbers)
bad = 0
for pr in prs:
number = pr.get("number")
state = pr.get("state")
if state != "MERGED":
bad += 1
print(f"WARNING: #{number} is {state} (not merged)")
if bad:
print(f"{bad} of {len(prs)} PRs are NOT merged", file=sys.stderr)
sys.exit(1)
print(f"OK: all {len(prs)} PRs are merged", file=sys.stderr)
# ─────────────────────────────────────────────
# Subcommand: cross-ref
# ─────────────────────────────────────────────
def cmd_cross_ref(args: argparse.Namespace) -> None:
"""Compare merged milestone PRs against the changelog section."""
with open(args.changes) as f:
content = f.read()
section = extract_version_section(content, args.milestone)
if not section:
print(
f'ERROR: no "## {args.milestone}" section found in {args.changes}',
file=sys.stderr,
)
sys.exit(1)
changelog_refs = collect_changelog_prs(section)
milestone_prs = fetch_milestone_prs(args.milestone, "all")
merged = {pr["number"] for pr in milestone_prs if pr.get("state") == "MERGED"}
closed = [pr for pr in milestone_prs if pr.get("state") == "CLOSED"]
missing = sorted(merged - changelog_refs)
print(f"Milestone merged PRs: {len(merged)}")
print(f"Changelog referenced PRs: {len(changelog_refs)}")
print(f"PRs in milestone but NOT in changelog: {len(missing)}")
for num in missing:
pr = next(p for p in milestone_prs if p["number"] == num)
print(f" #{num} {pr['title'][:80]}")
if closed:
print("WARNING: CLOSED (unmerged) PRs in milestone:")
for pr in closed:
print(f" #{pr['number']} {pr['title'][:80]}")
# ─────────────────────────────────────────────
# Subcommand: report
# ─────────────────────────────────────────────
def issue_url(n: int) -> str:
return f"https://github.com/{REPO}/issues/{n}"
def pr_url(n: int) -> str:
return f"https://github.com/{REPO}/pull/{n}"
def issue_link(n: int) -> str:
return f"[#{n}]({issue_url(n)})"
def pr_link(n: int) -> str:
return f"[#{n}]({pr_url(n)})"
def issue_link_title(n: int, title: str) -> str:
url = issue_url(n)
if title:
return f"[#{n}]({url}) — [{title}]({url})"
return f"[#{n}]({url})"
def pr_link_title(n: int, title: str) -> str:
url = pr_url(n)
if title:
return f"[#{n}]({url}) — [{title}]({url})"
return f"[#{n}]({url})"
def cmd_report(args: argparse.Namespace) -> None:
"""Generate the changelog anomaly report and write it to a file."""
milestone = args.milestone
all_issues = json.loads(run_gh_py("issues", milestone, "--state", "all"))
issue_by_num = {i["number"]: i for i in all_issues}
all_prs = fetch_milestone_prs(milestone, "all")
pr_by_num = {p["number"]: p for p in all_prs}
with open(args.changes) as f:
content = f.read()
section = extract_version_section(content, milestone)
changelog_issues = collect_changelog_issues(section)
changelog_prs = collect_changelog_prs(section)
# PRs and issues returned by milestone queries are KNOWN to be in the
# milestone. Everything else is resolved in bulk via batched gh.py
# lookups (one GraphQL call per 50 items, never one call per item).
pr_milestone_cache = {p["number"]: milestone for p in all_prs}
issue_milestone_cache = {i["number"]: milestone for i in all_issues}
unknown_prs = set()
for issue in issue_by_num.values():
unknown_prs.update(issue.get("closing_prs", []))
unknown_prs -= set(pr_milestone_cache)
if unknown_prs:
print(
f"Resolving milestones for {len(unknown_prs)} PRs outside {milestone}...",
file=sys.stderr,
)
for pr in json.loads(run_gh_py("prs", *[str(n) for n in sorted(unknown_prs)])):
pr_milestone_cache[pr["number"]] = pr.get("milestone")
unknown_issues = set()
for pr in pr_by_num.values():
unknown_issues.update(pr.get("closing_issues", []))
unknown_issues |= collect_changelog_issues(extract_subsection(section, r"### :boom:"))
unknown_issues -= set(issue_milestone_cache)
external_issues: dict[int, dict] = {}
if unknown_issues:
print(
f"Resolving milestones for {len(unknown_issues)} issues outside {milestone}...",
file=sys.stderr,
)
for issue in json.loads(
run_gh_py("issue", *[str(n) for n in sorted(unknown_issues)])
):
issue_milestone_cache[issue["number"]] = issue.get("milestone")
if "error" not in issue:
external_issues[issue["number"]] = issue
def get_issue_labels(issue_num: int) -> list[str] | None:
"""Return the labels of an issue, or None when they cannot be known."""
issue = issue_by_num.get(issue_num)
if issue is not None:
return issue.get("labels", [])
issue = external_issues.get(issue_num)
if issue is not None:
return issue.get("labels", [])
return None
def get_pr_milestone(pr_num: int) -> str | None:
"""Return the milestone title for a PR, or None if unassigned."""
return pr_milestone_cache.get(pr_num)
def get_issue_milestone(issue_num: int) -> str | None:
"""Return the milestone title for an issue, or None if unassigned."""
return issue_milestone_cache.get(issue_num)
excluded_labels = {"release blocker", "no changelog"}
def issue_excluded(issue: dict | None) -> bool:
if not issue:
return True
if issue.get("state") != "CLOSED":
return True
if issue.get("issue_type") in {"Task"}:
return True
if issue.get("project_status") in {"Rejected"}:
return True
if excluded_labels & set(issue.get("labels", [])):
return True
return False
# Type A: issue in milestone, referenced PR in different milestone or none
anomalies_a = []
for issue_num in sorted(changelog_issues):
issue = issue_by_num.get(issue_num)
if not issue:
continue
if get_issue_milestone(issue_num) != milestone:
continue
for pr_num in issue.get("closing_prs", []):
pr_ms = get_pr_milestone(pr_num)
if pr_ms != milestone:
anomalies_a.append(
{
"issue": issue_num,
"issue_title": issue.get("title", ""),
"pr": pr_num,
"pr_milestone": pr_ms,
}
)
# Type B: PR in milestone, the issue it closes is in a different milestone.
# An issue with NO milestone belongs to another (probably private) project
# and is NOT an anomaly.
anomalies_b = []
for pr_num in sorted(changelog_prs):
pr = pr_by_num.get(pr_num)
if not pr:
continue
if get_pr_milestone(pr_num) != milestone:
continue
for issue_num in pr.get("closing_issues", []):
issue_ms = get_issue_milestone(issue_num)
if issue_ms is None:
continue
if issue_ms != milestone:
anomalies_b.append(
{
"pr": pr_num,
"pr_title": pr.get("title", ""),
"issue": issue_num,
"issue_milestone": issue_ms,
}
)
# Type E: :boom: entry whose issue lacks the `breaking change` label.
# These stay in the changelog (they are preserved, not removed) — the
# human decides whether to label the issue or move the entry elsewhere.
anomalies_e = []
boom_body = extract_subsection(section, r"### :boom:")
if boom_body:
for issue_num in sorted(collect_changelog_issues(boom_body)):
labels = get_issue_labels(issue_num)
if labels is None:
continue
if "breaking change" not in labels:
issue = issue_by_num.get(issue_num) or external_issues.get(issue_num) or {}
anomalies_e.append(
{
"issue": issue_num,
"issue_title": issue.get("title", ""),
}
)
# Type C: released X.Y.0 sections without a :rocket: subsection.
# Patches (X.Y.Z with Z != 0) never carry :rocket: — only minors/majors.
anomalies_c = []
rocket_heading_re = re.compile(r"^### :rocket:", re.MULTILINE)
version_sections = re.split(
r"(?=^## \d+\.\d+\.\d+)", content, flags=re.MULTILINE
)
for vs in version_sections:
m = re.match(r"^## (\d+\.\d+\.\d+)(.*)", vs)
if not m:
continue
ver, suffix = m.group(1), m.group(2)
if "unreleased" in suffix.lower():
continue
if ver.split(".")[2] != "0":
continue
if not rocket_heading_re.search(vs):
anomalies_c.append(ver)
# Type D: :rocket: entries without issue AND PR references.
anomalies_d = []
issue_ref_re = re.compile(
r"\[#\d+\]\(https://github\.com/penpot/penpot/issues/\d+\)"
)
pr_ref_re = re.compile(
r"\(PR:\s*\[#\d+\]\(https://github\.com/penpot/penpot/pull/\d+\)"
r"(\s*,\s*\[#\d+\]\(https://github\.com/penpot/penpot/pull/\d+\))*\)"
)
for vs in version_sections:
m = re.match(r"^## (\d+\.\d+\.\d+)(.*)", vs)
if not m:
continue
ver = m.group(1)
rocket_body = extract_subsection(vs, r"### :rocket:")
if not rocket_body:
continue
for line in rocket_body.splitlines():
line = line.strip()
if line.startswith("- ") and not (
issue_ref_re.search(line) and pr_ref_re.search(line)
):
anomalies_d.append({"version": ver, "line": line[:100]})
def fmt_ms(ms: str | None) -> str:
return ms if ms else "_none_"
with open(args.output, "w") as f:
f.write(f"# Changelog Anomaly Report — {milestone}\n\n")
f.write(
f"Generated: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}\n\n"
)
f.write("---\n\n")
n_a, n_b, n_c, n_d, n_e = (
len(anomalies_a),
len(anomalies_b),
len(anomalies_c),
len(anomalies_d),
len(anomalies_e),
)
f.write("## Summary\n\n")
f.write(
f"- **Issue in {milestone}, referenced PR in different milestone or no milestone:** {n_a}\n"
)
f.write(
f"- **PR in {milestone}, closing issue in a different milestone:** {n_b}\n"
)
f.write(f"- **:boom: entry without breaking change label:** {n_e}\n")
f.write(f"- **Total anomalies:** {n_a + n_b + n_e}\n")
f.write(
f"- **Released X.Y.0 version missing :rocket: section (gap):** {n_c}\n"
)
f.write(f"- **:rocket: entry without issue AND PR references (gap):** {n_d}\n\n")
if n_a or n_b or n_e:
f.write("## Anomalies\n\n")
f.write(
"Types A and B are milestone mismatches between an issue in the changelog "
"and its referenced PR (or vice-versa). The changelog claim "
'"this issue is fixed by this PR, all in this milestone" is '
"inconsistent with the actual milestone assignments. "
"Resolve by either updating the milestone on the issue/PR or "
"removing the misleading entry from the changelog.\n\n"
"Type E entries stay in the changelog: either label the issue "
"as `breaking change` or move the entry out of `:boom:`.\n\n"
)
if n_a:
f.write(
f"### Issue in {milestone}, PR in different milestone or no milestone\n\n"
)
by_issue: dict[int, list[dict]] = {}
for a in anomalies_a:
by_issue.setdefault(a["issue"], []).append(a)
for issue_num in sorted(by_issue):
entries = by_issue[issue_num]
title = entries[0]["issue_title"]
f.write(f"- {issue_link_title(issue_num, title[:80])}\n")
for e in entries:
ms_label = fmt_ms(e["pr_milestone"])
badge = "🔴" if e["pr_milestone"] is None else "⚠️"
f.write(
f" - {badge} Referenced {pr_link(e['pr'])} is in milestone **{ms_label}** (expected: {milestone})\n"
)
f.write("\n")
if n_b:
f.write(
f"\n### PR in {milestone}, closing issue in a different milestone\n\n"
)
by_pr: dict[int, list[dict]] = {}
for b in anomalies_b:
by_pr.setdefault(b["pr"], []).append(b)
for pr_num in sorted(by_pr):
entries = by_pr[pr_num]
title = entries[0]["pr_title"]
f.write(f"- {pr_link_title(pr_num, title[:80])}\n")
for e in entries:
ms_label = fmt_ms(e["issue_milestone"])
badge = "🔴" if e["issue_milestone"] is None else "⚠️"
f.write(
f" - {badge} Closing {issue_link(e['issue'])} is in milestone **{ms_label}** (expected: {milestone})\n"
)
f.write("\n")
if n_e:
f.write("\n### :boom: entry without breaking change label\n\n")
f.write(
"These entries sit under `### :boom: Breaking changes & Deprecations` "
"but their issue has no `breaking change` label. They are kept "
"in the changelog — add the label on the issue or move the "
"entry to its regular section.\n\n"
)
for e in anomalies_e:
f.write(f"- ⚠️ {issue_link_title(e['issue'], e['issue_title'][:80])}\n")
f.write("\n")
else:
f.write(
"✅ No anomalies found. All (issue, PR) pairs in the changelog have aligned milestone assignments.\n\n"
)
if n_c or n_d:
f.write("## Highlight gaps\n\n")
f.write(
"These are warnings, not anomalies: they do not affect the "
"milestone-mismatch total above. They track `:rocket:` coverage "
"across all released X.Y.0 versions. Historical entries (e.g. "
"Taiga links) predate the current reference convention and are "
"expected to appear here.\n\n"
)
if n_c:
f.write("### Released X.Y.0 version missing :rocket: section\n\n")
f.write(
"These released minors/majors have no `### :rocket: Epics and highlights` subsection. "
"Add highlights to help self-hosted users understand what they are missing.\n\n"
)
for ver in anomalies_c:
f.write(f"- Version **{ver}**\n")
f.write("\n")
if n_d:
f.write("### :rocket: entry without issue AND PR references\n\n")
f.write(
"These highlight entries lack the required issue AND PR references. "
"Add `[#ISSUE](...)` and `(PR: [#PR](...))` links.\n\n"
)
for d in anomalies_d:
f.write(f"- **{d['version']}**: `{d['line']}`\n")
f.write("\n")
elif not (n_a or n_b or n_e):
f.write(
"✅ No highlight gaps found. All released X.Y.0 versions have properly referenced :rocket: entries.\n\n"
)
f.write("---\n\n")
f.write("## Context\n\n")
f.write(f"- Milestone: **{milestone}**\n")
f.write(f"- Milestone total issues (all states): {len(all_issues)}\n")
f.write(
f"- Closed issues in milestone: {sum(1 for i in all_issues if i.get('state') == 'CLOSED')}\n"
)
f.write(
f"- Valid issues after exclusions: {len([i for i in all_issues if not issue_excluded(i)])}\n"
)
f.write(f"- Issues referenced in changelog: {len(changelog_issues)}\n")
f.write(f"- PRs referenced in changelog: {len(changelog_prs)}\n")
print(f"Anomaly report written to {args.output}")
# ─────────────────────────────────────────────
# CLI entrypoint
# ─────────────────────────────────────────────
def main() -> None:
parser = argparse.ArgumentParser(
description="Helper commands for the update-changelog skill"
)
sub = parser.add_subparsers(dest="command", required=True, title="subcommands")
p_check = sub.add_parser(
"check-merged", help="Verify that every given PR is merged"
)
p_check.add_argument("numbers", type=int, nargs="*",
help="PR numbers to check (space-separated)")
p_check.add_argument("--file", type=str,
help="File with one PR number per line")
p_check.add_argument("--stdin", action="store_true",
help="Read PR numbers from stdin (one per line)")
p_check.set_defaults(func=cmd_check_merged)
p_cross = sub.add_parser(
"cross-ref",
help="Compare merged milestone PRs against the changelog section",
)
p_cross.add_argument("milestone", help="Milestone title (e.g. '2.16.0')")
p_cross.add_argument("--changes", default="CHANGES.md",
help="Path to the changelog file (default: CHANGES.md)")
p_cross.set_defaults(func=cmd_cross_ref)
p_report = sub.add_parser(
"report", help="Generate the anomaly report (CHANGES-ISSUES.md)"
)
p_report.add_argument("milestone", help="Milestone title (e.g. '2.16.0')")
p_report.add_argument("--changes", default="CHANGES.md",
help="Path to the changelog file (default: CHANGES.md)")
p_report.add_argument("--output", default="CHANGES-ISSUES.md",
help="Report output path (default: CHANGES-ISSUES.md)")
p_report.set_defaults(func=cmd_report)
args = parser.parse_args()
args.func(args)
if __name__ == "__main__":
main()

View File

@ -25,6 +25,7 @@ Usage:
cat prs.txt | python3 scripts/gh.py prs --stdin
python3 scripts/gh.py prs --milestone "2.16.0" (default: state=merged)
python3 scripts/gh.py prs --milestone "2.16.0" --state all
python3 scripts/gh.py issue 11235 11236
python3 scripts/gh.py advisories (list all advisories)
python3 scripts/gh.py advisories --severity critical (filter by severity)
python3 scripts/gh.py advisories GHSA-xvj6-fh9w-gjw7 (single advisory detail)
@ -40,6 +41,7 @@ import json
import re
import subprocess
import sys
import time
from typing import Any
@ -47,21 +49,70 @@ REPO = "penpot/penpot"
OWNER = "penpot"
REPO_NAME = "penpot"
# Transient gateway timeouts from the GitHub API are retried, nothing else.
GH_RETRIES = 3
GH_RETRY_DELAYS = (5, 15) # seconds waited before retry N (last delay repeats)
def run_gh_command(cmd: list[str], input_text: str | None = None) -> str:
"""Run a ``gh`` command, retrying transient HTTP 504s.
Raises `GhCommandFailed` on failure so callers can choose between
aborting (normal commands) and degrading (batched lookups).
"""
last_stderr = ""
for attempt in range(GH_RETRIES):
if attempt:
delay = GH_RETRY_DELAYS[min(attempt - 1, len(GH_RETRY_DELAYS) - 1)]
print(
f"gh: HTTP 504, retrying in {delay}s"
f" (attempt {attempt + 1}/{GH_RETRIES})...",
file=sys.stderr,
)
time.sleep(delay)
result = subprocess.run(
cmd, input=input_text, capture_output=True, text=True
)
if result.returncode == 0:
return result.stdout
last_stderr = result.stderr
if "504" not in result.stderr:
break
raise GhCommandFailed(last_stderr)
class GhCommandFailed(Exception):
"""A ``gh`` invocation failed (stderr kept for the caller to judge)."""
def __init__(self, stderr: str) -> None:
super().__init__(stderr)
self.stderr = stderr
def fail_gh(stderr: str) -> None:
"""Report a ``gh`` failure and exit (standard behavior for CLI commands)."""
print(f"gh error: {stderr}", file=sys.stderr)
sys.exit(1)
# ─────────────────────────────────────────────
# Shared helpers
# ─────────────────────────────────────────────
def post_graphql(query: str, variables: dict) -> Any:
"""POST a GraphQL query via ``gh`` and return the raw response body."""
payload = json.dumps({"query": query, "variables": variables})
stdout = run_gh_command(["gh", "api", "graphql", "--input", "-"], payload)
return json.loads(stdout)
def run_gh_graphql(query: str, variables: dict) -> Any:
"""Run a GraphQL query via ``gh api graphql --input -``."""
payload = json.dumps({"query": query, "variables": variables})
cmd = ["gh", "api", "graphql", "--input", "-"]
result = subprocess.run(cmd, input=payload, capture_output=True, text=True)
if result.returncode != 0:
print(f"gh error: {result.stderr}", file=sys.stderr)
sys.exit(1)
body = json.loads(result.stdout)
try:
body = post_graphql(query, variables)
except GhCommandFailed as err:
fail_gh(err.stderr)
if "errors" in body:
for err in body["errors"]:
print(f"GraphQL error: {err.get('message')}", file=sys.stderr)
@ -71,12 +122,10 @@ def run_gh_graphql(query: str, variables: dict) -> Any:
def run_gh_rest(path: str) -> Any:
"""Run a REST API call via ``gh api``."""
cmd = ["gh", "api", path]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f"gh error: {result.stderr}", file=sys.stderr)
sys.exit(1)
return json.loads(result.stdout)
try:
return json.loads(run_gh_command(["gh", "api", path]))
except GhCommandFailed as err:
fail_gh(err.stderr)
# ─────────────────────────────────────────────
@ -541,6 +590,7 @@ GQL_PRS_QUERY_ITEM = """\
state
mergedAt
createdAt
milestone {{ title }}
author {{ login }}
labels(first: 20) {{ nodes {{ name }} }}
closingIssuesReferences(first: 5) {{ nodes {{ number }} }}
@ -588,6 +638,7 @@ def fetch_prs_batch(pr_numbers: list[int]) -> list[dict]:
"state": pr["state"],
"merged_at": pr.get("mergedAt"),
"created_at": pr.get("createdAt"),
"milestone": (pr.get("milestone") or {}).get("title"),
"author": pr["author"]["login"] if pr["author"] else None,
"labels": [lbl["name"] for lbl in pr["labels"]["nodes"]],
"closing_issues": [iss["number"] for iss in pr["closingIssuesReferences"]["nodes"]],
@ -692,38 +743,19 @@ def cmd_prs(args: argparse.Namespace) -> None:
print(f"Fetching {args.state} PRs via GraphQL...", file=sys.stderr)
prs = fetch_milestone_prs(ms["number"], gql_states)
for pr in prs:
pr["milestone"] = ms["title"]
print(f"Fetched {len(prs)} PRs total", file=sys.stderr)
print(json.dumps(prs, indent=2))
return
# ── Number-based path ───────────────────────────────────────────
pr_numbers: list[int] = []
if args.numbers:
pr_numbers.extend(args.numbers)
if args.file:
with open(args.file) as f:
for line in f:
line = line.strip()
if line:
pr_numbers.append(int(line))
if args.stdin:
for line in sys.stdin:
line = line.strip()
if line:
pr_numbers.append(int(line))
pr_numbers = read_numbers_from_args(args)
if not pr_numbers:
print("ERROR: no PR numbers provided (pass numbers, --file, --stdin, or --milestone)",
file=sys.stderr)
sys.exit(1)
# Deduplicate while preserving order
seen: set[int] = set()
pr_numbers = [n for n in pr_numbers if not (n in seen or seen.add(n))]
print(f"Fetching {len(pr_numbers)} PRs in batches of {PRS_BATCH_SIZE}...",
file=sys.stderr)
@ -737,6 +769,159 @@ def cmd_prs(args: argparse.Namespace) -> None:
print(json.dumps(all_results, indent=2))
# ─────────────────────────────────────────────
# Subcommand: issue (fetch issues by number, batched)
# ─────────────────────────────────────────────
GQL_ISSUE_BY_NUMBER_QUERY_ITEM = """\
issue_{num}: issue(number: {num}) {{
number
title
state
milestone {{ title }}
issueType {{ name }}
labels(first: 20) {{ nodes {{ name }} }}
closedByPullRequestsReferences(first: 5) {{ nodes {{ number }} }}
projectItems(first: 10) {{
nodes {{
project {{ title }}
fieldValueByName(name: "Status") {{
... on ProjectV2ItemFieldSingleSelectValue {{
name
}}
}}
}}
}}
}}
"""
def issue_node_to_dict(node: dict) -> dict:
"""Convert a GraphQL issue node to the shared issue dict shape."""
issue_type = node.get("issueType")
ms = node.get("milestone")
project_status = None
for pi in (node.get("projectItems") or {}).get("nodes") or []:
project = pi.get("project") or {}
if project.get("title") == "Main":
status_field = pi.get("fieldValueByName") or {}
project_status = status_field.get("name")
break
return {
"number": node["number"],
"title": node["title"],
"state": node["state"],
"milestone": ms["title"] if ms else None,
"issue_type": issue_type["name"] if issue_type else None,
"labels": [lbl["name"] for lbl in node["labels"]["nodes"]],
"closing_prs": [pr["number"] for pr in node["closedByPullRequestsReferences"]["nodes"]],
"project_status": project_status,
}
class IssueBatchFailed(Exception):
"""One aliased issue lookup failed; the batch cannot be trusted as a whole."""
def _fetch_issues_batch(issue_numbers: list[int]) -> list[dict]:
"""Single batched issue lookup; raises `IssueBatchFailed` on any failure."""
items = "\n".join(
GQL_ISSUE_BY_NUMBER_QUERY_ITEM.format(num=n) for n in issue_numbers
)
query = GQL_PRS_QUERY_WRAPPER.format(items=items)
variables = {"owner": OWNER, "repo": REPO_NAME}
try:
body = post_graphql(query, variables)
except GhCommandFailed as err:
raise IssueBatchFailed(err.stderr) from err
if "errors" in body:
raise IssueBatchFailed("; ".join(e.get("message", "") for e in body["errors"]))
repo = body["data"]["repository"]
results: list[dict] = []
for num in issue_numbers:
node = repo.get(f"issue_{num}")
if node is None:
results.append({
"number": num,
"error": "not_found",
})
continue
results.append(issue_node_to_dict(node))
return results
def fetch_issues_batch(issue_numbers: list[int]) -> list[dict]:
"""
Fetch details for a list of issue numbers in a single GraphQL query.
Uses numbered aliases (issue_1234, …) so each issue is looked up by
number in one round-trip. Returns entries in the same order as the input.
Unlike PRs (which resolve to null), a dead issue number fails the whole
batch, so when the failure names an unresolvable issue each number is
retried on its own and reported as ``not_found`` instead of aborting.
Any other failure (auth, outage, exhausted 504 retries) still aborts.
"""
try:
return _fetch_issues_batch(issue_numbers)
except IssueBatchFailed as err:
if "Could not resolve" not in str(err):
print(f"GraphQL error: {err}", file=sys.stderr)
sys.exit(1)
print(" batch lookup failed, retrying issues one by one...",
file=sys.stderr)
results: list[dict] = []
for num in issue_numbers:
try:
results.extend(_fetch_issues_batch([num]))
except IssueBatchFailed:
results.append({"number": num, "error": "not_found"})
return results
def read_numbers_from_args(args: argparse.Namespace) -> list[int]:
"""Collect numbers from positional args, --file, and/or --stdin (deduplicated)."""
numbers: list[int] = []
if args.numbers:
numbers.extend(args.numbers)
if args.file:
with open(args.file) as f:
for line in f:
line = line.strip()
if line:
numbers.append(int(line))
if args.stdin:
for line in sys.stdin:
line = line.strip()
if line:
numbers.append(int(line))
seen: set[int] = set()
return [n for n in numbers if not (n in seen or seen.add(n))]
def cmd_issue(args: argparse.Namespace) -> None:
"""Handle the ``issue`` subcommand (batched by-number lookup, no filters)."""
issue_numbers = read_numbers_from_args(args)
if not issue_numbers:
print("ERROR: no issue numbers provided (pass numbers, --file, or --stdin)",
file=sys.stderr)
sys.exit(1)
print(f"Fetching {len(issue_numbers)} issues in batches of {PRS_BATCH_SIZE}...",
file=sys.stderr)
all_results: list[dict] = []
for i in range(0, len(issue_numbers), PRS_BATCH_SIZE):
batch = issue_numbers[i : i + PRS_BATCH_SIZE]
print(f" batch {i // PRS_BATCH_SIZE + 1}: issues {batch[0]}..{batch[-1]}",
file=sys.stderr)
all_results.extend(fetch_issues_batch(batch))
print(json.dumps(all_results, indent=2))
# ─────────────────────────────────────────────
# Subcommand: advisories
# ─────────────────────────────────────────────
@ -911,6 +1096,24 @@ def main() -> None:
p_link.add_argument("pr_number", type=int, help="Pull request number")
p_link.set_defaults(func=cmd_link_issue)
# --- issue (by-number lookup) ---
p_issue = sub.add_parser(
"issue", help="Fetch details for one or more issues by number (batched)"
)
p_issue.add_argument(
"numbers", type=int, nargs="*",
help="Issue numbers to fetch (space-separated)"
)
p_issue.add_argument(
"--file", type=str,
help="File with one issue number per line"
)
p_issue.add_argument(
"--stdin", action="store_true",
help="Read issue numbers from stdin (one per line)"
)
p_issue.set_defaults(func=cmd_issue)
# --- advisories ---
p_adv = sub.add_parser("advisories", help="List or inspect GitHub security advisories")
p_adv.add_argument(