Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 67 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'fix(workflow): retire status:pr-open on every terminal PR transition (Closes #780)' (#782) from fix/issue-780-terminal-pr-open-label-cleanup into master

Browse Source
This commit was merged in pull request #782.
This commit is contained in:
sysadmin
2026-07-21 15:14:21 -05:00
parent df3167488c ed0e8c82de
commit 8e149e6cfa
9 changed files with 1448 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/label-taxonomy.md
+42
View File
@@ -131,6 +131,44 @@ Suggested lifecycle:
The helper module `issue_workflow_labels.py` is the source of truth for the
canonical label specs and status transition replacement behavior.
## Terminal PR transitions retire `status:pr-open` (#780)
`status:pr-open` states that a linked PR is *currently open*. The moment that
stops being true the label must go, whatever ended the PR:
| Terminal reason | Raised by |
|---|---|
| `merged` | `gitea_merge_pr` |
| `closed_without_merge` | `gitea_edit_pr` closing the PR |
| `superseded` | `gitea_reconcile_superseded_by_merged_pr` |
| `already_landed` | `gitea_reconcile_already_landed_pr` |
| `controller_closure` | `gitea_close_issue` |
| `abandoned` | abandonment handling |
| `retry_recovery` | `gitea_cleanup_terminal_pr_labels` after a partial failure |
All of these route through one rule in `terminal_pr_label_cleanup.py`, so the
paths cannot drift apart. The rule guarantees:
- only `status:pr-open` is removed — every other label is preserved verbatim;
- an empty resulting label set is valid (it was the issue's only label);
- an issue that no longer carries the label is a no-op, so retries are safe;
- the result is confirmed by a read-after-write re-read, not assumed.
Controller closure runs the cleanup **before** changing issue state and fails
closed if it cannot be completed and verified — closing first would bake in the
stale label with no later step to catch it. Post-merge cleanup never blocks the
merge: the transition already happened, so failures are reported with a
`safe_next_action` instead.
Use `gitea_assess_terminal_label_hygiene` as terminal validation before
declaring a transition or cleanup batch complete. It enumerates issues plus the
live open PRs and reports any issue still carrying `status:pr-open` without an
open PR to justify it. Issues with a genuinely open PR are exempt, not
residual.
Recovery from a partial failure is `gitea_cleanup_terminal_pr_labels` with
`terminal_reason='retry_recovery'`.
## Discussion Issues
Discussion issues must be labeled `type:discussion`.
@@ -157,6 +195,10 @@ If a discussion produces implementation work, either:
be applied to the locked issue, then applies it after the PR is created.
- `gitea_set_issue_labels` accepts an explicit `worktree_path` so author
sessions can satisfy the branches-only mutation guard while changing labels.
- `gitea_cleanup_terminal_pr_labels` retires `status:pr-open` after a terminal
PR transition; it is idempotent, so it is also the retry/recovery path.
- `gitea_assess_terminal_label_hygiene` is the read-only terminal validation
for residual `status:pr-open`.
## Existing Non-Workflow Labels
docs/llm-workflow-runbooks.md
+3 -1
View File
@@ -706,7 +706,9 @@ do **not** improvise shell wrappers or fall back to direct API / temp scripts.
`fix/...` / `docs/...`); `cd` into that worktree; implement narrowly; add or
update tests if behavior changes; run the full suite; commit with an
issue-linked message; open a PR to `master`; move the issue to
`status:pr-open`. **Do not** review or merge your own PR. Include an
`status:pr-open` (every terminal transition later retires that label
automatically — see [`label-taxonomy.md`](label-taxonomy.md)). **Do not**
review or merge your own PR. Include an
`LLM Handoff Metadata` block (with `LLM-Agent-SHA`) in the PR body — see
[`llm-agent-sha.md`](llm-agent-sha.md).
- **Prompt:** `Use an author profile to implement issue #N and open a PR to
gitea_mcp_server.py
+499 -6
View File
@@ -2122,6 +2122,7 @@ def _seed_session_context(
)
import issue_work_duplicate_gate # noqa: E402
import issue_workflow_labels # noqa: E402
import terminal_pr_label_cleanup # noqa: E402 # #780 status:pr-open terminal rule
import reviewer_pr_lease # noqa: E402
import post_merge_moot_lease_gate # noqa: E402 # #745 reconciler cleanup gate
import merger_lease_adoption # noqa: E402
@@ -2698,6 +2699,12 @@ def _put_issue_label_names(
auth,
{"labels": ids},
)
# Clearing every label is a legitimate full-set replacement (#780 AC4): the
# last label may be the one being retired. Gitea answers that PUT with an
# empty body, which api_request surfaces as None — treat it as the empty
# label set rather than a protocol violation.
if res is None and not names:
res = []
if not isinstance(res, list):
raise RuntimeError(
"Post-mutation label verification failed: expected a list of labels "
@@ -2824,7 +2831,191 @@ def release_in_progress_label(issue_numbers: list[int], remote: str, host: str |
return results
def cleanup_in_progress_for_pr(pr_payload: dict, remote: str, host: str | None, org: str | None, repo: str | None) -> dict:
def _clear_pr_open_label_for_issue(
*,
base: str,
auth: str,
issue_number: int,
terminal_reason: str,
resolve_label_ids,
audit_kwargs: dict,
) -> dict:
"""Apply the authoritative #780 cleanup rule to exactly one issue.
Plans through :mod:`terminal_pr_label_cleanup`, mutates via the verified
full-set replacement, then re-reads the issue so the caller reports a
read-after-write proof rather than an assumption.
"""
try:
current = _issue_label_names(base, auth, issue_number)
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"issue_number": issue_number,
"status": "error",
"performed": False,
"verified": False,
"reasons": [f"could not read current labels: {_redact(str(exc))}"],
}
plan = terminal_pr_label_cleanup.plan_pr_open_cleanup(
current, terminal_reason=terminal_reason
)
if not plan["cleanup_required"]:
# Idempotent by construction: nothing to remove, nothing to verify
# beyond the read we already did (#780 AC5).
return {
"issue_number": issue_number,
"status": "not present",
"performed": False,
"verified": True,
"idempotent_noop": True,
"labels_after": plan["labels_after"],
"reasons": [],
}
try:
# Resolved only once a mutation is actually needed, so the no-op path
# stays free of extra API traffic.
label_ids_by_name = resolve_label_ids()
with _audited(
"set_issue_labels",
issue_number=issue_number,
request_metadata={
"source": f"terminal_pr_label_cleanup:{plan['terminal_reason']}",
"removed": plan["removed"],
"preserved": plan["preserved"],
},
**audit_kwargs,
):
_put_issue_label_names(
base=base,
auth=auth,
issue_number=issue_number,
names=plan["labels_after"],
label_ids_by_name=label_ids_by_name,
)
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"issue_number": issue_number,
"status": "error",
"performed": False,
"verified": False,
"labels_before": plan["labels_before"],
"reasons": [f"label replacement failed: {_redact(str(exc))}"],
"safe_next_action": (
"Re-run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason='{terminal_pr_label_cleanup.RETRY_RECOVERY}' "
f"for issue #{issue_number}."
),
}
try:
observed = _issue_label_names(base, auth, issue_number)
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"issue_number": issue_number,
"status": "unverified",
"performed": True,
"verified": False,
"labels_before": plan["labels_before"],
"reasons": [f"read-after-write check failed: {_redact(str(exc))}"],
"safe_next_action": (
"Re-run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason='{terminal_pr_label_cleanup.RETRY_RECOVERY}' "
f"for issue #{issue_number} to confirm the final label set."
),
}
verification = terminal_pr_label_cleanup.verify_pr_open_cleanup(
observed, plan=plan
)
return {
"issue_number": issue_number,
"status": "removed" if verification["verified"] else "verification_failed",
"performed": True,
"verified": verification["verified"],
"labels_before": plan["labels_before"],
"labels_after": verification["observed_labels"],
"empty_label_set": verification["empty_label_set"],
"verification": verification,
"reasons": verification["reasons"],
"safe_next_action": verification["safe_next_action"],
}
def clear_pr_open_label(
issue_numbers: list[int],
remote: str,
host: str | None,
org: str | None,
repo: str | None,
*,
terminal_reason: str,
) -> dict:
"""Authoritative ``status:pr-open`` cleanup for terminal PR states (#780).
Every sanctioned terminal path routes here — merge, close-without-merge,
supersession/abandonment, already-landed reconciliation, controller
closure, and retry/recovery — so the rule cannot drift between them. Only
``status:pr-open`` is ever removed; all other labels are preserved, and an
empty resulting set is a valid outcome.
"""
reason = terminal_pr_label_cleanup.canonical_terminal_reason(terminal_reason)
numbers: list[int] = []
for raw in issue_numbers or []:
try:
num = int(raw)
except (TypeError, ValueError):
continue
if num not in numbers:
numbers.append(num)
if not numbers:
return terminal_pr_label_cleanup.summarize_cleanup_results(
[], terminal_reason=reason
)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
audit_kwargs = {"host": h, "remote": remote, "org": o, "repo": r}
# Memoized, lazily resolved label inventory: an issue that no longer
# carries the label needs no ids at all, so a clean repository costs one
# read per issue and nothing else.
cached: dict[str, dict[str, int]] = {}
def resolve_label_ids() -> dict[str, int]:
if "map" not in cached:
cached["map"] = _repo_label_id_map(base, auth)
return cached["map"]
results = [
_clear_pr_open_label_for_issue(
base=base,
auth=auth,
issue_number=num,
terminal_reason=reason,
resolve_label_ids=resolve_label_ids,
audit_kwargs=audit_kwargs,
)
for num in numbers
]
return terminal_pr_label_cleanup.summarize_cleanup_results(
results, terminal_reason=reason
)
def cleanup_in_progress_for_pr(
pr_payload: dict,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
*,
terminal_reason: str = terminal_pr_label_cleanup.MERGED,
) -> dict:
body = pr_payload.get("body") or ""
title = pr_payload.get("title") or ""
branch = pr_payload.get("head", {}).get("ref") or ""
@@ -2833,10 +3024,38 @@ def cleanup_in_progress_for_pr(pr_payload: dict, remote: str, host: str | None,
issues = extract_linked_issue_numbers(text, branch)
if not issues:
return {"cleanup_status": "no linked issue found"}
return {
"cleanup_status": "no linked issue found",
"pr_open_label_cleanup": terminal_pr_label_cleanup.summarize_cleanup_results(
[], terminal_reason=terminal_reason
),
}
results = release_in_progress_label(issues, remote, host, org, repo)
return {"cleanup_status": results}
# #780: the PR has reached a terminal state, so status:pr-open must go too.
# Never raises — a failed cleanup is reported, never silently dropped, and
# never undoes the terminal transition that already happened.
try:
pr_open_cleanup = clear_pr_open_label(
issues, remote, host, org, repo, terminal_reason=terminal_reason
)
except Exception as exc: # noqa: BLE001 — redact before surfacing
pr_open_cleanup = {
"label": terminal_pr_label_cleanup.PR_OPEN_LABEL,
"clean": False,
"checked": issues,
"removed": [],
"already_absent": [],
"failed": issues,
"results": [],
"reasons": [f"terminal label cleanup failed: {_redact(str(exc))}"],
"safe_next_action": (
"Re-run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason='{terminal_pr_label_cleanup.RETRY_RECOVERY}' "
"for the linked issues."
),
}
return {"cleanup_status": results, "pr_open_label_cleanup": pr_open_cleanup}
# ── Helpers ───────────────────────────────────────────────────────────────────
@@ -8357,9 +8576,18 @@ def gitea_edit_pr(
data = api_request("PATCH", url, auth, payload)
cleanup_status = None
pr_open_label_cleanup = None
if state == "closed":
cleanup = cleanup_in_progress_for_pr(data, remote, host, org, repo)
cleanup = cleanup_in_progress_for_pr(
data,
remote,
host,
org,
repo,
terminal_reason=terminal_pr_label_cleanup.CLOSED_WITHOUT_MERGE,
)
cleanup_status = cleanup.get("cleanup_status")
pr_open_label_cleanup = cleanup.get("pr_open_label_cleanup")
if isinstance(cleanup_status, dict):
for issue_num, st in cleanup_status.items():
if st == "released":
@@ -8376,6 +8604,7 @@ def gitea_edit_pr(
"body": data.get("body", ""),
"state": data["state"],
"cleanup_status": cleanup_status,
"pr_open_label_cleanup": pr_open_label_cleanup,
}
return _with_optional_url(result, data.get("html_url"))
@@ -9111,12 +9340,33 @@ def gitea_merge_pr(
result["cleanup_status"] = "skipped (merge read-back failed)"
else:
try:
cleanup = cleanup_in_progress_for_pr(merged, remote, host, org, repo)
cleanup = cleanup_in_progress_for_pr(
merged,
remote,
host,
org,
repo,
terminal_reason=terminal_pr_label_cleanup.MERGED,
)
result["cleanup_status"] = cleanup.get("cleanup_status")
result["pr_open_label_cleanup"] = cleanup.get("pr_open_label_cleanup")
except Exception as cleanup_exc: # noqa: BLE001 — redact before surfacing
result["cleanup_status"] = (
f"skipped (cleanup error: {_redact(str(cleanup_exc))})"
)
result["pr_open_label_cleanup"] = {
"label": terminal_pr_label_cleanup.PR_OPEN_LABEL,
"clean": False,
"reasons": [
f"cleanup error: {_redact(str(cleanup_exc))}"
],
"safe_next_action": (
"Re-run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason="
f"'{terminal_pr_label_cleanup.RETRY_RECOVERY}' for the "
"linked issues."
),
}
except Exception as exc: # noqa: BLE001 — redact before surfacing
reasons.append(f"merge failed: {_redact(str(exc))}")
return result
@@ -11104,6 +11354,20 @@ def gitea_reconcile_already_landed_pr(
result["issue_closed"] = True
result["performed"] = True
# #780: once the PR is closed as already-landed, the linked issue is no
# longer awaiting an open PR — retire status:pr-open even when the issue
# itself stays open. Reported, never fatal: the reconciliation already
# happened and must not be reversed by a label failure.
if linked_issue and (result.get("pr_closed") or result.get("issue_closed")):
result["pr_open_label_cleanup"] = clear_pr_open_label(
[linked_issue],
remote,
host,
org,
repo,
terminal_reason=terminal_pr_label_cleanup.ALREADY_LANDED,
)
return result
@@ -11359,6 +11623,20 @@ def gitea_reconcile_superseded_by_merged_pr(
result["issue_closed"] = True
result["performed"] = True
# #780: supersession/abandonment is terminal for the target PR, so the
# target issue must not keep advertising an open PR.
if target_issue_number and (
result.get("pr_closed") or result.get("issue_closed")
):
result["pr_open_label_cleanup"] = clear_pr_open_label(
[target_issue_number],
remote,
host,
org,
repo,
terminal_reason=terminal_pr_label_cleanup.SUPERSEDED,
)
return result
@mcp.tool()
@@ -11417,6 +11695,33 @@ def gitea_close_issue(
"reasons": closure_gate["reasons"],
"safe_next_action": closure_gate["safe_next_action"],
}
# #780: controller closure is a terminal transition, so status:pr-open must
# be retired. Run it *before* the state change and fail closed if it cannot
# be completed — closing first would leave the exact stale-label state this
# gate exists to prevent, with no remaining step to catch it.
pr_open_cleanup = clear_pr_open_label(
[issue_number],
remote,
host,
org,
repo,
terminal_reason=terminal_pr_label_cleanup.CONTROLLER_CLOSURE,
)
if not pr_open_cleanup["clean"]:
return {
"success": False,
"blocked": True,
"performed": False,
"message": (
f"Issue #{issue_number} close blocked (#780): the terminal "
f"'{terminal_pr_label_cleanup.PR_OPEN_LABEL}' cleanup could not "
"be completed and verified."
),
"pr_open_label_cleanup": pr_open_cleanup,
"reasons": pr_open_cleanup["reasons"],
"safe_next_action": pr_open_cleanup["safe_next_action"],
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -11426,13 +11731,201 @@ def gitea_close_issue(
cleanup_result = release_in_progress_label([issue_number], remote, host, org, repo)
# Terminal validation (#780): re-read the closed issue and report any
# residual status:pr-open rather than assuming the earlier cleanup held.
try:
closed_labels = _issue_label_names(
repo_api_url(h, o, r), auth, issue_number
)
terminal_validation = terminal_pr_label_cleanup.detect_residual_pr_open(
[{"number": issue_number, "state": "closed", "labels": closed_labels}]
)
except Exception as exc: # noqa: BLE001 — redact before surfacing
terminal_validation = {
"label": terminal_pr_label_cleanup.PR_OPEN_LABEL,
"clean": False,
"checked_count": 0,
"residual_count": None,
"residual_issues": [],
"reasons": [
f"terminal label validation read-back failed: {_redact(str(exc))}"
],
"safe_next_action": (
"Re-check the issue labels with gitea_assess_terminal_label_hygiene."
),
}
return {
"success": True,
"message": f"Issue #{issue_number} closed.",
"cleanup_status": cleanup_result
"cleanup_status": cleanup_result,
"pr_open_label_cleanup": pr_open_cleanup,
"terminal_label_validation": terminal_validation,
}
@mcp.tool()
def gitea_cleanup_terminal_pr_labels(
issue_numbers: list[int],
terminal_reason: str = terminal_pr_label_cleanup.RETRY_RECOVERY,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Retire ``status:pr-open`` after a terminal PR transition (#780).
This is the sanctioned recovery path when a terminal transition completed
but its label cleanup did not — for example a merge whose read-back failed,
or a workflow interrupted between closing a PR and updating its issue.
The rule is the same one every terminal path uses: remove only
``status:pr-open``, preserve every other label, allow an empty resulting
set, and verify by read-after-write. Calling it on an issue that no longer
carries the label is a harmless no-op, so retries are always safe.
Args:
issue_numbers: Issues whose terminal label state should be repaired.
terminal_reason: Why the PR is terminal — one of merged,
closed_without_merge, superseded, already_landed,
controller_closure, abandoned, retry_recovery.
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Optional branches worktree to verify before mutation.
Returns:
dict with 'clean', per-issue 'results' carrying read-after-write proof,
'removed', 'already_absent', 'failed', and 'safe_next_action'.
"""
blocked = _profile_permission_block(
task_capability_map.required_permission("cleanup_terminal_pr_labels"),
remote=remote,
host=host,
org=org,
repo=repo,
org_explicit=org is not None,
repo_explicit=repo is not None,
)
if blocked:
return blocked
verify_preflight_purity(
remote,
worktree_path=worktree_path,
task="cleanup_terminal_pr_labels",
org=org,
repo=repo,
)
try:
reason = terminal_pr_label_cleanup.canonical_terminal_reason(terminal_reason)
except ValueError as exc:
return {
"success": False,
"clean": False,
"performed": False,
"reasons": [str(exc)],
"safe_next_action": (
"Re-call with terminal_reason set to one of: "
+ ", ".join(terminal_pr_label_cleanup.TERMINAL_REASONS)
),
}
summary = clear_pr_open_label(
issue_numbers, remote, host, org, repo, terminal_reason=reason
)
summary["success"] = summary["clean"]
summary["performed"] = bool(summary["removed"])
return summary
@mcp.tool()
def gitea_assess_terminal_label_hygiene(
state: str = "all",
limit: int = 500,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Read-only terminal validation for residual ``status:pr-open`` (#780).
Enumerates issues and the live open pull requests, then reports any issue
still carrying ``status:pr-open`` without an open PR to justify it. Use it
before declaring a terminal transition — or a cleanup batch — complete.
Args:
state: Issue state filter — 'open', 'closed', or 'all' (default).
limit: Max issues to enumerate across all pages.
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict with 'clean', 'residual_count', 'residual_issues',
'open_pr_count', and 'safe_next_action'.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"clean": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
try:
issues = api_get_all(
f"{base}/issues?state={state}&type=issues", auth, limit=limit
) or []
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"success": False,
"clean": False,
"reasons": [f"could not enumerate issues: {_redact(str(exc))}"],
"safe_next_action": (
"Retry once the Gitea API is reachable; an incomplete "
"enumeration cannot prove terminal label hygiene."
),
}
try:
open_pulls = _list_open_pulls(h, o, r, auth)
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"success": False,
"clean": False,
"reasons": [f"could not enumerate open pull requests: {_redact(str(exc))}"],
"safe_next_action": (
"Retry once the Gitea API is reachable; without the open-PR "
"inventory a legitimate status:pr-open cannot be distinguished "
"from a stale one."
),
}
open_pr_issues: list[int] = []
for pull in open_pulls:
head_obj = pull.get("head") or {}
branch = head_obj.get("ref") if isinstance(head_obj, dict) else None
text = f"{pull.get('title') or ''}\n{pull.get('body') or ''}"
open_pr_issues.extend(extract_linked_issue_numbers(text, branch))
assessment = terminal_pr_label_cleanup.detect_residual_pr_open(
issues, open_pr_issue_numbers=open_pr_issues
)
assessment["success"] = True
assessment["issue_state_filter"] = state
assessment["open_pr_count"] = len(open_pulls)
return assessment
@mcp.tool()
def gitea_list_issues(
state: str = "open",
skills/llm-project-workflow/schemas/reconcile-landed-final-report.md
+8
View File
@@ -39,6 +39,7 @@ occurred).
- Git ref mutations:
- MCP/Gitea mutations:
- Reconciliation mutations:
- Terminal label cleanup:
- External-state mutations:
- Read-only diagnostics:
- Blockers:
@@ -52,6 +53,13 @@ Identity format: `username / profile` (not personal email unless required — #3
`git fetch` belongs under `Git ref mutations`, not read-only diagnostics (#297).
`Terminal label cleanup` (#780) reports the `pr_open_label_cleanup` record the
reconciliation tool returned — `clean` / `failed` / `not applicable (no linked
issue)`, with the labels removed and preserved. Reconciliation is a terminal
transition, so a non-`clean` record blocks any "reconciled" claim; recover with
`gitea_cleanup_terminal_pr_labels` (`terminal_reason='retry_recovery'`) and
confirm with `gitea_assess_terminal_label_hygiene`.
The report must also carry the canonical self-propagating handoff block
(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to
the Gitea issue or PR thread.
skills/llm-project-workflow/schemas/review-merge-final-report.md
+15
View File
@@ -99,6 +99,21 @@ Narrative final report and controller handoff must agree on eligibility class,
candidate/reviewed head SHA, mutation state, worktree usage, review decision,
terminal review mutation, merge result, and linked issue status.
### Terminal label state (#780)
A run that takes a PR to a terminal state — merged, closed without merge,
superseded, or reconciled as already landed — must report what happened to the
linked issue's `status:pr-open` label, quoting the `pr_open_label_cleanup`
record the terminal tool returned:
- Terminal label cleanup: `clean` / `failed` / `not applicable (no linked issue)`
- Labels removed and preserved per issue, with the read-after-write read-back
Never claim the transition is complete while that record is not `clean`. A
failed cleanup does not undo the merge; the safe next action is
`gitea_cleanup_terminal_pr_labels` with `terminal_reason='retry_recovery'`,
confirmed by `gitea_assess_terminal_label_hygiene`.
### Proof-backed claims (#395)
Proof-sensitive claims must cite explicit command/tool evidence in the report
task_capability_map.py
+7
View File
@@ -36,6 +36,12 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.issue.comment",
"role": "author",
},
# #780: retire status:pr-open after a terminal PR transition. Same label
# authority as set_issue_labels — it is a strictly narrower operation.
"cleanup_terminal_pr_labels": {
"permission": "gitea.issue.comment",
"role": "author",
},
"create_label": {
"permission": "gitea.issue.comment",
"role": "author",
@@ -507,6 +513,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
"gitea_create_issue_comment": "comment_issue",
"gitea_mark_issue": "mark_issue",
"gitea_set_issue_labels": "set_issue_labels",
"gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels",
"gitea_create_label": "create_label",
"gitea_commit_files": "commit_files",
}
terminal_pr_label_cleanup.py
+308
View File
@@ -0,0 +1,308 @@
"""Authoritative terminal-transition cleanup for ``status:pr-open`` (#780).
``status:pr-open`` is applied by ``gitea_create_pr`` while a linked pull
request is open. Nothing removed it again: the workflow's terminal paths
(merge, close-without-merge, supersession, already-landed reconciliation,
controller closure) each ended without touching the label, so a repository
audit found 40 closed issues still carrying it.
This module is the single source of truth for that cleanup. Every sanctioned
terminal path plans its label mutation here rather than implementing its own
rule, so the paths cannot drift apart:
- :func:`plan_pr_open_cleanup` decides the exact resulting label set. It only
ever removes ``status:pr-open``; every other label is preserved verbatim,
including the case where the result is an empty label set.
- :func:`verify_pr_open_cleanup` is the read-after-write check. It proves the
label is gone *and* that no unrelated label was dropped or added.
- :func:`detect_residual_pr_open` is the terminal validation: given issues, it
reports any that still carry the label, so a controller closure or audit
fails loudly instead of leaving the leak behind.
The rule is idempotent by construction: an issue without the label plans no
mutation, so retries and recovery re-runs are harmless.
This module performs no I/O — callers own the Gitea API calls.
"""
from __future__ import annotations
from typing import Any, Iterable, Mapping, Sequence
import issue_workflow_labels
#: The single label this module is responsible for retiring.
PR_OPEN_LABEL = "status:pr-open"
# Canonical terminal reasons — the sanctioned ways an issue can end up
# associated with a pull request that is no longer open.
MERGED = "merged"
CLOSED_WITHOUT_MERGE = "closed_without_merge"
SUPERSEDED = "superseded"
ALREADY_LANDED = "already_landed"
CONTROLLER_CLOSURE = "controller_closure"
ABANDONED = "abandoned"
RETRY_RECOVERY = "retry_recovery"
TERMINAL_REASONS: tuple[str, ...] = (
MERGED,
CLOSED_WITHOUT_MERGE,
SUPERSEDED,
ALREADY_LANDED,
CONTROLLER_CLOSURE,
ABANDONED,
RETRY_RECOVERY,
)
_REASON_ALIASES: dict[str, str] = {
"merge": MERGED,
"merged": MERGED,
"pr_merged": MERGED,
"closed": CLOSED_WITHOUT_MERGE,
"close": CLOSED_WITHOUT_MERGE,
"closed_without_merge": CLOSED_WITHOUT_MERGE,
"pr_closed": CLOSED_WITHOUT_MERGE,
"supersede": SUPERSEDED,
"superseded": SUPERSEDED,
"supersession": SUPERSEDED,
"already_landed": ALREADY_LANDED,
"reconcile_already_landed": ALREADY_LANDED,
"controller_closure": CONTROLLER_CLOSURE,
"close_issue": CONTROLLER_CLOSURE,
"abandon": ABANDONED,
"abandoned": ABANDONED,
"retry": RETRY_RECOVERY,
"recovery": RETRY_RECOVERY,
"retry_recovery": RETRY_RECOVERY,
}
#: Human-readable phrasing used in audit comments and diagnostics.
REASON_DESCRIPTIONS: dict[str, str] = {
MERGED: "the linked PR was merged",
CLOSED_WITHOUT_MERGE: "the linked PR was closed without merging",
SUPERSEDED: "the linked PR was superseded by another merged PR",
ALREADY_LANDED: "the linked PR's change was already on the target branch",
CONTROLLER_CLOSURE: "the issue reached controller closure",
ABANDONED: "the linked PR was abandoned",
RETRY_RECOVERY: "a partial terminal transition is being recovered",
}
def canonical_terminal_reason(reason: str | None) -> str:
"""Normalize a terminal reason, failing closed on anything unrecognized."""
name = (reason or "").strip()
if name in TERMINAL_REASONS:
return name
normalized = name.lower().replace("-", "_").replace(" ", "_")
try:
return _REASON_ALIASES[normalized]
except KeyError as exc:
raise ValueError(
f"unknown terminal PR reason '{reason}' (expected one of: "
+ ", ".join(TERMINAL_REASONS)
+ ")"
) from exc
def plan_pr_open_cleanup(
current_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
*,
terminal_reason: str,
) -> dict[str, Any]:
"""Plan the label set an issue must carry after a terminal PR transition.
The plan removes ``status:pr-open`` and nothing else. When the label is
absent the plan is an explicit no-op (``cleanup_required`` False), which is
what makes repeated cleanup calls harmless. When it was the only label the
resulting set is legitimately empty.
"""
reason = canonical_terminal_reason(terminal_reason)
before = issue_workflow_labels.label_names(current_labels)
after = [name for name in before if name != PR_OPEN_LABEL]
present = len(after) != len(before)
return {
"terminal_reason": reason,
"terminal_reason_description": REASON_DESCRIPTIONS[reason],
"label": PR_OPEN_LABEL,
"label_present": present,
"cleanup_required": present,
"idempotent_noop": not present,
"labels_before": before,
"labels_after": after,
"removed": [PR_OPEN_LABEL] if present else [],
"preserved": list(after),
"empty_label_set": not after,
}
def verify_pr_open_cleanup(
observed_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
*,
plan: Mapping[str, Any],
) -> dict[str, Any]:
"""Read-after-write check for a planned cleanup.
Verifies the label is gone and that the observed set matches the plan
exactly, so an unrelated label silently dropped (or re-added) by the API is
reported rather than accepted.
"""
observed = issue_workflow_labels.label_names(observed_labels)
expected = list(plan.get("labels_after") or [])
observed_set = set(observed)
expected_set = set(expected)
residual = PR_OPEN_LABEL in observed_set
unexpected_removals = sorted(expected_set - observed_set)
unexpected_additions = sorted(observed_set - expected_set - {PR_OPEN_LABEL})
reasons: list[str] = []
if residual:
reasons.append(
f"'{PR_OPEN_LABEL}' is still present after terminal cleanup"
)
if unexpected_removals:
reasons.append(
"unrelated labels were dropped by the cleanup: "
+ ", ".join(unexpected_removals)
)
if unexpected_additions:
reasons.append(
"unexpected labels appeared during the cleanup: "
+ ", ".join(unexpected_additions)
)
verified = not reasons
return {
"verified": verified,
"residual": residual,
"observed_labels": observed,
"expected_labels": expected,
"unexpected_removals": unexpected_removals,
"unexpected_additions": unexpected_additions,
"empty_label_set": not observed,
"reasons": reasons,
"safe_next_action": (
""
if verified
else (
"Re-run the terminal cleanup for this issue with "
f"terminal_reason='{RETRY_RECOVERY}' and confirm the read-back "
f"no longer reports '{PR_OPEN_LABEL}'."
)
),
}
def summarize_cleanup_results(
results: Sequence[Mapping[str, Any]],
*,
terminal_reason: str,
) -> dict[str, Any]:
"""Aggregate per-issue cleanup outcomes into one reportable record."""
reason = canonical_terminal_reason(terminal_reason)
entries = [dict(entry) for entry in results]
removed = [e.get("issue_number") for e in entries if e.get("status") == "removed"]
absent = [
e.get("issue_number") for e in entries if e.get("status") == "not present"
]
failed = [
e.get("issue_number")
for e in entries
if e.get("status") not in ("removed", "not present") or not e.get("verified")
]
reasons: list[str] = []
for entry in entries:
for text in entry.get("reasons") or []:
reasons.append(f"issue #{entry.get('issue_number')}: {text}")
clean = not failed
return {
"label": PR_OPEN_LABEL,
"terminal_reason": reason,
"clean": clean,
"checked": [e.get("issue_number") for e in entries],
"removed": removed,
"already_absent": absent,
"failed": failed,
"results": entries,
"reasons": reasons,
"safe_next_action": (
""
if clean
else (
"Terminal label cleanup did not complete for "
+ ", ".join(f"#{num}" for num in failed)
+ ". Re-run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason='{RETRY_RECOVERY}' for those issues."
)
),
}
def detect_residual_pr_open(
issues: Iterable[Mapping[str, Any]],
*,
open_pr_issue_numbers: Iterable[int] = (),
) -> dict[str, Any]:
"""Terminal validation: report issues still carrying ``status:pr-open``.
An issue with a genuinely open pull request is allowed to keep the label,
so *open_pr_issue_numbers* is excluded from the residual set rather than
being reported as a leak.
"""
legitimate: set[int] = set()
for num in open_pr_issue_numbers or ():
try:
legitimate.add(int(num))
except (TypeError, ValueError):
continue
checked = 0
residual: list[dict[str, Any]] = []
exempt: list[int] = []
for issue in issues or []:
checked += 1
names = issue_workflow_labels.label_names(issue)
if PR_OPEN_LABEL not in names:
continue
try:
number = int(issue.get("number"))
except (TypeError, ValueError):
number = None
if number is not None and number in legitimate:
exempt.append(number)
continue
residual.append(
{
"number": number,
"state": issue.get("state"),
"labels": names,
}
)
clean = not residual
reasons = [
(
f"issue #{entry['number']} ({entry.get('state') or 'unknown state'}) "
f"still carries '{PR_OPEN_LABEL}' with no open PR"
)
for entry in residual
]
return {
"label": PR_OPEN_LABEL,
"clean": clean,
"checked_count": checked,
"residual_count": len(residual),
"residual_issues": residual,
"exempt_open_pr_issues": sorted(exempt),
"reasons": reasons,
"safe_next_action": (
""
if clean
else (
"Run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason='{RETRY_RECOVERY}' for issues "
+ ", ".join(f"#{entry['number']}" for entry in residual)
+ " before declaring the terminal transition complete."
)
),
}
tests/test_audit.py
+11 -1
View File
@@ -238,7 +238,17 @@ class TestSimpleToolAudit(_AuditWiringBase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_close_issue_audited(self, _auth, mock_api):
mock_api.side_effect = [{"state": "closed"}, {"login": "mgr-bot"}]
# Keyed rather than positional: closing an issue also reads its labels
# before and after the state change for the #780 terminal cleanup and
# its read-after-write check, so call order is not a fixed sequence.
def api(method, url, auth, payload=None):
if method == "PATCH":
return {"state": "closed"}
if "/issues/" in url:
return {"number": 42, "labels": []}
return {"login": "mgr-bot"}
mock_api.side_effect = api
with patch.dict(os.environ, self._env(), clear=True):
gitea_close_issue(issue_number=42, remote="prgs")
recs = self._records()
tests/test_terminal_pr_label_cleanup.py
+555
View File
@@ -0,0 +1,555 @@
"""#780: ``status:pr-open`` must not survive a terminal PR transition.
The leak this file locks down: ``gitea_create_pr`` applied ``status:pr-open``
and no terminal path ever removed it, so a repository audit found 40 closed
issues still advertising an open PR that had long since merged or closed.
Coverage mirrors the issue's acceptance criteria: merge, close-without-merge,
supersession, already-landed reconciliation, controller closure, retry /
idempotency, unrelated-label preservation, the only-label (empty set) case,
and terminal validation of any residual label.
"""
from __future__ import annotations
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
import terminal_pr_label_cleanup as tplc
FAKE_AUTH = "token test-token"
PR_OPEN = tplc.PR_OPEN_LABEL
def _lb(name: str, lid: int) -> dict:
return {"id": lid, "name": name, "color": "000000"}
# ---------------------------------------------------------------------------
# Pure rule: planning
# ---------------------------------------------------------------------------
class TestPlanPrOpenCleanup(unittest.TestCase):
def test_removes_only_the_pr_open_label(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", PR_OPEN, "workflow-hardening"],
terminal_reason=tplc.MERGED,
)
self.assertTrue(plan["cleanup_required"])
self.assertEqual(plan["removed"], [PR_OPEN])
self.assertEqual(plan["labels_after"], ["type:bug", "workflow-hardening"])
def test_preserves_unrelated_labels_in_original_order(self):
labels = ["workflow-hardening", "type:bug", PR_OPEN, "role:author", "leases"]
plan = tplc.plan_pr_open_cleanup(labels, terminal_reason=tplc.MERGED)
self.assertEqual(
plan["labels_after"],
["workflow-hardening", "type:bug", "role:author", "leases"],
)
self.assertNotIn(PR_OPEN, plan["labels_after"])
def test_only_label_yields_empty_set(self):
plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=tplc.MERGED)
self.assertTrue(plan["cleanup_required"])
self.assertEqual(plan["labels_after"], [])
self.assertTrue(plan["empty_label_set"])
def test_absent_label_is_an_idempotent_noop(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", "status:done"], terminal_reason=tplc.RETRY_RECOVERY
)
self.assertFalse(plan["cleanup_required"])
self.assertTrue(plan["idempotent_noop"])
self.assertEqual(plan["labels_after"], ["type:bug", "status:done"])
def test_accepts_gitea_label_objects(self):
plan = tplc.plan_pr_open_cleanup(
{"labels": [{"name": PR_OPEN}, {"name": "type:bug"}]},
terminal_reason=tplc.SUPERSEDED,
)
self.assertEqual(plan["labels_after"], ["type:bug"])
def test_every_terminal_reason_is_planable(self):
for reason in tplc.TERMINAL_REASONS:
plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=reason)
self.assertEqual(plan["terminal_reason"], reason)
self.assertTrue(plan["terminal_reason_description"])
def test_unknown_terminal_reason_fails_closed(self):
with self.assertRaises(ValueError):
tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason="whenever")
def test_reason_aliases_normalize(self):
self.assertEqual(tplc.canonical_terminal_reason("merge"), tplc.MERGED)
self.assertEqual(
tplc.canonical_terminal_reason("already-landed"), tplc.ALREADY_LANDED
)
self.assertEqual(
tplc.canonical_terminal_reason("controller-closure"),
tplc.CONTROLLER_CLOSURE,
)
# ---------------------------------------------------------------------------
# Pure rule: read-after-write verification
# ---------------------------------------------------------------------------
class TestVerifyPrOpenCleanup(unittest.TestCase):
def test_verified_when_observed_matches_plan(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
)
result = tplc.verify_pr_open_cleanup(["type:bug"], plan=plan)
self.assertTrue(result["verified"])
self.assertFalse(result["residual"])
self.assertEqual(result["reasons"], [])
def test_residual_label_is_reported(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
)
result = tplc.verify_pr_open_cleanup(["type:bug", PR_OPEN], plan=plan)
self.assertFalse(result["verified"])
self.assertTrue(result["residual"])
self.assertIn(PR_OPEN, result["reasons"][0])
self.assertTrue(result["safe_next_action"])
def test_dropped_unrelated_label_is_reported(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", "leases", PR_OPEN], terminal_reason=tplc.MERGED
)
result = tplc.verify_pr_open_cleanup(["type:bug"], plan=plan)
self.assertFalse(result["verified"])
self.assertEqual(result["unexpected_removals"], ["leases"])
def test_unexpected_added_label_is_reported(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
)
result = tplc.verify_pr_open_cleanup(["type:bug", "surprise"], plan=plan)
self.assertFalse(result["verified"])
self.assertEqual(result["unexpected_additions"], ["surprise"])
def test_empty_observed_set_verifies_for_only_label_case(self):
plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=tplc.MERGED)
result = tplc.verify_pr_open_cleanup([], plan=plan)
self.assertTrue(result["verified"])
self.assertTrue(result["empty_label_set"])
# ---------------------------------------------------------------------------
# Terminal validation
# ---------------------------------------------------------------------------
class TestDetectResidualPrOpen(unittest.TestCase):
def test_clean_repository(self):
issues = [
{"number": 1, "state": "closed", "labels": [{"name": "type:bug"}]},
{"number": 2, "state": "open", "labels": []},
]
result = tplc.detect_residual_pr_open(issues)
self.assertTrue(result["clean"])
self.assertEqual(result["residual_count"], 0)
self.assertEqual(result["checked_count"], 2)
def test_reports_each_stale_issue(self):
issues = [
{"number": 626, "state": "closed", "labels": [{"name": PR_OPEN}]},
{"number": 772, "state": "closed", "labels": [{"name": PR_OPEN}]},
{"number": 9, "state": "open", "labels": [{"name": "type:bug"}]},
]
result = tplc.detect_residual_pr_open(issues)
self.assertFalse(result["clean"])
self.assertEqual(result["residual_count"], 2)
self.assertEqual(
[entry["number"] for entry in result["residual_issues"]], [626, 772]
)
self.assertTrue(result["safe_next_action"])
def test_issue_with_a_live_open_pr_is_not_residual(self):
issues = [{"number": 42, "state": "open", "labels": [{"name": PR_OPEN}]}]
result = tplc.detect_residual_pr_open(issues, open_pr_issue_numbers=[42])
self.assertTrue(result["clean"])
self.assertEqual(result["exempt_open_pr_issues"], [42])
# ---------------------------------------------------------------------------
# Executor: one authoritative rule, with read-after-write proof
# ---------------------------------------------------------------------------
class _ExecutorHarness(unittest.TestCase):
"""Drives mcp_server.clear_pr_open_label against a fake Gitea."""
def setUp(self):
self.issue_labels: dict[int, list[str]] = {}
self.repo_labels = {
PR_OPEN: 4,
"type:bug": 1,
"workflow-hardening": 2,
"leases": 3,
"status:done": 5,
}
self.puts: list[tuple[int, list[int]]] = []
patch("mcp_server._resolve", return_value=("h", "o", "r")).start()
patch("mcp_server._auth", return_value=FAKE_AUTH).start()
patch(
"mcp_server.repo_api_url",
return_value="https://gitea.example/api/v1/repos/o/r",
).start()
patch("gitea_audit.audit_enabled", return_value=False).start()
patch("mcp_server.api_request", side_effect=self._api).start()
# api_get_all resolves api_request inside gitea_auth, so patching the
# mcp_server binding alone would let the label inventory hit the network.
patch("mcp_server.api_get_all", side_effect=self._api_get_all).start()
self.addCleanup(patch.stopall)
def _api_get_all(self, url, auth, **_kwargs):
if "/labels" in url:
return [_lb(name, lid) for name, lid in self.repo_labels.items()]
raise AssertionError(f"unexpected paginated GET: {url}")
def _api(self, method, url, auth, payload=None):
if method == "GET" and "/issues/" in url:
num = int(url.rsplit("/issues/", 1)[1].split("?")[0])
return {
"number": num,
"labels": [
{"name": n, "id": self.repo_labels[n]}
for n in self.issue_labels.get(num, [])
],
}
if method == "PUT" and url.endswith("/labels"):
num = int(url.rsplit("/issues/", 1)[1].split("/")[0])
ids = payload["labels"]
by_id = {lid: name for name, lid in self.repo_labels.items()}
names = [by_id[i] for i in ids]
self.puts.append((num, ids))
self.issue_labels[num] = names
return [_lb(n, self.repo_labels[n]) for n in names]
raise AssertionError(f"unexpected API call: {method} {url}")
def _clear(self, numbers, reason=tplc.MERGED):
return mcp_server.clear_pr_open_label(
numbers, "prgs", None, None, None, terminal_reason=reason
)
class TestClearPrOpenLabel(_ExecutorHarness):
def test_removes_label_and_preserves_the_rest(self):
self.issue_labels[780] = ["type:bug", PR_OPEN, "workflow-hardening"]
summary = self._clear([780])
self.assertTrue(summary["clean"])
self.assertEqual(summary["removed"], [780])
self.assertEqual(
self.issue_labels[780], ["type:bug", "workflow-hardening"]
)
def test_only_label_results_in_empty_set(self):
self.issue_labels[626] = [PR_OPEN]
summary = self._clear([626])
self.assertTrue(summary["clean"])
self.assertEqual(self.issue_labels[626], [])
self.assertEqual(self.puts, [(626, [])])
self.assertTrue(summary["results"][0]["empty_label_set"])
def test_read_after_write_proof_is_returned(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
summary = self._clear([780])
entry = summary["results"][0]
self.assertTrue(entry["verified"])
self.assertEqual(entry["labels_before"], ["type:bug", PR_OPEN])
self.assertEqual(entry["labels_after"], ["type:bug"])
self.assertEqual(entry["verification"]["observed_labels"], ["type:bug"])
def test_repeated_cleanup_is_harmless(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
first = self._clear([780])
second = self._clear([780], reason=tplc.RETRY_RECOVERY)
third = self._clear([780], reason=tplc.RETRY_RECOVERY)
self.assertTrue(first["clean"] and second["clean"] and third["clean"])
self.assertEqual(second["already_absent"], [780])
self.assertEqual(third["already_absent"], [780])
# Exactly one mutation across three calls.
self.assertEqual(len(self.puts), 1)
self.assertEqual(self.issue_labels[780], ["type:bug"])
def test_noop_path_never_reads_the_label_inventory(self):
self.issue_labels[780] = ["type:bug"]
with patch("mcp_server._repo_label_id_map") as mock_map:
summary = self._clear([780])
self.assertTrue(summary["clean"])
mock_map.assert_not_called()
def test_duplicate_issue_numbers_are_collapsed(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
summary = self._clear([780, 780, "780"])
self.assertEqual(summary["checked"], [780])
self.assertEqual(len(self.puts), 1)
def test_no_issue_numbers_is_a_clean_noop(self):
summary = self._clear([])
self.assertTrue(summary["clean"])
self.assertEqual(summary["checked"], [])
def test_failed_mutation_is_reported_not_swallowed(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
def boom(*_a, **_kw):
raise RuntimeError("gitea exploded")
with patch("mcp_server._put_issue_label_names", side_effect=boom):
summary = self._clear([780])
self.assertFalse(summary["clean"])
self.assertEqual(summary["failed"], [780])
self.assertTrue(summary["safe_next_action"])
self.assertIn(PR_OPEN, self.issue_labels[780])
def test_residual_label_after_write_fails_verification(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
real_api = self._api
# Simulate a write that reports success but leaves the label behind.
def sticky(method, url, auth, payload=None):
if method == "PUT" and url.endswith("/labels"):
num = int(url.rsplit("/issues/", 1)[1].split("/")[0])
self.puts.append((num, payload["labels"]))
return [_lb("type:bug", 1), _lb(PR_OPEN, 4)]
return real_api(method, url, auth, payload)
with patch("mcp_server.api_request", side_effect=sticky):
summary = self._clear([780])
self.assertFalse(summary["clean"])
self.assertEqual(summary["failed"], [780])
# ---------------------------------------------------------------------------
# Terminal workflow paths
# ---------------------------------------------------------------------------
class TestTerminalPathsUseTheSharedRule(_ExecutorHarness):
"""Merge, close-without-merge, supersession and already-landed."""
def test_merge_path_clears_the_label_for_linked_issues(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
merged_pr = {
"title": "fix: terminal label cleanup",
"body": "Closes #780",
"head": {"ref": "fix/issue-780-terminal-pr-open-label-cleanup"},
}
with patch(
"mcp_server.release_in_progress_label", return_value={780: "released"}
):
result = mcp_server.cleanup_in_progress_for_pr(
merged_pr, "prgs", None, None, None, terminal_reason=tplc.MERGED
)
cleanup = result["pr_open_label_cleanup"]
self.assertTrue(cleanup["clean"])
self.assertEqual(cleanup["terminal_reason"], tplc.MERGED)
self.assertEqual(self.issue_labels[780], ["type:bug"])
def test_close_without_merge_clears_the_label(self):
self.issue_labels[781] = ["type:bug", PR_OPEN, "leases"]
closed_pr = {
"title": "chore: abandoned",
"body": "Closes #781",
"head": {"ref": "chore/issue-781-abandoned"},
}
with patch(
"mcp_server.release_in_progress_label", return_value={781: "released"}
):
result = mcp_server.cleanup_in_progress_for_pr(
closed_pr,
"prgs",
None,
None,
None,
terminal_reason=tplc.CLOSED_WITHOUT_MERGE,
)
cleanup = result["pr_open_label_cleanup"]
self.assertTrue(cleanup["clean"])
self.assertEqual(cleanup["terminal_reason"], tplc.CLOSED_WITHOUT_MERGE)
self.assertEqual(self.issue_labels[781], ["type:bug", "leases"])
def test_pr_without_linked_issue_reports_an_empty_cleanup(self):
pr = {"title": "chore: no link", "body": "", "head": {"ref": "chore/none"}}
result = mcp_server.cleanup_in_progress_for_pr(
pr, "prgs", None, None, None, terminal_reason=tplc.MERGED
)
self.assertEqual(result["cleanup_status"], "no linked issue found")
self.assertTrue(result["pr_open_label_cleanup"]["clean"])
self.assertEqual(result["pr_open_label_cleanup"]["checked"], [])
def test_supersession_reason_is_recorded(self):
self.issue_labels[600] = [PR_OPEN, "type:bug"]
summary = self._clear([600], reason=tplc.SUPERSEDED)
self.assertTrue(summary["clean"])
self.assertEqual(summary["terminal_reason"], tplc.SUPERSEDED)
self.assertEqual(self.issue_labels[600], ["type:bug"])
def test_already_landed_reconciliation_reason_is_recorded(self):
self.issue_labels[601] = [PR_OPEN]
summary = self._clear([601], reason=tplc.ALREADY_LANDED)
self.assertTrue(summary["clean"])
self.assertEqual(summary["terminal_reason"], tplc.ALREADY_LANDED)
self.assertEqual(self.issue_labels[601], [])
def test_issue_780_regression_stale_label_survived_every_terminal_path(self):
"""Regression for the observed leak.
Before the fix each terminal path finished without touching
``status:pr-open``, so the audit found closed issues still carrying it.
Every path now routes through the one shared rule and leaves nothing
behind — while preserving each issue's other labels.
"""
stale = {
626: (["type:bug", PR_OPEN], tplc.CONTROLLER_CLOSURE),
772: (["workflow-hardening", PR_OPEN], tplc.MERGED),
768: ([PR_OPEN], tplc.CLOSED_WITHOUT_MERGE),
758: (["leases", PR_OPEN, "type:bug"], tplc.SUPERSEDED),
755: (["status:done", PR_OPEN], tplc.ALREADY_LANDED),
}
for number, (labels, _reason) in stale.items():
self.issue_labels[number] = list(labels)
for number, (_labels, reason) in stale.items():
summary = self._clear([number], reason=reason)
self.assertTrue(summary["clean"], msg=f"issue #{number}")
audit = tplc.detect_residual_pr_open(
[
{"number": num, "state": "closed", "labels": names}
for num, names in self.issue_labels.items()
]
)
self.assertTrue(audit["clean"])
self.assertEqual(audit["residual_count"], 0)
# Unrelated labels survived every path.
self.assertEqual(self.issue_labels[626], ["type:bug"])
self.assertEqual(self.issue_labels[772], ["workflow-hardening"])
self.assertEqual(self.issue_labels[768], [])
self.assertEqual(self.issue_labels[758], ["leases", "type:bug"])
self.assertEqual(self.issue_labels[755], ["status:done"])
# ---------------------------------------------------------------------------
# Controller closure
# ---------------------------------------------------------------------------
class TestControllerClosure(unittest.TestCase):
def test_close_issue_clears_label_before_closing_and_validates(self):
calls: list[str] = []
def fake_clear(numbers, *_a, **kwargs):
calls.append(f"clear:{kwargs['terminal_reason']}")
return {
"label": PR_OPEN,
"clean": True,
"checked": list(numbers),
"removed": list(numbers),
"already_absent": [],
"failed": [],
"results": [],
"reasons": [],
"safe_next_action": "",
"terminal_reason": kwargs["terminal_reason"],
}
def fake_api(method, url, auth, payload=None):
if method == "PATCH":
calls.append("patch:closed")
return {"state": "closed"}
return {"labels": [{"name": "type:bug"}]}
with patch("mcp_server.clear_pr_open_label", side_effect=fake_clear), \
patch("mcp_server.api_request", side_effect=fake_api), \
patch("mcp_server._profile_permission_block", return_value=None), \
patch("mcp_server.verify_preflight_purity", return_value=None), \
patch("mcp_server.release_in_progress_label", return_value={}), \
patch("mcp_server._resolve", return_value=("h", "o", "r")), \
patch("mcp_server._auth", return_value=FAKE_AUTH), \
patch("gitea_audit.audit_enabled", return_value=False):
result = mcp_server.gitea_close_issue(issue_number=780, remote="prgs")
self.assertTrue(result["success"])
# Cleanup precedes the state change: closing first would bake in the leak.
self.assertEqual(calls[0], f"clear:{tplc.CONTROLLER_CLOSURE}")
self.assertIn("patch:closed", calls)
self.assertTrue(result["terminal_label_validation"]["clean"])
def test_close_issue_fails_closed_when_cleanup_cannot_complete(self):
def fake_clear(numbers, *_a, **kwargs):
return {
"label": PR_OPEN,
"clean": False,
"checked": list(numbers),
"removed": [],
"already_absent": [],
"failed": list(numbers),
"results": [],
"reasons": ["label replacement failed: boom"],
"safe_next_action": "retry",
"terminal_reason": kwargs["terminal_reason"],
}
def fail_on_patch(method, url, auth, payload=None):
if method == "PATCH":
raise AssertionError("issue must not be closed when cleanup failed")
return {}
with patch("mcp_server.clear_pr_open_label", side_effect=fake_clear), \
patch("mcp_server.api_request", side_effect=fail_on_patch), \
patch("mcp_server._profile_permission_block", return_value=None), \
patch("mcp_server.verify_preflight_purity", return_value=None), \
patch("mcp_server._resolve", return_value=("h", "o", "r")), \
patch("mcp_server._auth", return_value=FAKE_AUTH), \
patch("gitea_audit.audit_enabled", return_value=False):
result = mcp_server.gitea_close_issue(issue_number=780, remote="prgs")
self.assertFalse(result["success"])
self.assertTrue(result["blocked"])
self.assertFalse(result["performed"])
self.assertIn("#780", result["message"])
self.assertTrue(result["safe_next_action"])
# ---------------------------------------------------------------------------
# Capability wiring
# ---------------------------------------------------------------------------
class TestCapabilityWiring(unittest.TestCase):
def test_task_is_registered_with_label_authority(self):
import task_capability_map
self.assertEqual(
task_capability_map.required_permission("cleanup_terminal_pr_labels"),
"gitea.issue.comment",
)
self.assertEqual(
task_capability_map.required_role("cleanup_terminal_pr_labels"),
"author",
)
self.assertEqual(
task_capability_map.tool_required_permission(
"gitea_cleanup_terminal_pr_labels"
),
"gitea.issue.comment",
)
def test_recovery_tool_rejects_an_unknown_reason_without_mutating(self):
with patch("mcp_server._profile_permission_block", return_value=None), \
patch("mcp_server.verify_preflight_purity", return_value=None), \
patch("mcp_server.clear_pr_open_label") as mock_clear:
result = mcp_server.gitea_cleanup_terminal_pr_labels(
issue_numbers=[780], terminal_reason="sometime", remote="prgs"
)
self.assertFalse(result["success"])
self.assertFalse(result["clean"])
mock_clear.assert_not_called()
if __name__ == "__main__":
unittest.main()