220 lines
9.3 KiB
Python
220 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Maintenance drift reconciliation — file or update a deduped drift issue (#192).
|
|
|
|
Called by ``.gitea/workflows/maintenance.yml`` when
|
|
``scripts/version_convergence.py`` reports drift. Advisory only: the issue
|
|
tells humans/agent to merge the pending bump PR (or land one); this script
|
|
never auto-releases without the PR flow (plan Design Decision 3).
|
|
|
|
Env-driven: GITEA_TOKEN, RELEASE_SERVER_URL, RELEASE_REPO, MASTER_VERSION,
|
|
LATEST_TAG, CONVERGENCE_JSON (the checker's stdout).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from typing import Any
|
|
import sys as _sys
|
|
_SCRIPT_DIR = str(__import__('pathlib').Path(__file__).resolve().parent)
|
|
if _SCRIPT_DIR not in _sys.path:
|
|
_sys.path.insert(0, _SCRIPT_DIR)
|
|
import gitea_auth
|
|
|
|
DRIFT_ISSUE_LABEL = "release-drift"
|
|
DRIFT_ISSUE_LABEL_COLOR = "ff0000"
|
|
|
|
|
|
def resolve_label_ids(api: Any, label_names: list[str]) -> list[int]:
|
|
"""Resolve label NAMES to the integer IDs Gitea's POST /issues requires
|
|
(spec-gate finding 6, #192; repo convention per shared/custodian.py).
|
|
|
|
A label missing from the repo is created first; if that also fails the
|
|
label is skipped — the issue still files, just unlabeled.
|
|
"""
|
|
status, labels = api("/labels?limit=50")
|
|
by_name = {l.get("name"): l.get("id") for l in (labels if isinstance(labels, list) else [])}
|
|
ids: list[int] = []
|
|
for name in label_names:
|
|
label_id = by_name.get(name)
|
|
if label_id is None:
|
|
status, created = api("/labels", "POST", {"name": name, "color": DRIFT_ISSUE_LABEL_COLOR})
|
|
label_id = created.get("id") if status in (200, 201) and isinstance(created, dict) else None
|
|
if label_id is None:
|
|
print(f"label {name!r} unavailable (resolve/create failed, status {status}) — filing without it")
|
|
continue
|
|
ids.append(label_id)
|
|
return ids
|
|
|
|
|
|
def drift_title(master_version: str, latest_tag: str) -> str:
|
|
master = f"v{master_version}" if master_version else "(unknown)"
|
|
return f"Release drift: master {master} vs tag {latest_tag}"
|
|
|
|
|
|
def drift_body(decision: dict[str, Any], master_version: str, latest_tag: str) -> str:
|
|
subjects = "\n".join(f"- `{s}`" for s in decision.get("sample", []))
|
|
anomaly = decision.get("anomaly")
|
|
anomaly_line = ""
|
|
if anomaly == "version_behind_tag":
|
|
# Spec-gate finding 3 (#192): master's version is BELOW the latest tag.
|
|
anomaly_line = (
|
|
"- anomaly: **version_behind_tag** — master's `__version__` is *below* the latest tag\n"
|
|
" (a bump merged without riding a release, or a tag moved backwards).\n"
|
|
" The safety net deliberately does not act on this state; reconcile manually.\n"
|
|
)
|
|
elif anomaly:
|
|
anomaly_line = f"- anomaly: **{anomaly}**\n"
|
|
return (
|
|
"The weekly maintenance reconciliation detected that master and the latest release tag have diverged.\n\n"
|
|
f"- master `__version__`: **{master_version or '(unknown)'}**\n"
|
|
f"- latest tag: **{latest_tag}**\n"
|
|
f"- commits since tag: **{decision.get('commits_since_tag', 0)}**\n"
|
|
f"- warranted bump (from commit subjects): **{decision.get('warranted_bump') or 'none'}**\n"
|
|
f"{anomaly_line}\n"
|
|
"Commit subjects since the tag:\n\n"
|
|
f"{subjects}\n\n"
|
|
"This issue is advisory — the reconciliation never auto-releases without the PR flow.\n"
|
|
"Next step: land a version-bump PR (or let the auto-version safety net file one),\n"
|
|
"then cut the release. See docs/release-flow.md for the two-path release flow (#192)."
|
|
)
|
|
|
|
|
|
def find_open_drift_issue(issues: list[dict[str, Any]], title: str) -> dict[str, Any] | None:
|
|
"""First open issue whose title matches exactly (dedupe; PRs are not issues)."""
|
|
for issue in issues:
|
|
if issue.get("title") == title and "pull_request" not in issue:
|
|
return issue
|
|
return None
|
|
|
|
|
|
def main(argv: list[str] | None = None, api: Any = None) -> int:
|
|
parser = argparse.ArgumentParser(description="File/update the deduped release-drift issue (#192)")
|
|
parser.add_argument("--dry-run", action="store_true", help="print title/body without any API call")
|
|
parser.add_argument(
|
|
"--create-comment", action="store_true", help="comment on the existing issue instead of a no-op update"
|
|
)
|
|
parser.add_argument(
|
|
"--close-if-resolved",
|
|
action="store_true",
|
|
help="drift is resolved (converged): close any OPEN release-drift issue with an honest resolution comment",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
|
|
decision = json.loads(os.environ["CONVERGENCE_JSON"])
|
|
master_version = os.environ.get("MASTER_VERSION", "")
|
|
latest_tag = os.environ.get("LATEST_TAG", "")
|
|
|
|
if args.close_if_resolved:
|
|
# the reconciliation detected NO drift (converged): any still-open
|
|
# release-drift issue is resolved — close it with an honest comment
|
|
# (gap found 2026-10-18: drift issues were filed but never closed
|
|
# on resolution — #2489 sat open 4h after the repair).
|
|
token = gitea_auth.get_token()
|
|
base = os.environ["RELEASE_SERVER_URL"] + "/api/v1/repos/" + os.environ["RELEASE_REPO"]
|
|
|
|
def _api( # noqa: F811 — local client for the close path
|
|
path: str, method: str = "GET", data: dict[str, Any] | None = None
|
|
) -> tuple[int, dict[str, Any] | list[dict[str, Any]]]:
|
|
req = urllib.request.Request(
|
|
base + path,
|
|
data=json.dumps(data).encode() if data is not None else None,
|
|
headers={"Authorization": "token " + token, "Content-Type": "application/json"},
|
|
method=method,
|
|
)
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
payload = r.read()
|
|
return r.status, (json.loads(payload) if payload else {})
|
|
|
|
status, issues = _api("/issues?state=open&type=issues&limit=50")
|
|
closed_any = False
|
|
for issue in issues if isinstance(issues, list) else []:
|
|
if not issue.get("title", "").startswith("Release drift:"):
|
|
continue
|
|
number = issue["number"]
|
|
comment = {
|
|
"body": (
|
|
"RESOLVED: master `__version__` and the latest release tag are converged "
|
|
f"(master {master_version} == {latest_tag}) — the drift condition is gone. "
|
|
"Auto-closed by the maintenance reconciliation (--close-if-resolved)."
|
|
)
|
|
}
|
|
_api(f"/issues/{number}/comments", "POST", comment)
|
|
st, _ = _api(f"/issues/{number}", "PATCH", {"state": "closed"})
|
|
if st == 200:
|
|
closed_any = True
|
|
print(f"closed resolved drift issue #{number}")
|
|
if not closed_any:
|
|
print("no open drift issues to close")
|
|
return 0
|
|
title = drift_title(master_version, latest_tag)
|
|
body = drift_body(decision, master_version, latest_tag)
|
|
|
|
if args.dry_run:
|
|
print(f"title: {title}")
|
|
print("---")
|
|
print(body)
|
|
return 0
|
|
|
|
token = gitea_auth.get_token()
|
|
base = os.environ["RELEASE_SERVER_URL"] + "/api/v1/repos/" + os.environ["RELEASE_REPO"]
|
|
|
|
if api is None:
|
|
|
|
def api( # noqa: F811 — default client when not injected (tests)
|
|
path: str, method: str = "GET", data: dict[str, Any] | None = None
|
|
) -> tuple[int, dict[str, Any] | list[dict[str, Any]]]:
|
|
req = urllib.request.Request(
|
|
base + path,
|
|
data=json.dumps(data).encode() if data is not None else None,
|
|
headers={"Authorization": "token " + token, "Content-Type": "application/json"},
|
|
method=method,
|
|
)
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
payload = r.read()
|
|
return r.status, (json.loads(payload) if payload else {})
|
|
|
|
status, issues = api(f"/issues?state=open&type=issues&limit=50&q={urllib.parse.quote(title)}")
|
|
if status != 200:
|
|
print(f"issue search failed ({status})")
|
|
return 1
|
|
existing = find_open_drift_issue(issues if isinstance(issues, list) else [], title)
|
|
if existing is not None:
|
|
print(f"drift issue #{existing['number']} already open — {'commenting' if args.create_comment else 'no-op'}")
|
|
if args.create_comment:
|
|
st, _ = api(f"/issues/{existing['number']}/comments", "POST", {"body": body})
|
|
if st not in (200, 201):
|
|
print(f"comment failed ({st})")
|
|
return 1
|
|
return 0
|
|
|
|
label_ids = resolve_label_ids(api, [DRIFT_ISSUE_LABEL])
|
|
st, created = api(
|
|
"/issues",
|
|
"POST",
|
|
{
|
|
"title": title,
|
|
"body": body,
|
|
"labels": label_ids,
|
|
},
|
|
)
|
|
if st not in (200, 201):
|
|
print(f"issue create failed ({st})")
|
|
return 1
|
|
number = created.get("number") if isinstance(created, dict) else None
|
|
print(f"filed drift issue #{number}: {title}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except (KeyError, json.JSONDecodeError, urllib.error.HTTPError, urllib.error.URLError) as exc:
|
|
print(f"file_drift_issue: failed ({type(exc).__name__}: {exc})", file=sys.stderr)
|
|
sys.exit(1)
|